From b503a5018e0757313acc0988f503d6df936448ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mose=20M=C3=BCller?= Date: Wed, 2 Aug 2023 12:06:20 +0200 Subject: [PATCH] web server: adding function handler --- src/pyDataInterface/server/web_server.py | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/src/pyDataInterface/server/web_server.py b/src/pyDataInterface/server/web_server.py index 1900c35..563092e 100644 --- a/src/pyDataInterface/server/web_server.py +++ b/src/pyDataInterface/server/web_server.py @@ -1,6 +1,6 @@ from enum import Enum from pathlib import Path -from typing import Any, TypedDict +from typing import Any, TypedDict, get_type_hints import socketio from fastapi import FastAPI @@ -50,8 +50,8 @@ class WebAPI: else: sio = socketio.AsyncServer(async_mode="asgi") - @sio.on("frontend_update") # type: ignore - def handle_frontend_update(sid: str, data: FrontendUpdate) -> None: + @sio.event # type: ignore + def frontend_update(sid: str, data: FrontendUpdate) -> Any: logger.debug(f"Received frontend update: {data}") attr = getattr(self.service, data["name"]) if isinstance(attr, DataService): @@ -60,6 +60,24 @@ class WebAPI: setattr( self.service, data["name"], attr.__class__[data["value"]["value"]] ) + elif callable(attr): + args: dict[str, Any] = data["value"]["args"] + type_hints = get_type_hints(attr) + + # Convert arguments to their hinted types + for arg_name, arg_value in args.items(): + if arg_name in type_hints: + arg_type = type_hints[arg_name] + if isinstance(arg_type, type): + # Attempt to convert the argument to its hinted type + try: + args[arg_name] = arg_type(arg_value) + except ValueError: + msg = f"Failed to convert argument '{arg_name}' to type {arg_type.__name__}" + logger.error(msg) + return msg + + return attr(**args) else: setattr(self.service, data["name"], data["value"])