Merge pull request #4 from NSLS-II-SRX/more-caproto-ioc-flavors

String/Char caproto IOC with demo tests

.github/workflows/ci.yml

@@ -53,19 +53,18 @@ jobs:
     env:
       TZ: America/New_York
 
+    defaults:
+      run:
+        shell: bash -l {0}
+
     steps:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
 
-      - uses: actions/setup-python@v5
-        with:
-          python-version: ${{ matrix.python-version }}
-          allow-prereleases: true
-
       - name: Set env vars
         run: |
-
+          set -x
           export REPOSITORY_NAME=${GITHUB_REPOSITORY#*/}  # just the repo, as opposed to org/repo
           echo "REPOSITORY_NAME=${REPOSITORY_NAME}" >> $GITHUB_ENV
 
@@ -75,9 +74,23 @@ jobs:
           export DATETIME_STRING=$(date +%Y%m%d%H%M%S)
           echo "DATETIME_STRING=${DATETIME_STRING}" >> $GITHUB_ENV
 
+      # - uses: actions/setup-python@v5
+      #   with:
+      #     python-version: ${{ matrix.python-version }}
+      #     allow-prereleases: true
+
+      - name: Set up Python ${{ matrix.python-version }} with conda
+        uses: mamba-org/setup-micromamba@v1
+        with:
+          init-shell: bash
+          environment-name: ${{env.REPOSITORY_NAME}}-py${{matrix.python-version}}
+          create-args: >-
+            python=${{ matrix.python-version }} epics-base setuptools<67
+
       - name: Install package
         run: |
           set -vxeuo pipefail
+          which caput
           python -m pip install .[test]
 
       - name: Test package

pyproject.toml

@@ -41,7 +41,6 @@ dependencies = [
 
 [project.optional-dependencies]
 test = [
-  "netifaces",
   "pytest >=6",
   "pytest-cov >=3",
 ]
@@ -92,6 +91,7 @@ markers = [
     "hardware: marks tests as requiring the hardware IOC to be available/running (deselect with '-m \"not hardware\"')",
     "tiled: marks tests as requiring tiled",
     "cloud_friendly: marks tests to be able to execute in the CI in the cloud",
+    "needs_epics_core: marks tests as requiring epics-core executables such as caget, caput, etc."
 ]
 
 [tool.coverage]
@@ -135,7 +135,7 @@ extend-select = [
   "PT",       # flake8-pytest-style
   "PTH",      # flake8-use-pathlib
   "RET",      # flake8-return
-  "RUF",      # Ruff-specific
+  # "RUF",      # Ruff-specific
   "SIM",      # flake8-simplify
   # "T20",      # flake8-print
   "UP",       # pyupgrade

scripts/run-act.sh

@@ -1,3 +1,7 @@
 #!/bin/bash
 
-act -W .github/workflows/ci.yml -j checks --matrix python-version:3.11
+set -vxeuo pipefail
+
+PYTHON_VERSION="${1:-3.11}"
+
+act -W .github/workflows/ci.yml -j checks --matrix python-version:"${PYTHON_VERSION}"

scripts/run-caproto-ioc.sh

@@ -2,6 +2,8 @@
 
 set -vxeuo pipefail
 
+CAPROTO_IOC="${1:-srx_caproto_iocs.base}"
+
 # shellcheck source=/dev/null
 if [ -f "/etc/profile.d/epics.sh" ]; then
     . /etc/profile.d/epics.sh
@@ -10,4 +12,4 @@ fi
 export EPICS_CAS_AUTO_BEACON_ADDR_LIST="no"
 export EPICS_CAS_BEACON_ADDR_LIST="${EPICS_CA_ADDR_LIST:-127.0.0.255}"
 
-python -m srx_caproto_iocs.base --prefix="BASE:{{Dev:Save1}}:" --list-pvs
+python -m "${CAPROTO_IOC}" --prefix="BASE:{{Dev:Save1}}:" --list-pvs

scripts/test-file-saving.sh

@@ -10,10 +10,7 @@ fi
 num="${1:-50}"
 
 data_dir="/tmp/test/$(date +%Y/%m/%d)"
-
-if [ ! -d "${data_dir}" ]; then
-	mkdir -v -p "${data_dir}"
-fi
+mkdir -v -p "${data_dir}"
 
 caput "BASE:{Dev:Save1}:write_dir" "${data_dir}"
 caput "BASE:{Dev:Save1}:file_name" "saveme_{num:06d}_{uid}.h5"

src/srx_caproto_iocs/base.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import re
 import textwrap
 import threading
 import time as ttime
@@ -39,14 +40,14 @@ class CaprotoSaveIOC(PVGroup):
         value="/tmp",
         doc="The directory to write data to. It support datetime formatting, e.g. '/tmp/det/%Y/%m/%d/'",
         string_encoding="utf-8",
-        report_as_string=True,
+        dtype=ChannelType.CHAR,
         max_length=255,
     )
     file_name = pvproperty(
         value="test.h5",
         doc="The file name of the file to write to. It support <str>.format() based formatting, e.g. 'scan_{num:06d}.h5'",
         string_encoding="utf-8",
-        report_as_string=True,
+        dtype=ChannelType.CHAR,
         max_length=255,
     )
     full_file_path = pvproperty(
@@ -116,12 +117,15 @@ async def stage(self, instance, value):
 
         if value == StageStates.STAGED.value:
             # Steps:
-            # 1. Render 'write_dir' with datetime lib and replace any blank spaces with underscores.
-            # 2. Render 'file_name' with .format().
-            # 3. Replace blank spaces with underscores.
+            # 1. Render 'write_dir' with datetime lib
+            # 2. Replace unsupported characters with underscores (sanitize).
+            # 3. Check if sanitized 'write_dir' exists
+            # 4. Render 'file_name' with .format().
+            # 5. Replace unsupported characters with underscores.
 
+            sanitizer = re.compile(pattern=r"[\":<>|\*\?\s]")
             date = now(as_object=True)
-            write_dir = Path(date.strftime(self.write_dir.value).replace(" ", "_"))
+            write_dir = Path(sanitizer.sub("_", date.strftime(self.write_dir.value)))
             if not write_dir.exists():
                 msg = f"Path '{write_dir}' does not exist."
                 print(msg)
@@ -136,8 +140,7 @@ async def stage(self, instance, value):
             full_file_path = write_dir / file_name.format(
                 num=self.frame_num.value, uid=uid, suid=uid[:8]
             )
-            full_file_path = str(full_file_path)
-            full_file_path.replace(" ", "_")
+            full_file_path = sanitizer.sub("_", str(full_file_path))
 
             print(f"{now()}: {full_file_path = }")
 

src/srx_caproto_iocs/example/__init__.py

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

src/srx_caproto_iocs/example/caproto_ioc.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+import textwrap
+
+from caproto import ChannelType
+from caproto.server import PVGroup, pvproperty, run, template_arg_parser
+
+from ..base import check_args
+
+
+class CaprotoStringIOC(PVGroup):
+    """Test channel types for strings."""
+
+    common_kwargs = {"max_length": 255, "string_encoding": "utf-8"}
+
+    bare_string = pvproperty(
+        value="bare_string", doc="A test for a bare string", **common_kwargs
+    )
+    implicit_string_type = pvproperty(
+        value="implicit_string_type",
+        doc="A test for an implicit string type",
+        report_as_string=True,
+        **common_kwargs,
+    )
+    string_type = pvproperty(
+        value="string_type",
+        doc="A test for a string type",
+        dtype=str,
+        report_as_string=True,
+        **common_kwargs,
+    )
+    string_type_enum = pvproperty(
+        value="string_type_enum",
+        doc="A test for a string type",
+        dtype=ChannelType.STRING,
+        **common_kwargs,
+    )
+    char_type_as_string = pvproperty(
+        value="char_type_as_string",
+        doc="A test for a char type reported as string",
+        report_as_string=True,
+        dtype=ChannelType.CHAR,
+        **common_kwargs,
+    )
+    char_type = pvproperty(
+        value="char_type",
+        doc="A test for a char type not reported as string",
+        dtype=ChannelType.CHAR,
+        **common_kwargs,
+    )
+
+
+if __name__ == "__main__":
+    parser, split_args = template_arg_parser(
+        default_prefix="", desc=textwrap.dedent(CaprotoStringIOC.__doc__)
+    )
+    ioc_options, run_options = check_args(parser, split_args)
+    ioc = CaprotoStringIOC(**ioc_options)
+
+    run(ioc.pvdb, **run_options)

src/srx_caproto_iocs/example/ophyd.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from ophyd import Component as Cpt
+from ophyd import Device, EpicsSignal
+
+
+class OphydChannelTypes(Device):
+    """An ophyd Device which works with the CaprotoStringIOC caproto IOC."""
+
+    bare_string = Cpt(EpicsSignal, "bare_string", string=True)
+    implicit_string_type = Cpt(EpicsSignal, "implicit_string_type", string=True)
+    string_type = Cpt(EpicsSignal, "string_type", string=True)
+    string_type_enum = Cpt(EpicsSignal, "string_type_enum", string=True)
+    char_type_as_string = Cpt(EpicsSignal, "char_type_as_string", string=True)
+    char_type = Cpt(EpicsSignal, "char_type", string=True)

tests/conftest.py

@@ -2,65 +2,96 @@
 
 import os
 import socket
+import string
 import subprocess
 import sys
 import time as ttime
-from pprint import pformat
 
-import netifaces
 import pytest
 
 from srx_caproto_iocs.base import OphydDeviceWithCaprotoIOC
+from srx_caproto_iocs.example.ophyd import OphydChannelTypes
 
 CAPROTO_PV_PREFIX = "BASE:{{Dev:Save1}}:"
 OPHYD_PV_PREFIX = CAPROTO_PV_PREFIX.replace("{{", "{").replace("}}", "}")
 
 
-@pytest.fixture()
-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 base_caproto_ioc(wait=5):
+def get_epics_env():
     first_three = ".".join(socket.gethostbyname(socket.gethostname()).split(".")[:3])
     broadcast = f"{first_three}.255"
 
     print(f"{broadcast = }")
 
-    env = {
+    # from pprint import pformat
+    # import netifaces
+    # interfaces = netifaces.interfaces()
+    # print(f"{interfaces = }")
+    # for interface in interfaces:
+    #     addrs = netifaces.ifaddresses(interface)
+    #     try:
+    #         print(f"{interface = }: {pformat(addrs[netifaces.AF_INET])}")
+    #     except Exception as e:
+    #         print(f"{interface = }: exception:\n  {e}")
+
+    return {
         "EPICS_CAS_BEACON_ADDR_LIST": os.getenv("EPICS_CA_ADDR_LIST", broadcast),
         "EPICS_CAS_AUTO_BEACON_ADDR_LIST": "no",
     }
 
-    print(f"Updating env with:\n\n{pformat(env)}\n")
-    os.environ.update(env)
 
-    interfaces = netifaces.interfaces()
-    print(f"{interfaces = }")
-    for interface in interfaces:
-        addrs = netifaces.ifaddresses(interface)
-        try:
-            print(f"{interface = }: {pformat(addrs[netifaces.AF_INET])}")
-        except Exception as e:
-            print(f"{interface = }: exception:\n  {e}")
+def start_ioc_subprocess(ioc_name="srx_caproto_iocs.base", pv_prefix=CAPROTO_PV_PREFIX):
+    env = get_epics_env()
 
-    command = f"{sys.executable} -m srx_caproto_iocs.base --prefix={CAPROTO_PV_PREFIX} --list-pvs"
+    command = f"{sys.executable} -m {ioc_name} --prefix={pv_prefix} --list-pvs"
     print(
         f"\nStarting caproto IOC in via a fixture using the following command:\n\n  {command}\n"
     )
-    p = subprocess.Popen(
+    os.environ.update(env)
+    return subprocess.Popen(
         command.split(),
         start_new_session=True,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
         shell=False,
         env=os.environ,
     )
+
+
+@pytest.fixture(scope="session")
+def base_caproto_ioc(wait=5):
+    p = start_ioc_subprocess(
+        ioc_name="srx_caproto_iocs.base", pv_prefix=CAPROTO_PV_PREFIX
+    )
+
+    print(f"Wait for {wait} seconds...")
+    ttime.sleep(wait)
+
+    yield p
+
+    p.terminate()
+
+    std_out, std_err = p.communicate()
+    std_out = std_out.decode()
+    sep = "=" * 80
+    print(f"STDOUT:\n{sep}\n{std_out}")
+    print(f"STDERR:\n{sep}\n{std_err}")
+
+
+@pytest.fixture()
+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_channel_types(wait=5):
+    p = start_ioc_subprocess(
+        ioc_name="srx_caproto_iocs.example.caproto_ioc", pv_prefix=CAPROTO_PV_PREFIX
+    )
+
     print(f"Wait for {wait} seconds...")
     ttime.sleep(wait)
 
@@ -73,3 +104,12 @@ def base_caproto_ioc(wait=5):
     sep = "=" * 80
     print(f"STDOUT:\n{sep}\n{std_out}")
     print(f"STDERR:\n{sep}\n{std_err}")
+
+
+@pytest.fixture()
+def ophyd_channel_types():
+    dev = OphydChannelTypes(OPHYD_PV_PREFIX, name="ophyd_channel_type")
+    letters = iter(string.ascii_letters)
+    for cpt in sorted(dev.component_names):
+        getattr(dev, cpt).put(next(letters))
+    return dev