Add fastcs launcher

Author: marcelldls

src/fastcs/__init__.py

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

src/fastcs/controller.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from copy import copy
+from typing import Any
 
 from .attributes import Attribute
 
@@ -55,6 +56,9 @@ class Controller(BaseController):
     def __init__(self) -> None:
         super().__init__()
 
+    def startup(self, config: Any) -> None:
+        pass
+
     async def initialise(self) -> None:
         pass
 

src/fastcs/main.py

@@ -1,5 +1,10 @@
-from dataclasses import dataclass
-from typing import Any
+import json
+from pathlib import Path
+from typing import Annotated, Any, get_type_hints
+
+import typer
+from pydantic import create_model
+from ruamel.yaml import YAML
 
 from .backend import Backend
 from .controller import Controller
@@ -8,44 +13,114 @@
 from .transport.tango.options import TangoOptions
 
 
-@dataclass
-class FastCSOptions:
-    transport: EpicsOptions | TangoOptions
-    controller: Any = None
-
-
 class FastCS:
     def __init__(
         self,
-        controller: Controller,
-        options: FastCSOptions,
+        controller: type[Controller],
+        controller_options: Any,
+        transport_options: EpicsOptions | TangoOptions,
     ):
-        self.backend = Backend(controller)
-        self.transport: TransportAdapter
-        match options.transport:
+        self._controller = controller()
+        self._controller.startup(controller_options)
+
+        self._backend = Backend(self._controller)
+        self._transport: TransportAdapter
+        match transport_options:
             case TangoOptions():
                 from .transport.tango.adapter import TangoTransport
 
-                self.transport = TangoTransport(
-                    self.backend.mapping,
-                    options.transport,
+                self._transport = TangoTransport(
+                    self._backend.mapping,
+                    transport_options,
                 )
             case EpicsOptions():
                 from .transport.epics.adapter import EpicsTransport
 
-                self.transport = EpicsTransport(
-                    self.backend.mapping,
-                    self.backend.context,
-                    self.backend.dispatcher,
-                    options.transport,
+                self._transport = EpicsTransport(
+                    self._backend.mapping,
+                    self._backend.context,
+                    self._backend.dispatcher,
+                    transport_options,
                 )
 
     def create_docs(self) -> None:
-        self.transport.create_docs()
+        self._transport.create_docs()
 
     def create_gui(self) -> None:
-        self.transport.create_gui()
+        self._transport.create_gui()
 
     def run(self) -> None:
-        self.backend.run()
-        self.transport.run()
+        self._backend.run()
+        self._transport.run()
+
+
+def launch(controller_class: type[Controller]):
+    try:
+        options_type = get_type_hints(controller_class.startup)["config"]
+    except KeyError as e:
+        raise KeyError(
+            "Missing type hint for config argument, expected: "
+            "`def startup(self, config: <type hint>)`"
+        ) from e
+
+    fastcs_options = create_model(
+        f"{controller_class.__name__}",
+        controller=(options_type, ...),
+        transport=(EpicsOptions | TangoOptions, ...),
+    )
+
+    _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.parse_obj(options_yaml)
+
+        instance = FastCS(
+            controller_class,
+            instance_options.controller,
+            instance_options.transport,
+        )
+
+        if hasattr(instance_options.transport, "gui"):
+            instance.create_gui()
+        if hasattr(instance_options.transport, "docs"):
+            instance.create_docs()
+        instance.run()
+
+    _launch_typer()

src/fastcs/transport/adapter.py

@@ -1,10 +1,4 @@
 from abc import ABC, abstractmethod
-from dataclasses import dataclass
-
-
-@dataclass
-class TransportOptions:
-    pass
 
 
 class TransportAdapter(ABC):

tests/ioc.py

@@ -1,7 +1,7 @@
 from fastcs.attributes import AttrR, AttrRW, AttrW
 from fastcs.controller import Controller, SubController
 from fastcs.datatypes import Int
-from fastcs.main import FastCS, FastCSOptions
+from fastcs.main import FastCS
 from fastcs.transport.epics.options import EpicsIOCOptions, EpicsOptions
 from fastcs.wrappers import command
 
@@ -10,6 +10,9 @@ class ParentController(Controller):
     a: AttrR = AttrR(Int())
     b: AttrRW = AttrRW(Int())
 
+    def startup(self, config):
+        self.register_sub_controller("Child", ChildController())
+
 
 class ChildController(SubController):
     c: AttrW = AttrW(Int())
@@ -20,10 +23,8 @@ async def d(self):
 
 
 def run():
-    controller = ParentController()
-    controller.register_sub_controller("Child", ChildController())
-    options = EpicsOptions(ioc=EpicsIOCOptions(pv_prefix="DEVICE"))
-    fastcs = FastCS(controller, FastCSOptions(options))
+    epics_options = EpicsOptions(ioc=EpicsIOCOptions(pv_prefix="DEVICE"))
+    fastcs = FastCS(ParentController, None, epics_options)
     fastcs.run()