refactor: convert PviDevice and PviTree operations into flattened free functions

Author: shihab-dls

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

@@ -2,7 +2,7 @@
 
 from p4p.server import Server, StaticProvider
 
-from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW
+from fastcs.attributes import AttrR, AttrRW, AttrW
 from fastcs.controller_api import ControllerAPI
 from fastcs.transport.epics.util import controller_pv_prefix
 from fastcs.util import snake_to_pascal
@@ -12,32 +12,20 @@
     make_shared_read_pv,
     make_shared_write_pv,
 )
-from .pvi_tree import AccessModeType, PviTree
-
-
-def _attribute_to_access(attribute: Attribute) -> AccessModeType:
-    match attribute:
-        case AttrRW():
-            return "rw"
-        case AttrR():
-            return "r"
-        case AttrW():
-            return "w"
-        case _:
-            raise ValueError(f"Unknown attribute type {type(attribute)}")
+from .pvi import add_pvi_info
 
 
 async def parse_attributes(
     root_pv_prefix: str, root_controller_api: ControllerAPI
-) -> list[StaticProvider]:
+) -> StaticProvider:
     """Parses `Attribute` s into p4p signals in handlers."""
-    pvi_tree = PviTree(root_pv_prefix)
     provider = StaticProvider(root_pv_prefix)
 
     for controller_api in root_controller_api.walk_api():
         pv_prefix = controller_pv_prefix(root_pv_prefix, controller_api)
-
-        pvi_tree.add_sub_device(pv_prefix, controller_api.description, controller_api)
+        provider = add_pvi_info(
+            provider=provider, pv_prefix=pv_prefix, controller_api=controller_api
+        )
 
         for attr_name, attribute in controller_api.attributes.items():
             full_pv_name = f"{pv_prefix}:{snake_to_pascal(attr_name)}"
@@ -47,23 +35,19 @@ async def parse_attributes(
                     attribute_pv_rbv = make_shared_read_pv(attribute)
                     provider.add(f"{full_pv_name}", attribute_pv)
                     provider.add(f"{full_pv_name}_RBV", attribute_pv_rbv)
-                    pvi_tree.add_signal(f"{full_pv_name}", "rw")
                 case AttrR():
                     attribute_pv = make_shared_read_pv(attribute)
                     provider.add(f"{full_pv_name}", attribute_pv)
-                    pvi_tree.add_signal(f"{full_pv_name}", "r")
                 case AttrW():
                     attribute_pv = make_shared_write_pv(attribute)
                     provider.add(f"{full_pv_name}", attribute_pv)
-                    pvi_tree.add_signal(f"{full_pv_name}", "w")
 
         for attr_name, method in controller_api.command_methods.items():
             full_pv_name = f"{pv_prefix}:{snake_to_pascal(attr_name)}"
             command_pv = make_command_pv(method.fn)
             provider.add(f"{full_pv_name}", command_pv)
-            pvi_tree.add_signal(f"{full_pv_name}", "x")
 
-    return [provider, pvi_tree.make_provider()]
+    return provider
 
 
 class P4PIOC:
@@ -74,8 +58,8 @@ def __init__(self, pv_prefix: str, controller_api: ControllerAPI):
         self.controller_api = controller_api
 
     async def run(self):
-        providers = await parse_attributes(self.pv_prefix, self.controller_api)
+        provider = await parse_attributes(self.pv_prefix, self.controller_api)
 
         endless_event = asyncio.Event()
-        with Server(providers):
+        with Server([provider]):
             await endless_event.wait()

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

@@ -0,0 +1,116 @@
+from collections import defaultdict
+from typing import Literal
+
+from p4p import Type, Value
+from p4p.nt.common import alarm, timeStamp
+from p4p.server import StaticProvider
+from p4p.server.asyncio import SharedPV
+
+from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW
+from fastcs.controller_api import ControllerAPI
+from fastcs.util import snake_to_pascal
+
+from .types import p4p_alarm_states, p4p_timestamp_now
+
+AccessModeType = Literal["r", "w", "rw", "d", "x"]
+
+
+def _attribute_to_access(attribute: Attribute) -> AccessModeType:
+    match attribute:
+        case AttrRW():
+            return "rw"
+        case AttrR():
+            return "r"
+        case AttrW():
+            return "w"
+        case _:
+            raise ValueError(f"Unknown attribute type {type(attribute)}")
+
+
+def add_pvi_info(
+    provider: StaticProvider,
+    pv_prefix: str,
+    controller_api: ControllerAPI,
+    description: str | None = None,
+) -> StaticProvider:
+    provider.add(
+        f"{pv_prefix}:PVI",
+        SharedPV(initial=_make_p4p_value(pv_prefix, controller_api, description)),
+    )
+    return provider
+
+
+def _make_p4p_value(
+    pv_prefix: str, controller_api: ControllerAPI, description: str | None
+) -> Value:
+    display = (
+        {"display": {"description": description}} if description is not None else {}
+    )  # Defined here so the value can be (none)
+
+    raw_value = _make_p4p_raw_value(pv_prefix, controller_api)
+    p4p_type = _make_type_for_raw_value(raw_value)
+
+    try:
+        return Value(
+            p4p_type,
+            {
+                **p4p_alarm_states(),
+                **p4p_timestamp_now(),
+                **display,
+                "value": raw_value,
+            },
+        )
+    except KeyError as e:
+        raise ValueError(f"Failed to create p4p Value from {raw_value}") from e
+
+
+def _make_p4p_raw_value(pv_prefix: str, controller_api: ControllerAPI) -> dict:
+    p4p_raw_value = defaultdict(dict)
+    # Sub-controller api returned if current item is a Controller
+    for pv_leaf, sub_controller_api in controller_api.sub_apis.items():
+        # Add Controller entry
+        # Sub-device of a ControllerVector
+        controller_pvi_name = f"{snake_to_pascal(pv_leaf)}"
+        pv = f"{pv_prefix}:{controller_pvi_name}:PVI"
+        if sub_controller_api.path[-1].isdigit():
+            p4p_raw_value[f"__{int(pv_leaf)}"]["d"] = pv
+        else:
+            p4p_raw_value[controller_pvi_name]["d"] = pv
+    for pv_leaf, attribute in controller_api.attributes.items():
+        # Add attribute entry
+        attr_pvi_name = f"{snake_to_pascal(pv_leaf)}"
+        pv = f"{pv_prefix}:{attr_pvi_name}"
+        p4p_raw_value[attr_pvi_name][_attribute_to_access(attribute)] = pv
+
+    return p4p_raw_value
+
+
+def _make_type_for_raw_value(raw_value: dict) -> Type:
+    p4p_raw_type = []
+    for pvi_group_name, access_to_field in raw_value.items():
+        pvi_group_structure = []
+        for access, field in access_to_field.items():
+            if isinstance(field, str):
+                pvi_group_structure.append((access, "s"))
+            elif isinstance(field, dict):
+                pvi_group_structure.append(
+                    (
+                        access,
+                        (
+                            "S",
+                            None,
+                            [(v, "s") for v, _ in field.items()],
+                        ),
+                    )
+                )
+
+        p4p_raw_type.append((pvi_group_name, ("S", "structure", pvi_group_structure)))
+
+    return Type(
+        [
+            ("alarm", alarm),
+            ("timeStamp", timeStamp),
+            ("display", ("S", None, [("description", "s")])),
+            ("value", ("S", "structure", p4p_raw_type)),
+        ]
+    )