ums.management.main

View Source

  1# Agenten Plattform
  2#
  3# (c) 2024 Magnus Bender
  4# 	Institute of Humanities-Centered Artificial Intelligence (CHAI)
  5# 	Universitaet Hamburg
  6# 	https://www.chai.uni-hamburg.de/~bender
  7#  
  8# source code released under the terms of GNU Public License Version 3
  9# https://www.gnu.org/licenses/gpl-3.0.txt
 10
 11import os
 12
 13from datetime import datetime
 14
 15from fastapi import FastAPI, Request, BackgroundTasks, HTTPException
 16from fastapi.responses import HTMLResponse
 17from fastapi.templating import Jinja2Templates
 18
 19from jinja2.runtime import Undefined as JinjaUndefined
 20
 21from ums.management.interface import Interface
 22from ums.management.db import DB, MessageDbRow
 23from ums.management.process import MessageProcessor
 24
 25from ums.utils import AgentMessage, AgentResponse, TEMPLATE_PATH
 26
 27class WebMain():
 28
 29	_TIME_FORMAT = "%H:%M:%S %d.%m.%Y"
 30
 31	def __init__(self):
 32		self._init_app()
 33		self._init_templates()
 34
 35		self.db = DB()
 36		self.msg_process = MessageProcessor(self.db)
 37
 38		self._add_routes()
 39		self._add_routers()
 40
 41
 42	def _init_app(self):
 43		self.app = FastAPI(
 44			title="Agenten Plattform",
 45			description="Agenten Plattform – Management",
 46			openapi_url="/api/schema.json",
 47			docs_url='/api',
 48			redoc_url=None
 49		)
 50
 51	def _init_templates(self):
 52		self.template = Jinja2Templates(
 53			directory=TEMPLATE_PATH,
 54			auto_reload=True
 55		)
 56		
 57		def timestamp2date(t:int|JinjaUndefined) -> str:
 58			return "" if isinstance(t, JinjaUndefined) \
 59				else datetime.fromtimestamp(t).strftime(self._TIME_FORMAT)
 60		self.template.env.globals["timestamp2date"] = timestamp2date
 61
 62		def date2timestamp(d:str|JinjaUndefined) -> int|str:
 63			return "" if isinstance(d, JinjaUndefined) \
 64				else int(datetime.strptime(d, self._TIME_FORMAT).timestamp())
 65		self.template.env.globals["date2timestamp"] = date2timestamp
 66
 67
 68	def _add_routers(self):
 69		interface_router = Interface(self.template, self.db)
 70		self.app.include_router(interface_router.router)
 71
 72	def _add_routes(self):
 73
 74		@self.app.get("/index", response_class=HTMLResponse, summary="Link list")
 75		def index(request: Request):
 76			return self.template.TemplateResponse(
 77				'index.html',
 78				{"request" : request}
 79			)
 80
 81		@self.app.post("/message", summary="Send a message to the management")
 82		def message(request: Request, message:AgentMessage, background_tasks: BackgroundTasks) -> AgentResponse:
 83
 84			receiver = request.headers['host']
 85			if ':' in receiver:
 86				receiver = receiver[:receiver.rindex(':')]
 87			
 88			sender = request.headers['x-forwarded-for']
 89
 90			return self.msg_process.new_message(sender, receiver, message, background_tasks)
 91		
 92		@self.app.get("/status", summary="Get status of a message")
 93		def status(count:int) -> MessageDbRow:
 94			msg = self.db.by_count(count)
 95			if msg is None:
 96				raise HTTPException(status_code=404, detail="Message not found")
 97			
 98			return msg
 99
100if __name__ == "ums.management.main" and os.environ.get('SERVE', 'false') == 'true':	
101	main = WebMain()
102	app = main.app

class WebMain: View Source

28class WebMain():
29
30	_TIME_FORMAT = "%H:%M:%S %d.%m.%Y"
31
32	def __init__(self):
33		self._init_app()
34		self._init_templates()
35
36		self.db = DB()
37		self.msg_process = MessageProcessor(self.db)
38
39		self._add_routes()
40		self._add_routers()
41
42
43	def _init_app(self):
44		self.app = FastAPI(
45			title="Agenten Plattform",
46			description="Agenten Plattform – Management",
47			openapi_url="/api/schema.json",
48			docs_url='/api',
49			redoc_url=None
50		)
51
52	def _init_templates(self):
53		self.template = Jinja2Templates(
54			directory=TEMPLATE_PATH,
55			auto_reload=True
56		)
57		
58		def timestamp2date(t:int|JinjaUndefined) -> str:
59			return "" if isinstance(t, JinjaUndefined) \
60				else datetime.fromtimestamp(t).strftime(self._TIME_FORMAT)
61		self.template.env.globals["timestamp2date"] = timestamp2date
62
63		def date2timestamp(d:str|JinjaUndefined) -> int|str:
64			return "" if isinstance(d, JinjaUndefined) \
65				else int(datetime.strptime(d, self._TIME_FORMAT).timestamp())
66		self.template.env.globals["date2timestamp"] = date2timestamp
67
68
69	def _add_routers(self):
70		interface_router = Interface(self.template, self.db)
71		self.app.include_router(interface_router.router)
72
73	def _add_routes(self):
74
75		@self.app.get("/index", response_class=HTMLResponse, summary="Link list")
76		def index(request: Request):
77			return self.template.TemplateResponse(
78				'index.html',
79				{"request" : request}
80			)
81
82		@self.app.post("/message", summary="Send a message to the management")
83		def message(request: Request, message:AgentMessage, background_tasks: BackgroundTasks) -> AgentResponse:
84
85			receiver = request.headers['host']
86			if ':' in receiver:
87				receiver = receiver[:receiver.rindex(':')]
88			
89			sender = request.headers['x-forwarded-for']
90
91			return self.msg_process.new_message(sender, receiver, message, background_tasks)
92		
93		@self.app.get("/status", summary="Get status of a message")
94		def status(count:int) -> MessageDbRow:
95			msg = self.db.by_count(count)
96			if msg is None:
97				raise HTTPException(status_code=404, detail="Message not found")
98			
99			return msg

msg_process