--- /dev/null
+import json
+import asyncio
+import websockets
+
+import typs
+
+
+class Agent:
+ """Session interface for Crimata Agent
+
+ Receive core functionalities for communicating with Crimata
+ Agent in an OOP way.
+
+ """
+ def __init__(self, connection):
+ self.connection = connection
+
+ # Submit a request for info, return response.
+ # Must disclose what service it's for.
+ async def fetch(self, entities, service):
+
+ # Assert list.
+ if type(entities) == str:
+ entities = [entities]
+
+ print(f"Fetching {[e for e in entities]} for {service}")
+
+ # Build intent.
+ params = {
+ "service": service,
+ "entities": entities
+ }; intent = typs.Intent("fetch", params)
+
+ # Send intent to agent, wait for response.
+ await self.send_agent_intent(intent)
+ response = await self.recv_agent_intent()
+
+ # Parse response and return entities.
+ entities = response.params.get("found")
+ print(f"Fetch response: {entities}")
+ return entities
+
+ # Text the client something. Also, let agent know
+ # when your service is complete via the complete
+ # attribute.
+ # Must disclose what service it's for.
+ async def notify(self, text, end=False):
+
+ print(f"Notifying: {text} for {self.service}")
+
+ params = {
+ "service": self.service,
+ "text": text,
+ "end": end
+ }
+
+ # Build the intent
+ intent = typs.Intent("notify", params)
+
+ # Send intent to agent.
+ await self.send_agent_intent(intent)
+
+ # Get user info.
+ async def sync(self):
+
+ intent = typs.Intent("sync", params={})
+
+ # Send intent to agent.
+ await self.send_agent_intent(intent)
+
+ # Recv and return profile.
+ intent = await self.recv_agent_intent()
+ profile = intent.params.get("profile")
+ return profile
+
+ async def recv_agent_intent(self):
+ package = await self.connection.recv()
+ intent = self.__decode(package)
+ return intent
+
+ async def send_agent_intent(self, intent):
+ package = self.__encode(intent)
+ await self.connection.send(package)
+
+ def __encode(self, intent: typs.Intent):
+ package = json.dumps(intent.__dict__)
+ return package
+
+ def __decode(self, package) -> typs.Intent:
+ jpackage = json.loads(package)
+ name = jpackage.get("name")
+ params = jpackage.get("params")
+ intent = typs.Intent(name, params)
+ return intent
+
+
+
+
--- /dev/null
+# mail.py
+
+import typs
+import mailroom
+
+
+class Mail:
+ """Session interface for mailroom.
+
+ Two main IO methods. Will translate intents to mail and vice versa.
+ """
+ def __init__(self):
+ pass
+
+ # Receive intent.
+ async def mailbox_recv(self):
+ mail = await mailroom.get_mail(self.crimata_id)
+ intent = self.__decode(mail)
+ return intent
+
+ # Send intent.
+ def mailbox_send(self, intent):
+ mail = self.__encode(intent)
+ mailroom.put_mail(self.crimata_id, mail)
+
+ def __encode(self, intent: typs.Intent) -> typs.Mail:
+ p = intent.params
+ owner = self.crimata_id
+ target = p.get("target")
+ text = p.get("text")
+ mail = typs.Mail(owner, target, text)
+ return mail
+
+ @staticmethod
+ def __decode(mail: typs.Mail) -> typs.Intent:
+ name = "notify"
+ params = {
+ "text": mail.text #dev only
+ }
+ intent = typs.Intent(name, params)
+ return intent
+
\ No newline at end of file
--- /dev/null
+# mailroom.py
+
+import queue
+import asyncio
+
+import typs
+
+# Subcribers (Crimata IDs)
+subs = []
+
+# Mailroom (all mailboxes)
+que = queue.Queue()
+ref = {}
+
+# Mailroom functions
+# ------------------
+
+# Create mailbox and add to mailroom.
+def create_mailbox(crimata_id):
+ mailbox = typs.MailBox(name=crimata_id)
+ que.put(mailbox)
+ ref.update({crimata_id: mailbox})
+ subs.append(crimata_id)
+
+# Get mailbox by Crimata ID.
+def get_mailbox(crimata_id):
+ if not crimata_id in subs:
+ create_mailbox(crimata_id)
+ mailbox = ref.get(crimata_id)
+ return mailbox
+
+# Inbox -> receiving mail.
+async def get_mail(crimata_id):
+ mailbox = get_mailbox(crimata_id)
+ mail = await mailbox.inbox.get()
+ return mail
+
+# Outbox -> sending mail.
+def put_mail(crimata_id, mail: typs.Mail):
+ mailbox = get_mailbox(crimata_id)
+ mailbox.outbox.put(mail)
+
\ No newline at end of file
--- /dev/null
+# server.py
+
+import typs
+import queue
+import asyncio
+import websockets
+
+import server
+import message
+
+
+# Run the ws server and the messenger concurrently.
+async def main():
+ print(f"Crimata Messenger")
+ await asyncio.gather(server.lift(), message.run())
+
+if __name__ == '__main__':
+ asyncio.run(main())
--- /dev/null
+# messenge.py
+
+import asyncio
+
+import mailroom
+
+
+# Independent loop that updates mailboxes.
+async def messenger():
+
+ print("Running Messenger")
+
+ while True:
+
+ # Yield if no mailboxes.
+ if mailroom.que.empty():
+ await asyncio.sleep(1)
+ continue
+
+ # Grab a mailbox.
+ mailbox = mailroom.que.get()
+ print(f"Handling mailbox: {mailbox.name}.")
+
+ # Handle every message in outbox.
+ while not mailbox.outbox.empty():
+ message = mailbox.outbox.get()
+
+ # Get mailbox of target and put mail there.
+ target_mailbox = mailroom.get_mailbox(message.target)
+ await target_mailbox.inbox.put(message)
+
+ # Put the mailbox back
+ mailroom.que.put(mailbox)
+ await asyncio.sleep(1)
+
+async def run():
+ await messenger()
\ No newline at end of file
--- /dev/null
+# schemes.py
+
+
+class Schemes:
+
+ def __init__(self):
+ self.service = None
+
+ async def handle_intent(self, intent):
+ print(f"Handling intent {intent.name}.")
+
+ # Set the service
+ self.service = intent.name
+
+ if intent.name == "init":
+ await self.init(intent)
+ if intent.name == "message":
+ self.message(intent)
+
+ async def init(self, intent):
+ self.crimata_id = intent.params.get("crimata_id")
+ await self.notify("Messaging is live.")
+
+ def message(self, intent):
+ text = intent.params.get("text")
+ target = intent.params.get("target") #crimata_id
+ print(f"Messaging {target}: '{text}'")
+
+ # Uses mail.Mail interface.
+ self.mailbox_send(intent)
\ No newline at end of file
--- /dev/null
+# server.py
+
+import asyncio
+import websockets
+
+import session
+
+HOST = "localhost"
+PORT = 8764
+
+
+# Makes instance of a Messenger session for every client that connects.
+async def launch_session(connection, path):
+ s = session.Session(connection)
+
+ await asyncio.gather(
+
+ s.do_agent_intents(),
+
+ s.deliver_mail()
+
+ )
+
+# Run run for every new connection.
+async def lift():
+ print(f"Listening for connections on {HOST}:{PORT}")
+ await websockets.serve(launch_session, HOST, PORT)
\ No newline at end of file
--- /dev/null
+# session.py
+
+import time
+import asyncio
+
+import mail
+import agent
+import schemes
+
+
+class Session(schemes.Schemes, agent.Agent, mail.Mail):
+ """Launch instance for every client connection.
+
+ Will recv messages from client and push them to mailbox.
+ Also, will pull messages from mailbox and send to client.
+
+ """
+ def __init__(self, connection):
+ agent.Agent.__init__(self, connection)
+
+ self.crimata_id = False
+
+ print("Launched new session")
+
+ # Recv intents run desired endpoint.
+ async def do_agent_intents(self):
+ while True:
+
+ # Recv intent from agent.
+ intent = await self.recv_agent_intent()
+ print(f"Intent: {intent.name}")
+
+ # Call corresponding endpoint.
+ await self.handle_intent(intent)
+
+ await asyncio.sleep(0.1)
+
+ # Send mail back to agent.
+ async def deliver_mail(self):
+ while True:
+
+ # Sleep until Crimata ID.
+ if not self.crimata_id:
+ await asyncio.sleep(1)
+ continue
+
+ intent = await self.mailbox_recv()
+ await self.send_agent_intent(intent)
+
+ await asyncio.sleep(0.1)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ # async def hello(self):
+
+ # intent = await self.recv_agent_intent()
+
+ # response = await self.fetch("crimata_id", service="message")
+ # self.profile = response.get("profile")
+
+ # target = intent.params.get("target")
+ # await self.notify(f"Messenger APP: Received Message Intent",
+ # service="message", complete=True)
+
+ # print("Shutting down in 10s")
+
+ # await asyncio.sleep(10)
--- /dev/null
+# typs.py
+
+import queue
+import asyncio
+
+
+class Mail:
+
+ def __init__(self, owner, target, text):
+ self.owner = owner
+ self.target = target
+ self.text = text
+
+
+class MailBox:
+
+ def __init__(self, name):
+
+ self.name = name # just for logging purposes.
+ self.inbox = asyncio.Queue()
+ self.outbox = queue.Queue()
+
+
+class Intent:
+
+ def __init__(self, name, params: dict):
+ self.name = name
+ self.params = params
\ No newline at end of file
projects / abc-messenger / commitdiff