Merge pull request #41 from DiamondLightSource/controller-nesting

Allow arbitrary nesting of SubControllers

pyproject.toml

@@ -14,7 +14,7 @@ dependencies = [
  "aioserial",
  "numpy",
  "pydantic",
- "pvi~=0.8.1",
+ "pvi~=0.9.0",
  "softioc",
 ]
 dynamic = ["version"]

src/fastcs/backends/epics/gui.py

@@ -28,7 +28,7 @@
 from fastcs.cs_methods import Command
 from fastcs.datatypes import Bool, DataType, Float, Int, String
 from fastcs.exceptions import FastCSException
-from fastcs.mapping import Mapping, SingleMapping
+from fastcs.mapping import Mapping, SingleMapping, _get_single_mapping
 from fastcs.util import snake_to_pascal
 
 
@@ -49,12 +49,10 @@ def __init__(self, mapping: Mapping, pv_prefix: str) -> None:
  self._mapping = mapping
  self._pv_prefix = pv_prefix
 
- def _get_pv(self, attr_path: str, name: str):
- if attr_path:
- attr_path = ":" + attr_path
- attr_path += ":"
-
- return f"{self._pv_prefix}{attr_path.upper()}{name.title().replace('_', '')}"
+ def _get_pv(self, attr_path: list[str], name: str):
+ attr_prefix = ":".join(attr_path)
+ pv_prefix = ":".join((self._pv_prefix, attr_prefix))
+ return f"{pv_prefix}:{name.title().replace('_', '')}"
 
  @staticmethod
  def _get_read_widget(datatype: DataType) -> ReadWidget:
@@ -80,7 +78,9 @@ def _get_write_widget(datatype: DataType) -> WriteWidget:
  case _:
  raise FastCSException(f"Unsupported type {type(datatype)}: {datatype}")
 
- def _get_attribute_component(self, attr_path: str, name: str, attribute: Attribute):
+ def _get_attribute_component(
+ self, attr_path: list[str], name: str, attribute: Attribute
+ ):
  pv = self._get_pv(attr_path, name)
  name = name.title().replace("_", "")
 
