Add proposal API, launcher

Author: marcelldls

src/fastcs/__init__.py

@@ -7,5 +7,7 @@
 """
 
 from ._version import __version__
+from .main import FastCS as FastCS
+from .main import launch as launch
 
 __all__ = ["__version__"]

src/fastcs/backend.py

@@ -4,7 +4,7 @@
 from concurrent.futures import Future
 from types import MethodType
 
-from softioc.asyncio_dispatcher import AsyncioDispatcher
+from fastcs.util import AsyncioDispatcher
 
 from .attributes import AttrR, AttrW, Sender, Updater
 from .controller import Controller
@@ -14,10 +14,12 @@
 
 class Backend:
     def __init__(
-        self, controller: Controller, loop: asyncio.AbstractEventLoop | None = None
+        self,
+        controller: Controller,
+        loop: asyncio.AbstractEventLoop | None = None,
     ):
-        self._dispatcher = AsyncioDispatcher(loop)
-        self._loop = self._dispatcher.loop
+        self.dispatcher = AsyncioDispatcher(loop)
+        self._loop = self.dispatcher.loop
         self._controller = controller
 
         self._initial_coros = [controller.connect]
@@ -27,17 +29,17 @@ def __init__(
             self._controller.initialise(), self._loop
         ).result()
 
-        self._mapping = Mapping(self._controller)
+        self.mapping = Mapping(self._controller)
         self._link_process_tasks()
 
-        self._context = {
-            "dispatcher": self._dispatcher,
+        self.context = {
+            "dispatcher": self.dispatcher,
             "controller": self._controller,
-            "mapping": self._mapping,
+            "mapping": self.mapping,
         }
 
     def _link_process_tasks(self):
-        for single_mapping in self._mapping.get_controller_mappings():
+        for single_mapping in self.mapping.get_controller_mappings():
             _link_single_controller_put_tasks(single_mapping)
             _link_attribute_sender_class(single_mapping)
 
@@ -47,7 +49,6 @@ def __del__(self):
     def run(self):
         self._run_initial_futures()
         self.start_scan_futures()
-        self._run()
 
     def _run_initial_futures(self):
         for coro in self._initial_coros:
@@ -57,7 +58,7 @@ def _run_initial_futures(self):
     def start_scan_futures(self):
         self._scan_futures = {
             asyncio.run_coroutine_threadsafe(coro(), self._loop)
-            for coro in _get_scan_coros(self._mapping)
+            for coro in _get_scan_coros(self.mapping)
         }
 
     def stop_scan_futures(self):
@@ -68,9 +69,6 @@ def stop_scan_futures(self):
                 except asyncio.CancelledError:
                     pass
 
-    def _run(self):
-        raise NotImplementedError("Specific Backend must implement _run")
-
 
 def _link_single_controller_put_tasks(single_mapping: SingleMapping) -> None:
     for name, method in single_mapping.put_methods.items():

src/fastcs/backends/__init__.py

@@ -1,3 +1,7 @@
 from .ip_connection import IPConnection
+from .ip_connection import IPConnectionSettings as IPConnectionSettings
+from .ip_connection import StreamConnection as StreamConnection
+from .serial_connection import SerialConnection as SerialConnection
+from .serial_connection import SerialConnectionSettings as SerialConnectionSettings
 
 __all__ = ["IPConnection"]

src/fastcs/backends/asyncio_backend.py

@@ -1,2 +1,6 @@
 class FastCSException(Exception):
     pass
+
+
+class LaunchError(FastCSException):
+    pass

src/fastcs/backends/epics/backend.py

@@ -0,0 +1,178 @@
+import inspect
+import json
+from pathlib import Path
+from typing import Annotated, TypeAlias, get_type_hints
+
+import typer
+from pydantic import create_model
+from ruamel.yaml import YAML
+
+from .backend import Backend
+from .controller import Controller
+from .exceptions import LaunchError
+from .transport.adapter import TransportAdapter
+from .transport.epics.options import EpicsOptions
+from .transport.rest.options import RestOptions
+from .transport.tango.options import TangoOptions
+
+# Define a type alias for transport options
+TransportOptions: TypeAlias = EpicsOptions | TangoOptions | RestOptions
+
+
+class FastCS:
+    def __init__(
+        self,
+        controller: Controller,
+        transport_options: TransportOptions,
+    ):
+        self._backend = Backend(controller)
+        self._transport: TransportAdapter
+        match transport_options:
+            case EpicsOptions():
+                from .transport.epics.adapter import EpicsTransport
+
+                self._transport = EpicsTransport(
+                    self._backend.mapping,
+                    self._backend.context,
+                    self._backend.dispatcher,
+                    transport_options,
+                )
+            case TangoOptions():
+                from .transport.tango.adapter import TangoTransport
+
+                self._transport = TangoTransport(
+                    self._backend.mapping,
+                    transport_options,
+                )
+            case RestOptions():
+                from .transport.rest.adapter import RestTransport
+
+                self._transport = RestTransport(
+                    self._backend.mapping,
+                    transport_options,
+                )
+
+    def create_docs(self) -> None:
+        self._transport.create_docs()
+
+    def create_gui(self) -> None:
+        self._transport.create_gui()
+
+    def run(self) -> None:
+        self._backend.run()
+        self._transport.run()
+
+
+def launch(controller_class: type[Controller]) -> None:
+    """
+    Serves as an entry point for starting FastCS applications.
+
+    By utilizing type hints in a Controller's __init__ method, this
+    function provides a command-line interface to describe and gather the
+    required configuration before instantiating the application.
+
+    Args:
+        controller_class (type[Controller]): The FastCS Controller to instantiate.
+        It must have a type-hinted __init__ method and no more than 2 arguments.
+
+    Raises:
+        LaunchError: If the class's __init__ is not as expected
+
+    Example of the expected Controller implementation:
+        class MyController(Controller):
+            def __init__(self, my_arg: MyControllerOptions) -> None:
+                ...
+
+    Typical usage:
+        if __name__ == "__main__":
+            launch(MyController)
+    """
+    _launch(controller_class)()
+
+
+def _launch(controller_class: type[Controller]) -> typer.Typer:
+    sig = inspect.signature(controller_class.__init__)
+    args = inspect.getfullargspec(controller_class.__init__)[0]
+    if len(args) == 1:
+        fastcs_options = create_model(
+            f"{controller_class.__name__}",
+            transport=(TransportOptions, ...),
+            __config__={"extra": "forbid"},
+        )
+    elif len(args) == 2:
+        hints = get_type_hints(controller_class.__init__)
+        if hints:
+            options_type = list(hints.values())[-1]
+        else:
+            raise LaunchError(
+                f"Expected typehinting in '{controller_class.__name__}"
+                f".__init__' but received {sig}. Add a typehint for `{args[-1]}`."
+            )
+        fastcs_options = create_model(
+            f"{controller_class.__name__}",
+            controller=(options_type, ...),
+            transport=(TransportOptions, ...),
+            __config__={"extra": "forbid"},
+        )
+    else:
+        raise LaunchError(
+            f"Expected no more than 2 arguments for '{controller_class.__name__}"
+            f".__init__' but received {len(args)} as `{sig}`"
+        )
+
+    launch_typer = typer.Typer()
+
+    class LaunchContext:
+        def __init__(self, controller_class, fastcs_options):
+            self.controller_class = controller_class
+            self.fastcs_options = fastcs_options
+
+    @launch_typer.callback()
+    def create_context(ctx: typer.Context):
+        ctx.obj = LaunchContext(
+            controller_class,
+            fastcs_options,
+        )
+
+    @launch_typer.command(help=f"Produce json schema for a {controller_class.__name__}")
+    def schema(ctx: typer.Context):
+        system_schema = ctx.obj.fastcs_options.model_json_schema()
+        print(json.dumps(system_schema, indent=2))
+
+    @launch_typer.command(help=f"Start up a {controller_class.__name__}")
+    def run(
+        ctx: typer.Context,
+        config: Annotated[
+            Path,
+            typer.Argument(
+                help=f"A yaml file matching the {controller_class.__name__} schema"
+            ),
+        ],
+    ):
+        """
+        Start the controller
+        """
+        controller_class = ctx.obj.controller_class
+        fastcs_options = ctx.obj.fastcs_options
+
+        yaml = YAML(typ="safe")
+        options_yaml = yaml.load(config)
+        # To do: Handle a k8s "values.yaml" file
+        instance_options = fastcs_options.model_validate(options_yaml)
+        if hasattr(instance_options, "controller"):
+            controller = controller_class(instance_options.controller)
+        else:
+            controller = controller_class()
+
+        instance = FastCS(
+            controller,
+            instance_options.transport,
+        )
+
+        if "gui" in options_yaml["transport"]:
+            instance.create_gui()
+        if "docs" in options_yaml["transport"]:
+            instance.create_docs()
+        instance.run()
+
+    return launch_typer

src/fastcs/backends/tango/backend.py

@@ -0,0 +1,8 @@
+from .epics.options import EpicsDocsOptions as EpicsDocsOptions
+from .epics.options import EpicsGUIOptions as EpicsGUIOptions
+from .epics.options import EpicsIOCOptions as EpicsIOCOptions
+from .epics.options import EpicsOptions as EpicsOptions
+from .tango.options import TangoDSROptions as TangoDSROptions
+from .tango.options import TangoOptions as TangoOptions
+
+__all__ = ["EpicsOptions", "TangoOptions"]