Begin Types for Messages

Readme.md

@@ -1,2 +1,14 @@
 # Agenten-Plattform
 
+## Management
+
+- `./docker-mgmt/`
+- `./ums/management/`
+- `./web/`
+- `./build-mgmt.sh`
+
+
+## Basic Agent
+
+- `./docker-agent/`
+- `./ums/agent/`

ums/management/main.py

@@ -1,9 +1,12 @@
 
 # TEST ONLY
 
+from ums.messages import Message, Riddle
+
 from typing import Union
 
 from fastapi import FastAPI
+from fastapi.responses import PlainTextResponse
 
 app = FastAPI()
 
@@ -11,6 +14,14 @@ app = FastAPI()
 def read_root():
 	return {"Hello": "World"}
 
+@app.get("/test", response_class=PlainTextResponse)
+def huhu():
+	a = Message(
+		id="hu", riddle="lala"
+	)
+	a.status.steps["extract"] = False
+	return a.to_json()
+
 
 @app.get("/items/{item_id}")
 def read_item(item_id: int, q: Union[str, None] = None):

ums/messages/__init__.py

@@ -0,0 +1,4 @@
+
+from ums.messages.message import Message
+from ums.messages.specials import Riddle, RiddleData, RiddleStatus, Solution
+

ums/messages/message.py

@@ -0,0 +1,102 @@
+import importlib, json
+
+from typing import Any, List
+
+class Message():
+
+	_USM_AGENT_CODE = "CZDygPSF2HJTLKVIqys1";
+
+	_ATTRIBUTES = { # * in name -> optional
+		"id" : str,
+		"sub_ids*" : List[str],
+		"riddle" : "ums.messages.specials.Riddle",
+		"solution*" : "ums.messages.specials.Solution",
+		"data" : "ums.messages.specials.RiddleData",
+		"status" : "ums.messages.specials.RiddleStatus"
+	}
+
+	_DEFAULTS = {
+		"data" : ("ums.messages.specials", "RiddleData"),
+		"status" : ("ums.messages.specials", "RiddleStatus")
+	}
+
+	def __init__(self, *args, **kwargs):
+		if len(args) > 0:
+			self.__dict__['items'] = list(args)
+		elif len(kwargs) > 0:
+			self.__dict__['items'] = kwargs
+		else:
+			self.__dict__['items'] = self._DEFAULTS
+
+		self._check_attr()
+
+	def __getattr__(self, name: str) -> Any:
+		if 'items' in self.__dict__ and isinstance(self.items, dict) and name in self.items:
+			return self.items[name]
+		
+	def __setattr__(self, name: str, value: Any) -> None:
+		if hasattr(self, name):
+			setattr(self, name, value)
+		elif isinstance(self.items, dict):
+			self.items[name] = value
+			self._check_attr()
+
+	def _check_attr(self):
+		if isinstance(self._ATTRIBUTES, list):
+			assert isinstance(self.items, list), \
+				"{} content must be a list and not a dict!".format(self.__class__.__name__)
+
+		elif isinstance(self._ATTRIBUTES, dict):
+			assert isinstance(self.items, dict), \
+				"{} content must be a dict and not a list!".format(self.__class__.__name__)
+
+			for k,v in self._ATTRIBUTES.items():
+				if not k.endswith('*') and not k in self.items:
+					if k in self._DEFAULTS:
+						if isinstance(self._DEFAULTS[k], tuple):
+							def_class = getattr(
+								importlib.import_module(self._DEFAULTS[k][0]),
+								self._DEFAULTS[k][1]
+							)
+							self.items[k] = def_class()
+						else:
+							self.items[k] = self._DEFAULTS[k]
+					else:
+						raise ValueError("Message requires field {}, but not set (and no default)!".format(k))
+				
+				if isinstance(v, str):
+					# a class name
+					pass
+
+				elif isinstance(v, dict):
+					# a sub structure
+					pass
+
+				else: # a type
+					pass
+
+
+	def _to_json(self):
+		return self.items
+
+	def to_json(self) -> str:
+		return json.dumps(
+			self._to_json(),
+			indent=2,
+			cls=JSONEncodeHelper,
+			sort_keys=True
+		)
+
+
+	def from_json(json:str):
+		# TODO
+		
+		pass
+
+class JSONEncodeHelper(json.JSONEncoder):
+	def default(self, o):
+
+		if isinstance(o, Message):
+			return o._to_json()
+
+		return super().default(o)

ums/messages/specials.py

@@ -0,0 +1,99 @@
+import os 
+
+from typing import Generator, Tuple, IO
+
+from ums.messages.message import Message
+
+class Riddle(Message):
+
+	_ATTRIBUTES = {  # * in name -> optional
+		"context" : str,
+		"question" : str,
+		"solution_before*" : str,
+		"review_before*" : str
+	}
+	_DEFAULTS = {}
+
+
+class RiddleStatus(Message):
+
+	_ATTRIBUTES = {  # * in name -> optional
+		"steps" : {
+			"extract" : bool,
+			"solve" : bool,
+			"validate" : bool
+		},
+		"trial" : int,
+		"solved" : bool
+	}
+	_DEFAULTS = {
+		"steps" : {
+			"extract" : True,
+			"solve" : True,
+			"validate" : True
+		},
+		"trial" : 0,
+		"solved" : False
+	}
+
+class RiddleData(Message):
+
+	TYPE_TEXT = "text"
+	TYPE_IMAGE = "image"
+	TYPE_AUDIO = "audio"
+
+	DATA_TYPES = [
+		TYPE_IMAGE,
+		TYPE_TEXT,
+		TYPE_AUDIO
+	]
+
+	FILE_BASEPATH = '/ums-agenten/share'
+
+	_ATTRIBUTES = [
+		{
+			"type" : str,
+			"file_plain" : str,
+			"file_extracted*" : str
+		}
+	]
+	_DEFAULTS = []
+
+	def iterate(self, 
+			yield_plain:bool=True, yield_extracted:bool=True,
+			file_pointer:bool=False
+		) -> Generator[Tuple[str, str|IO|None, str|IO|None], None, None]:
+			for item in self.items:
+				if item["type"] in self.DATA_TYPES:
+					
+					f_e = None
+					if yield_extracted and "file_extracted" in item:
+						f_e_p = os.path.join(self.FILE_BASEPATH, item["file_extracted"])
+						if os.path.isfile(f_e_p):
+							f_e = open(f_e_p, 'r') if file_pointer else f_e_p
+
+					f_p = None
+					if yield_plain:
+						f_p_p = os.path.join(self.FILE_BASEPATH, item["file_plain"])
+						if os.path.isfile(f_p_p):
+							f_e = open(f_p_p, 'r') if file_pointer else f_p_p
+						
+					yield  item["type"], f_p, f_e
+
+					if hasattr(f_p, 'close'):
+						f_p.close()
+					if hasattr(f_e, 'close'):
+						f_e.close()
+
+	def __iter__(self):
+		yield from self.iterate()
+
+class Solution(Message):
+
+	_ATTRIBUTES = {
+		"solution" : str,
+		"explanation" : str,
+		"used_data*" : "ums.messages.specials.RiddleData",
+		"review*" : str
+	}
+	_DEFAULTS = {}