@@ -102,7 +102,7 @@ def _get_attribute_component(self, attr_path: str, name: str, attribute: Attribu
  write_widget = self._get_write_widget(attribute.datatype)
  return SignalW(name=name, write_pv=pv, write_widget=write_widget)
 
- def _get_command_component(self, attr_path: str, name: str):
+ def _get_command_component(self, attr_path: list[str], name: str):
  pv = self._get_pv(attr_path, name)
  name = name.title().replace("_", "")
 
@@ -122,30 +122,30 @@ def create_gui(self, options: EpicsGUIOptions | None = None) -> None:
 
  assert options.output_path.suffix == options.file_format.value
 
- formatter = DLSFormatter()
-
  controller_mapping = self._mapping.get_controller_mappings()[0]
- sub_controller_mappings = self._mapping.get_controller_mappings()[1:]
-
  components = self.extract_mapping_components(controller_mapping)
-
- for sub_controller_mapping in sub_controller_mappings:
- components.append(
- Group(
- name=snake_to_pascal(sub_controller_mapping.controller.path),
- layout=SubScreen(),
- children=self.extract_mapping_components(sub_controller_mapping),
- )
- )
-
  device = Device(label=options.title, children=components)
 
+ formatter = DLSFormatter()
  formatter.format(device, options.output_path)
 
  def extract_mapping_components(self, mapping: SingleMapping) -> list[Component]:
  components: Tree[Component] = []
  attr_path = mapping.controller.path
 
+ for sub_controller in mapping.controller.get_sub_controllers():
+ components.append(
+ Group(
+ # TODO: Build assumption that SubController has at least one path
+ # element into typing
+ name=snake_to_pascal(sub_controller.path[-1]),
+ layout=SubScreen(),
+ children=self.extract_mapping_components(
+ _get_single_mapping(sub_controller)
+ ),
+ )
+ )
+
  groups: dict[str, list[Component]] = {}
  for attr_name, attribute in mapping.attributes.items():
  signal = self._get_attribute_component(
@@ -159,6 +159,9 @@ def extract_mapping_components(self, mapping: SingleMapping) -> list[Component]:
  if group not in groups:
  groups[group] = []
 
+ # Remove duplication of group name and signal name
+ signal.name = signal.name.removeprefix(group)
+
  groups[group].append(signal)
  case _:
  components.append(signal)

src/fastcs/backends/epics/ioc.py

@@ -88,7 +88,7 @@ def _create_and_link_attribute_pvs(mapping: Mapping) -> None:
  path = single_mapping.controller.path
  for attr_name, attribute in single_mapping.attributes.items():
  attr_name = attr_name.title().replace("_", "")
- pv_name = path.upper() + ":" + attr_name if path else attr_name
+ pv_name = f"{':'.join(path).upper()}:{attr_name}" if path else attr_name
 
  match attribute:
  case AttrRW():
@@ -103,9 +103,9 @@ def _create_and_link_attribute_pvs(mapping: Mapping) -> None:
 def _create_and_link_command_pvs(mapping: Mapping) -> None:
  for single_mapping in mapping.get_controller_mappings():
  path = single_mapping.controller.path
- for name, method in single_mapping.command_methods.items():
- name = name.title().replace("_", "")
- pv_name = path.upper() + ":" + name if path else name
+ for attr_name, method in single_mapping.command_methods.items():
+ attr_name = attr_name.title().replace("_", "")
+ pv_name = f"{':'.join(path).upper()}:{attr_name}" if path else attr_name
 
  _create_and_link_command_pv(
  pv_name, MethodType(method.fn, single_mapping.controller)

src/fastcs/controller.py

@@ -6,12 +6,15 @@
 
 
 class BaseController:
- def __init__(self, path="") -> None:
- self._path: str = path
+ def __init__(self, path: list[str] | None = None) -> None:
+ self._path: list[str] = path or []
+ self.__sub_controllers: list[SubController] = []
+
  self._bind_attrs()
 
  @property
- def path(self):
+ def path(self) -> list[str]:
+ """Path prefix of attributes, recursively including parent ``Controller``s."""
  return self._path
 
  def _bind_attrs(self) -> None:
@@ -21,6 +24,12 @@ def _bind_attrs(self) -> None:
  new_attribute = copy(attr)
  setattr(self, attr_name, new_attribute)
 
+ def register_sub_controller(self, controller: SubController):
+ self.__sub_controllers.append(controller)
+
+ def get_sub_controllers(self) -> list[SubController]:
+ return self.__sub_controllers
+
 
 class Controller(BaseController):
  """Top-level controller for a device.
@@ -33,17 +42,10 @@ class Controller(BaseController):
 
  def __init__(self) -> None:
  super().__init__()
- self.__sub_controllers: list[SubController] = []
 
  async def connect(self) -> None:
  pass
 
- def register_sub_controller(self, controller: SubController):
- self.__sub_controllers.append(controller)
-
- def get_sub_controllers(self) -> list[SubController]:
- return self.__sub_controllers
-
 
 class SubController(BaseController):
  """A subordinate to a ``Controller`` for managing a subset of a device.
@@ -52,5 +54,5 @@ class SubController(BaseController):
  it as part of a larger device.
  """
 
- def __init__(self, path: str) -> None:
+ def __init__(self, path: list[str]) -> None:
  super().__init__(path)

src/fastcs/mapping.py

@@ -1,3 +1,4 @@
+from collections.abc import Iterator
 from dataclasses import dataclass
 
 from .attributes import Attribute
@@ -18,36 +19,7 @@ class SingleMapping:
 class Mapping:
  def __init__(self, controller: Controller) -> None:
  self.controller = controller
-
- self._controller_mappings: list[SingleMapping] = []
- self._controller_mappings.append(self._get_single_mapping(controller))
-
- for sub_controller in controller.get_sub_controllers():
- self._controller_mappings.append(self._get_single_mapping(sub_controller))
-
- @staticmethod
- def _get_single_mapping(controller: BaseController) -> SingleMapping:
- scan_methods = {}
- put_methods = {}
- command_methods = {}
- attributes = {}
- for attr_name in dir(controller):
- attr = getattr(controller, attr_name)
- match attr:
- case WrappedMethod(fastcs_method=fastcs_method):
- match fastcs_method:
- case Put():
- put_methods[attr_name] = fastcs_method
- case Scan():
- scan_methods[attr_name] = fastcs_method
- case Command():
- command_methods[attr_name] = fastcs_method
- case Attribute():
- attributes[attr_name] = attr
-
- return SingleMapping(
- controller, scan_methods, put_methods, command_methods, attributes
- )
+ self._controller_mappings = list(_walk_mappings(controller))
 
  def __str__(self) -> str:
  result = "Controller mappings:\n"
@@ -57,3 +29,33 @@ def __str__(self) -> str:
 
  def get_controller_mappings(self) -> list[SingleMapping]:
  return self._controller_mappings
+
+
+def _walk_mappings(controller: BaseController) -> Iterator[SingleMapping]:
+ yield _get_single_mapping(controller)
+ for sub_controller in controller.get_sub_controllers():
+ yield from _walk_mappings(sub_controller)
+
+
+def _get_single_mapping(controller: BaseController) -> SingleMapping:
+ scan_methods = {}
+ put_methods = {}
+ command_methods = {}
+ attributes = {}
+ for attr_name in dir(controller):
+ attr = getattr(controller, attr_name)
+ match attr:
+ case WrappedMethod(fastcs_method=fastcs_method):
+ match fastcs_method:
+ case Put():
+ put_methods[attr_name] = fastcs_method
+ case Scan():
+ scan_methods[attr_name] = fastcs_method
+ case Command():
+ command_methods[attr_name] = fastcs_method
+ case Attribute():
+ attributes[attr_name] = attr
+
+ return SingleMapping(
+ controller, scan_methods, put_methods, command_methods, attributes
+ )

src/fastcs/util.py

@@ -1,3 +1,5 @@
 def snake_to_pascal(input: str) -> str:
- """Convert a snake_case or UPPER_SNAKE_CASE string to PascalCase."""
- return input.lower().replace("_", " ").title().replace(" ", "")
+ """Convert a snake_case string to PascalCase."""
+ return "".join(
+ part.title() if part.islower() else part for part in input.split("_")
+ )

tests/test_controller.py

@@ -0,0 +1,17 @@
+from fastcs.controller import Controller, SubController
+from fastcs.mapping import _get_single_mapping, _walk_mappings
+
+
+def test_controller_nesting():
+ controller = Controller()
+ sub_controller = SubController(["a"])
+ sub_sub_controller = SubController(["a", "b"])
+
+ controller.register_sub_controller(sub_controller)
+ sub_controller.register_sub_controller(sub_sub_controller)
+
+ assert list(_walk_mappings(controller)) == [
+ _get_single_mapping(controller),
+ _get_single_mapping(sub_controller),
+ _get_single_mapping(sub_sub_controller),
+ ]