Improve consistency of PVA and CA PV generation

Author: GDYendell

docs/conf.py

@@ -94,6 +94,7 @@
     ("py:class", "fastcs.logging._graylog.GraylogStaticFields"),
     ("py:class", "fastcs.logging._graylog.GraylogEnvFields"),
     ("py:obj", "fastcs.launch.build_controller_api"),
+    ("py:obj", "fastcs.transport.epics.util.controller_pv_prefix"),
     ("docutils", "fastcs.demo.controllers.TemperatureControllerSettings"),
     # TypeVar without docstrings still give warnings
     ("py:class", "fastcs.datatypes.T_Numerical"),

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

@@ -20,6 +20,7 @@
     record_metadata_from_datatype,
 )
 from fastcs.transport.epics.options import EpicsIOCOptions
+from fastcs.transport.epics.util import controller_pv_prefix
 from fastcs.util import snake_to_pascal
 
 EPICS_MAX_NAME_LENGTH = 60
@@ -111,27 +112,26 @@ def _add_sub_controller_pvi_info(pv_prefix: str, parent: ControllerAPI):
         parent: Controller to add PVI refs for
 
     """
-    parent_pvi = ":".join([pv_prefix] + parent.path + ["PVI"])
+    parent_pvi = f"{controller_pv_prefix(pv_prefix, parent)}:PVI"
 
     for child in parent.sub_apis.values():
-        child_pvi = ":".join([pv_prefix] + _controller_pv_prefix(child.path) + ["PVI"])
+        child_pvi = f"{controller_pv_prefix(pv_prefix, child)}:PVI"
         child_name = child.path[-1].lower()
 
         _add_pvi_info(child_pvi, parent_pvi, child_name)
-
         _add_sub_controller_pvi_info(pv_prefix, child)
 
 
 def _create_and_link_attribute_pvs(
     root_pv_prefix: str, root_controller_api: ControllerAPI
 ) -> None:
     for controller_api in root_controller_api.walk_api():
-        path = controller_api.path
+        pv_prefix = controller_pv_prefix(root_pv_prefix, controller_api)
+
         for attr_name, attribute in controller_api.attributes.items():
             pv_name = snake_to_pascal(attr_name)
-            pv_prefix = ":".join([root_pv_prefix] + _controller_pv_prefix(path))
-            full_pv_name_length = len(f"{pv_prefix}:{pv_name}")
 
+            full_pv_name_length = len(f"{pv_prefix}:{pv_name}")
             if full_pv_name_length > EPICS_MAX_NAME_LENGTH:
                 attribute.enabled = False
                 print(
@@ -224,9 +224,7 @@ async def async_write_display(value: T):
 
         record.set(cast_to_epics_type(attribute.datatype, value), process=False)
 
-    record = _make_record(
-        f"{pv_prefix}:{pv_name}", attribute, on_update=on_update, out_record=True
-    )
+    record = _make_record(pv, attribute, on_update=on_update, out_record=True)
 
     _add_attr_pvi_info(record, pv_prefix, attr_name, "w")
 
@@ -237,10 +235,11 @@ def _create_and_link_command_pvs(
     root_pv_prefix: str, root_controller_api: ControllerAPI
 ) -> None:
     for controller_api in root_controller_api.walk_api():
-        path = controller_api.path
+        pv_prefix = controller_pv_prefix(root_pv_prefix, controller_api)
+
         for attr_name, method in controller_api.command_methods.items():
             pv_name = snake_to_pascal(attr_name)
-            pv_prefix = ":".join([root_pv_prefix] + _controller_pv_prefix(path))
+
             if len(f"{pv_prefix}:{pv_name}") > EPICS_MAX_NAME_LENGTH:
                 print(
                     f"Not creating PV for {attr_name} as full name would exceed"
@@ -302,7 +301,3 @@ def _add_attr_pvi_info(
             }
         },
     )
-
-
-def _controller_pv_prefix(controller_path: list[str]) -> list[str]:
-    return [snake_to_pascal(node) for node in controller_path]

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

@@ -4,6 +4,7 @@
 
 from fastcs.attributes import Attribute, 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
 
 from ._pv_handlers import (
@@ -26,12 +27,6 @@ def _attribute_to_access(attribute: Attribute) -> AccessModeType:
             raise ValueError(f"Unknown attribute type {type(attribute)}")
 
 
-def get_pv_name(pv_prefix: str, *attribute_names: str) -> str:
-    """Converts from an attribute name to a pv name."""
-    pv_formatted = ":".join([snake_to_pascal(attr) for attr in attribute_names])
-    return f"{pv_prefix}:{pv_formatted}" if pv_formatted else pv_prefix
-
-
 async def parse_attributes(
     root_pv_prefix: str, root_controller_api: ControllerAPI
 ) -> list[StaticProvider]:
@@ -40,33 +35,33 @@ async def parse_attributes(
     provider = StaticProvider(root_pv_prefix)
 
     for controller_api in root_controller_api.walk_api():
-        pv_prefix = get_pv_name(root_pv_prefix, *controller_api.path)
+        pv_prefix = controller_pv_prefix(root_pv_prefix, controller_api)
 
         pvi_tree.add_sub_device(pv_prefix, controller_api.description)
 
         for attr_name, attribute in controller_api.attributes.items():
-            pv_name = get_pv_name(pv_prefix, attr_name)
+            full_pv_name = f"{pv_prefix}:{snake_to_pascal(attr_name)}"
             match attribute:
                 case AttrRW():
                     attribute_pv = make_shared_write_pv(attribute)
                     attribute_pv_rbv = make_shared_read_pv(attribute)
-                    provider.add(pv_name, attribute_pv)
-                    provider.add(f"{pv_name}_RBV", attribute_pv_rbv)
-                    pvi_tree.add_signal(pv_name, "rw")
+                    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(pv_name, attribute_pv)
-                    pvi_tree.add_signal(pv_name, "r")
+                    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(pv_name, attribute_pv)
-                    pvi_tree.add_signal(pv_name, "w")
+                    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():
-            pv_name = get_pv_name(pv_prefix, attr_name)
+            full_pv_name = f"{pv_prefix}:{snake_to_pascal(attr_name)}"
             command_pv = make_command_pv(method.fn)
-            provider.add(pv_name, command_pv)
-            pvi_tree.add_signal(pv_name, "x")
+            provider.add(f"{full_pv_name}", command_pv)
+            pvi_tree.add_signal(f"{full_pv_name}", "x")
 
     return [provider, pvi_tree.make_provider()]
 

src/fastcs/transport/epics/util.py

@@ -0,0 +1,6 @@
+from fastcs.controller_api import ControllerAPI
+from fastcs.util import snake_to_pascal
+
+
+def controller_pv_prefix(prefix: str, controller_api: ControllerAPI) -> str:
+    return ":".join([prefix] + [snake_to_pascal(node) for node in controller_api.path])