# Written by retoor@molodetz.nl
# This module defines a Component class that facilitates WebSocket communication and management of various tasks in an asynchronous environment. It allows dynamic creation of child components and interaction through callbacks and events.
# Imports used: aiohttp (external library for client-server communication)
# MIT License
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
import asyncio
import time
import uuid
from aiohttp import web
class Component:
@classmethod
def define(cls):
return cls
def __init__(self, app, id_, description=None, ws: web.WebSocketResponse = None):
self.id = id_
self.ws = ws
self.app = app
self.description = description
self.children = []
self._callbacks = {}
self.value = None
self._running = False
if not hasattr(self, "Children"):
return
for name in dir(self.Children):
if name.startswith("__"):
continue
obj = getattr(self.Children, name)
instance = obj(app=self.app, id_=name, ws=ws)
self.add_child(instance)
instance.app = self
instance.ws = self.ws
setattr(self, name, instance)
@classmethod
def from_json(cls, json):
obj = cls(None, None)
obj.__dict__ = json
return obj
@classmethod
def to_json(cls):
obj = cls.__dict__.copy()
return obj
@classmethod
def clone(cls):
return cls.from_json(cls.to_json())
@property
def running(self):
if not hasattr(self.app, "_running"):
return self._running
return self.app._running
@property
def tasks(self):
tasks_ = [
getattr(self, name)
for name in dir(self)
if name.startswith("task_") and hasattr(self, name)
]
for child in self.children:
tasks_ += child.tasks
return tasks_
async def communicate(self, event_id=None):
async for msg in self.ws:
if msg.type == web.WSMsgType.TEXT:
data = msg.json()
if not event_id:
pass
else:
if data.get("event_id") == event_id:
return data
@property
def callbacks(self):
return hasattr(self.app, "callbacks") and self.app.callbacks or self._callbacks
async def trigger(self, id_, event, data):
if self.id == id_:
method_name = "on_" + event
if hasattr(self, method_name):
method = getattr(self, method_name)
await method(data)
for child in self.children:
await child.trigger(id_, event, data)
async def register_callback(self, event_id, callback):
self.callbacks[event_id] = callback
async def call(self, method, args=None, id_=None, callback=True):
while not self.running:
await asyncio.sleep(0.1)
if not args:
args = []
event_id = str(uuid.uuid4())
loop = asyncio.get_running_loop()
future = loop.create_future()
self.callbacks[event_id] = lambda data: future.set_result(data)
await self.ws.send_json(
{
"event_id": event_id,
"event": "call",
"id": id_ and id_ or self.id,
"method": method,
"args": args,
"callback": callback,
}
)
if callback:
response = await self.communicate(event_id=event_id)
return response["result"]
return True
async def get_attr(self, key, default=None):
result = await self.call("getAttr", [self.id, key], True)
return result or default
async def set_attr(self, key, value):
result = await self.call("setAttr", [self.id, key, value], callback=False)
return result
async def get(self, id_):
if self.id == id_:
return self
for child in self.children:
child = await child.get(id_)
if child:
return child
async def set_data(self, key, value):
result = await self.call("setData", [self.id, key, value], callback=False)
return result
async def get_data(self, key, default=None):
result = await self.call("getData", [self.id, key], default, True)
return result or default
async def set_style(self, key, value):
result = await self.call("setStyle", [self.id, key, value], callback=False)
return result
async def toggle(self):
value = await self.get_style("display", "block")
if value == "none":
value = ""
else:
value = "none"
await self.set_style("display", value)
async def get_style(self, key, default=None):
result = await self.call("getStyle", [self.id, key], default)
return result or default
async def on_keyup(self, event):
value = await self.get_attr("value")
if self.value != value:
if hasattr(self, "on_change"):
value = await self.on_change(value)
self.value = value
return self.value
async def get_tasks(self):
tasks = self.tasks
for child in self.children:
tasks += child.tasks
return tasks
async def set_running(self):
self._running = True
async def get_message(self):
async for msg in self.ws:
return msg
async def service(self):
tasks = self.tasks
tasks.append(self.set_running)
async def events():
try:
async for msg in self.ws:
if msg.type == web.WSMsgType.TEXT:
data = msg.json()
response = {"event_id": data["event_id"], "success": True}
response["time_start"] = time.time()
if self.callbacks.get(data["event_id"]):
self.callbacks[data["event_id"]](data["result"])
elif data.get("data") and not data["data"].get("id"):
response["handled"] = False
elif data.get("data"):
response["handled"] = True
response["data"] = await self.trigger(
data["data"]["id"], data["event"], data["data"]
)
response["cancel"] = True
response["time_end"] = time.time()
response["time_duration"] = (
response["time_end"] - response["time_start"]
)
await self.ws.send_json(response)
elif msg.type == web.WSMsgType.ERROR:
print(f"WebSocket error: {self.ws.exception()}")
except Exception as ex:
print(ex)
pass
tasks.append(events)
await asyncio.gather(*[task() for task in tasks])
def add_child(self, child):
child.app = self.app
child.ws = self.ws
self.children.append(child)
Reference in New Issue View Git Blame Copy Permalink