Update transport model creation for new class structure (#223)

Author: GDYendell

src/fastcs/demo/controller.yaml

@@ -5,7 +5,7 @@ controller:
     port: 25565
   num_ramp_controllers: 4
 transport:
-  - gql:
+  - graphql:
       host: localhost
       port: 8083
       log_level: info

src/fastcs/launch.py

@@ -2,7 +2,7 @@
 import inspect
 import json
 import signal
-from collections.abc import Coroutine
+from collections.abc import Coroutine, Sequence
 from functools import partial
 from pathlib import Path
 from typing import Annotated, Any, Optional, TypeAlias, get_type_hints
@@ -13,27 +13,31 @@
 from ruamel.yaml import YAML
 
 from fastcs import __version__
+from fastcs.transport.epics.ca.transport import EpicsCATransport
+from fastcs.transport.epics.pva.transport import EpicsPVATransport
+from fastcs.transport.graphql.transport import GraphQLTransport
+from fastcs.transport.rest.transport import RestTransport
+from fastcs.transport.tango.transport import TangoTransport
 
 from .backend import Backend
 from .controller import Controller
 from .exceptions import LaunchError
 from .transport import Transport
-from .transport.epics.ca.options import EpicsCAOptions
-from .transport.epics.pva.options import EpicsPVAOptions
-from .transport.graphql.options import GraphQLOptions
-from .transport.rest.options import RestOptions
-from .transport.tango.options import TangoOptions
 
 # Define a type alias for transport options
-TransportOptions: TypeAlias = list[
-    EpicsPVAOptions | EpicsCAOptions | TangoOptions | RestOptions | GraphQLOptions
+TransportList: TypeAlias = list[
+    EpicsPVATransport
+    | EpicsCATransport
+    | TangoTransport
+    | RestTransport
+    | GraphQLTransport
 ]
 
 
 class FastCS:
     """For launching a controller with given transport(s)."""
 
-    def __init__(self, controller: Controller, transports: list[Transport]):
+    def __init__(self, controller: Controller, transports: Sequence[Transport]):
         self._loop = asyncio.get_event_loop()
         self._controller = controller
         self._backend = Backend(controller, self._loop)
@@ -45,13 +49,11 @@ def __init__(self, controller: Controller, transports: list[Transport]):
 
     def create_docs(self) -> None:
         for transport in self._transports:
-            if hasattr(transport.options, "docs"):
-                transport.create_docs()
+            transport.create_docs()
 
     def create_gui(self) -> None:
         for transport in self._transports:
-            if hasattr(transport.options, "gui"):
-                transport.create_gui()
+            transport.create_gui()
 
     def run(self):
         serve = asyncio.ensure_future(self.serve())
@@ -231,7 +233,7 @@ def _extract_options_model(controller_class: type[Controller]) -> type[BaseModel
     if len(args) == 1:
         fastcs_options = create_model(
             f"{controller_class.__name__}",
-            transport=(TransportOptions, ...),
+            transport=(TransportList, ...),
             __config__={"extra": "forbid"},
         )
     elif len(args) == 2:
@@ -248,7 +250,7 @@ def _extract_options_model(controller_class: type[Controller]) -> type[BaseModel
         fastcs_options = create_model(
             f"{controller_class.__name__}",
             controller=(options_type, ...),
-            transport=(TransportOptions, ...),
+            transport=(TransportList, ...),
             __config__={"extra": "forbid"},
         )
     else:

src/fastcs/transport/__init__.py

@@ -1,12 +1,10 @@
-from .epics.ca.options import EpicsCAOptions as EpicsCAOptions
+from .epics.ca.transport import EpicsCATransport as EpicsCATransport
 from .epics.options import EpicsDocsOptions as EpicsDocsOptions
 from .epics.options import EpicsGUIOptions as EpicsGUIOptions
 from .epics.options import EpicsIOCOptions as EpicsIOCOptions
-from .epics.pva.options import EpicsPVAOptions as EpicsPVAOptions
-from .graphql.options import GraphQLOptions as GraphQLOptions
-from .graphql.options import GraphQLServerOptions as GraphQLServerOptions
-from .rest.options import RestOptions as RestOptions
-from .rest.options import RestServerOptions as RestServerOptions
+from .epics.pva.transport import EpicsPVATransport as EpicsPVATransport
+from .graphql.transport import GraphQLTransport as GraphQLTransport
+from .rest.transport import RestTransport as RestTransport
 from .tango.options import TangoDSROptions as TangoDSROptions
-from .tango.options import TangoOptions as TangoOptions
+from .tango.transport import TangoTransport as TangoTransport
 from .transport import Transport as Transport

src/fastcs/transport/epics/ca/adapter.py …c/fastcs/transport/epics/ca/transport.pysrc/fastcs/transport/epics/ca/adapter.py renamed to src/fastcs/transport/epics/ca/transport.py src/fastcs/transport/epics/ca/adapter.py renamed to src/fastcs/transport/epics/ca/transport.py

@@ -1,21 +1,28 @@
 import asyncio
+from dataclasses import dataclass, field
 from typing import Any
 
 from softioc import softioc
 
 from fastcs.controller_api import ControllerAPI
-from fastcs.transport import Transport
 from fastcs.transport.epics.ca.ioc import EpicsCAIOC
-from fastcs.transport.epics.ca.options import EpicsCAOptions
 from fastcs.transport.epics.docs import EpicsDocs
 from fastcs.transport.epics.gui import EpicsGUI
+from fastcs.transport.epics.options import (
+    EpicsDocsOptions,
+    EpicsGUIOptions,
+    EpicsIOCOptions,
+)
+from fastcs.transport.transport import Transport
 
 
+@dataclass
 class EpicsCATransport(Transport):
     """Channel access transport."""
 
-    def __init__(self, options: EpicsCAOptions | None = None):
-        self._options = options or EpicsCAOptions()
+    docs: EpicsDocsOptions = field(default_factory=EpicsDocsOptions)
+    gui: EpicsGUIOptions = field(default_factory=EpicsGUIOptions)
+    ca_ioc: EpicsIOCOptions = field(default_factory=EpicsIOCOptions)
 
     def initialise(
         self,
@@ -24,22 +31,14 @@ def initialise(
     ) -> None:
         self._controller_api = controller_api
         self._loop = loop
-        self._pv_prefix = self._options.ca_ioc.pv_prefix
-        self._ioc = EpicsCAIOC(
-            self._options.ca_ioc.pv_prefix,
-            controller_api,
-            self._options.ca_ioc,
-        )
-
-    @property
-    def options(self) -> EpicsCAOptions:
-        return self._options
+        self._pv_prefix = self.ca_ioc.pv_prefix
+        self._ioc = EpicsCAIOC(self.ca_ioc.pv_prefix, controller_api, self.ca_ioc)
 
     def create_docs(self) -> None:
-        EpicsDocs(self._controller_api).create_docs(self._options.docs)
+        EpicsDocs(self._controller_api).create_docs(self.docs)
 
     def create_gui(self) -> None:
-        EpicsGUI(self._controller_api, self._pv_prefix).create_gui(self._options.gui)
+        EpicsGUI(self._controller_api, self._pv_prefix).create_gui(self.gui)
 
     async def serve(self) -> None:
         print(f"Running FastCS IOC: {self._pv_prefix}")

src/fastcs/transport/epics/pva/options.py

@@ -1,39 +1,42 @@
 import asyncio
+from dataclasses import dataclass, field
 
 from fastcs.controller_api import ControllerAPI
-from fastcs.transport import Transport
 from fastcs.transport.epics.docs import EpicsDocs
 from fastcs.transport.epics.gui import PvaEpicsGUI
-from fastcs.transport.epics.pva.options import EpicsPVAOptions
+from fastcs.transport.epics.options import (
+    EpicsDocsOptions,
+    EpicsGUIOptions,
+    EpicsIOCOptions,
+)
+from fastcs.transport.transport import Transport
 
 from .ioc import P4PIOC
 
 
+@dataclass
 class EpicsPVATransport(Transport):
     """PV access transport."""
 
-    def __init__(self, options: EpicsPVAOptions | None = None):
-        self._options = options or EpicsPVAOptions()
+    docs: EpicsDocsOptions = field(default_factory=EpicsDocsOptions)
+    gui: EpicsGUIOptions = field(default_factory=EpicsGUIOptions)
+    pva_ioc: EpicsIOCOptions = field(default_factory=EpicsIOCOptions)
 
     def initialise(
         self,
         controller_api: ControllerAPI,
         loop: asyncio.AbstractEventLoop,
     ) -> None:
         self._controller_api = controller_api
-        self._pv_prefix = self.options.pva_ioc.pv_prefix
-        self._ioc = P4PIOC(self.options.pva_ioc.pv_prefix, controller_api)
-
-    @property
-    def options(self) -> EpicsPVAOptions:
-        return self._options
+        self._pv_prefix = self.pva_ioc.pv_prefix
+        self._ioc = P4PIOC(self.pva_ioc.pv_prefix, controller_api)
 
     async def serve(self) -> None:
         print(f"Running FastCS IOC: {self._pv_prefix}")
         await self._ioc.run()
 
     def create_docs(self) -> None:
-        EpicsDocs(self._controller_api).create_docs(self.options.docs)
+        EpicsDocs(self._controller_api).create_docs(self.docs)
 
     def create_gui(self) -> None:
-        PvaEpicsGUI(self._controller_api, self._pv_prefix).create_gui(self.options.gui)
+        PvaEpicsGUI(self._controller_api, self._pv_prefix).create_gui(self.gui)

src/fastcs/transport/epics/pva/adapter.py …/fastcs/transport/epics/pva/transport.pysrc/fastcs/transport/epics/pva/adapter.py renamed to src/fastcs/transport/epics/pva/transport.py src/fastcs/transport/epics/pva/adapter.py renamed to src/fastcs/transport/epics/pva/transport.py

@@ -1,15 +1,8 @@
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 
 
 @dataclass
 class GraphQLServerOptions:
     host: str = "localhost"
     port: int = 8080
     log_level: str = "info"
-
-
-@dataclass
-class GraphQLOptions:
-    """Options for the GraphQL transport."""
-
-    gql: GraphQLServerOptions = field(default_factory=GraphQLServerOptions)

src/fastcs/transport/graphql/adapter.py

@@ -0,0 +1,25 @@
+import asyncio
+from dataclasses import dataclass, field
+
+from fastcs.controller_api import ControllerAPI
+from fastcs.transport.transport import Transport
+
+from .graphql import GraphQLServer
+from .options import GraphQLServerOptions
+
+
+@dataclass
+class GraphQLTransport(Transport):
+    """GraphQL transport."""
+
+    graphql: GraphQLServerOptions = field(default_factory=GraphQLServerOptions)
+
+    def initialise(
+        self,
+        controller_api: ControllerAPI,
+        loop: asyncio.AbstractEventLoop,
+    ):
+        self._server = GraphQLServer(controller_api)
+
+    async def serve(self) -> None:
+        await self._server.serve(self.graphql)