Use base IOC/ophyd in CI tests

src/srx_caproto_iocs/base.py

@@ -1,12 +1,15 @@
 from __future__ import annotations
 
+import textwrap
 import threading
 import uuid
 from enum import Enum
 from pathlib import Path
 
 from caproto import ChannelType
-from caproto.server import PVGroup, pvproperty
+from caproto.server import PVGroup, pvproperty, run, template_arg_parser
+from ophyd import Component as Cpt
+from ophyd import Device, EpicsSignal, EpicsSignalRO
 
 from .utils import now
 
@@ -25,7 +28,7 @@ class StageStates(Enum):
     STAGED = "staged"
 
 
-class GenericSaveIOC(PVGroup):
+class CaprotoSaveIOC(PVGroup):
     """Generic Caproto Save IOC"""
 
     write_dir = pvproperty(
@@ -150,3 +153,33 @@ def saver(request_queue, response_queue):
 
             response = {"success": success, "error_message": error_message}
             response_queue.put(response)
+
+
+class OphydDeviceWithCaprotoIOC(Device):
+    """An ophyd Device which works with the base caproto extension IOC."""
+
+    write_dir = Cpt(EpicsSignal, "write_dir", string=True)
+    file_name = Cpt(EpicsSignal, "file_name", string=True)
+    full_file_path = Cpt(EpicsSignalRO, "full_file_path", string=True)
+    frame_num = Cpt(EpicsSignal, "frame_num")
+    ioc_stage = Cpt(EpicsSignal, "stage", string=True)
+
+
+def check_args(parser_, split_args_):
+    """Helper function to process caproto CLI args."""
+    parsed_args = parser_.parse_args()
+    prefix = parsed_args.prefix
+    if not prefix:
+        parser_.error("The 'prefix' argument must be specified.")
+
+    ioc_opts, run_opts = split_args_(parsed_args)
+    return ioc_opts, run_opts
+
+
+if __name__ == "__main__":
+    parser, split_args = template_arg_parser(
+        default_prefix="", desc=textwrap.dedent(CaprotoSaveIOC.__doc__)
+    )
+    ioc_options, run_options = check_args(parser, split_args)
+    ioc = CaprotoSaveIOC(**ioc_options)
+    run(ioc.pvdb, **run_options)

src/srx_caproto_iocs/sis_scaler/__init__.py

@@ -0,0 +1 @@
+""""SIS scaler Caproto IOC code."""

src/srx_caproto_iocs/zebra/__init__.py

@@ -0,0 +1 @@
+""""Zebra Caproto IOC code."""

src/srx_caproto_iocs/zebra/caproto_ioc.py

@@ -4,24 +4,17 @@
 
 from caproto.server import run, template_arg_parser
 
-from ..base import GenericSaveIOC
+from ..base import CaprotoSaveIOC, check_args
 
 
-class ZebraSaveIOC(GenericSaveIOC):
+class ZebraSaveIOC(CaprotoSaveIOC):
     """Zebra caproto save IOC."""
 
 
 if __name__ == "__main__":
     parser, split_args = template_arg_parser(
         default_prefix="", desc=textwrap.dedent(ZebraSaveIOC.__doc__)
     )
-
-    parsed_args = parser.parse_args()
-    prefix = parsed_args.prefix
-    if not prefix:
-        parser.error("The 'prefix' argument must be specified.")
-
-    ioc_options, run_options = split_args(parsed_args)
-
+    ioc_options, run_options = check_args(parser, split_args)
     ioc = ZebraSaveIOC(**ioc_options)  # TODO: pass libca IOC PVs of interest
     run(ioc.pvdb, **run_options)

src/srx_caproto_iocs/zebra/ophyd.py

@@ -1,14 +1,7 @@
 from __future__ import annotations
 
-from ophyd import Component as Cpt
-from ophyd import Device, EpicsSignal, EpicsSignalRO
+from ..base import OphydDeviceWithCaprotoIOC
 
 
-class ZebraWithCaprotoIOC(Device):
+class ZebraWithCaprotoIOC(OphydDeviceWithCaprotoIOC):
     """An ophyd Device which works with the Zebra caproto extension IOC."""
-
-    write_dir = Cpt(EpicsSignal, "write_dir", string=True)
-    file_name = Cpt(EpicsSignal, "file_name", string=True)
-    full_file_path = Cpt(EpicsSignalRO, "full_file_path", string=True)
-    frame_num = Cpt(EpicsSignal, "frame_num")
-    ioc_stage = Cpt(EpicsSignal, "stage", string=True)

tests/conftest.py

@@ -10,18 +10,23 @@
 import netifaces
 import pytest
 
-from srx_caproto_iocs.zebra.ophyd import ZebraWithCaprotoIOC
+from srx_caproto_iocs.base import OphydDeviceWithCaprotoIOC
+
+CAPROTO_PV_PREFIX = "BASE:{{Dev:Save1}}:"
+OPHYD_PV_PREFIX = CAPROTO_PV_PREFIX.replace("{{", "{").replace("}}", "}")
 
 
 @pytest.fixture()
-def zebra_ophyd_caproto():
-    dev = ZebraWithCaprotoIOC("XF:05IDD-ES:1{Dev:Zebra2}:", name="zebra_ophyd_caproto")
+def base_ophyd_device():
+    dev = OphydDeviceWithCaprotoIOC(
+        OPHYD_PV_PREFIX, name="ophyd_device_with_caproto_ioc"
+    )
     yield dev
     dev.ioc_stage.put("unstaged")
 
 
 @pytest.fixture(scope="session")
-def caproto_ioc(wait=3):
+def base_caproto_ioc(wait=3):
     first_three = ".".join(socket.gethostbyname(socket.gethostname()).split(".")[:3])
     broadcast = f"{first_three}.255"
 
@@ -44,10 +49,7 @@ def caproto_ioc(wait=3):
         except Exception as e:
             print(f"{interface = }: exception:\n  {e}")
 
-    command = (
-        sys.executable
-        + " -m srx_caproto_iocs.zebra.caproto_ioc --prefix=XF:05IDD-ES:1{{Dev:Zebra2}}: --list-pvs"
-    )
+    command = f"{sys.executable} -m srx_caproto_iocs.base --prefix={CAPROTO_PV_PREFIX} --list-pvs"
     print(
         f"\nStarting caproto IOC in via a fixture using the following command:\n\n  {command}\n"
     )

tests/test_zebra_ophyd.py → tests/test_base_ophyd.py

@@ -11,7 +11,7 @@
 
 @pytest.mark.cloud_friendly()
 @pytest.mark.parametrize("date_template", ["%Y/%m/", "%Y/%m/%d", "mydir/%Y/%m/%d"])
-def test_zebra_ophyd_caproto(caproto_ioc, zebra_ophyd_caproto, date_template):
+def test_base_ophyd_templates(base_caproto_ioc, base_ophyd_device, date_template):
     with tempfile.TemporaryDirectory() as tmpdirname:
         date = now(as_object=True)
         write_dir_root = Path(tmpdirname)
@@ -21,7 +21,7 @@ def test_zebra_ophyd_caproto(caproto_ioc, zebra_ophyd_caproto, date_template):
 
         file_template = "scan_{num:06d}_{uid}.hdf5"
 
-        dev = zebra_ophyd_caproto
+        dev = base_ophyd_device
         dev.write_dir.put(dir_template)
         dev.file_name.put(file_template)