Use controller __init__ hinting for launcher

Author: marcelldls

src/fastcs/controller.py

@@ -56,9 +56,6 @@ 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,4 +1,5 @@
 import json
+from inspect import signature
 from pathlib import Path
 from typing import Annotated, Any, get_type_hints
 
@@ -16,14 +17,10 @@
 class FastCS:
     def __init__(
         self,
-        controller: type[Controller],
-        controller_options: Any,
+        controller: Controller,
         transport_options: EpicsOptions | TangoOptions,
     ):
-        self._controller = controller()
-        self._controller.startup(controller_options)
-
-        self._backend = Backend(self._controller)
+        self._backend = Backend(controller)
         self._transport: TransportAdapter
         match transport_options:
             case TangoOptions():
@@ -54,20 +51,64 @@ def run(self) -> None:
         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
+class LaunchError(Exception):
+    pass
 
-    fastcs_options = create_model(
-        f"{controller_class.__name__}",
-        controller=(options_type, ...),
-        transport=(EpicsOptions | TangoOptions, ...),
-    )
+
+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)
+    """
+    args_len = controller_class.__init__.__code__.co_argcount
+    sig = signature(controller_class.__init__)
+    if args_len == 1:
+        fastcs_options = create_model(
+            f"{controller_class.__name__}",
+            transport=(EpicsOptions | TangoOptions, ...),
+            __config__={"extra": "forbid"},
+        )
+    elif args_len == 2:
+        hints = get_type_hints(controller_class.__init__)
+        if "self" in hints:
+            del hints["self"]
+        if hints:
+            options_type = list(hints.values())[-1]
+        else:
+            raise LaunchError(
+                f"Expected typehinting in {controller_class.__name__}"
+                f".__init__ but received {sig}"
+            )
+        fastcs_options = create_model(
+            f"{controller_class.__name__}",
+            controller=(options_type, ...),
+            transport=(EpicsOptions | TangoOptions, ...),
+            __config__={"extra": "forbid"},
+        )
+    else:
+        raise LaunchError(
+            f"Expected up to 2 arguments for {controller_class.__name__}.__init__ "
+            f"but received {args_len} {sig}"
+        )
 
     _launch_typer = typer.Typer()
 
@@ -109,11 +150,14 @@ def run(
         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_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_class,
-            instance_options.controller,
+            controller,
             instance_options.transport,
         )
 

tests/ioc.py

@@ -10,7 +10,8 @@ class ParentController(Controller):
     a: AttrR = AttrR(Int())
     b: AttrRW = AttrRW(Int())
 
-    def startup(self, config):
+    def __init__(self):
+        super().__init__()
         self.register_sub_controller("Child", ChildController())
 
 
@@ -24,7 +25,8 @@ async def d(self):
 
 def run():
     epics_options = EpicsOptions(ioc=EpicsIOCOptions(pv_prefix="DEVICE"))
-    fastcs = FastCS(ParentController, None, epics_options)
+    controller = ParentController()
+    fastcs = FastCS(controller, epics_options)
     fastcs.run()