start splitting up into server_common[epics]

pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "server_common"  # REQUIRED, is the only field that cannot be marked as dynamic.
 dynamic = ["version"]
-description = "A collection of utilities for python based IOC servers used at ISIS"
+description = "A collection of helper utilities for python used at ISIS"
 readme = "README.md"
 requires-python = ">=3.12"
 license = {file = "LICENSE"}
@@ -39,14 +39,17 @@ classifiers = [
 ]
 
 dependencies = [
+]
+
+[project.optional-dependencies]
+epics = [
   "pcaspy",
   "genie-python[plot]",
   "CaChannel",
   "pysnmp",
 ]
-
-[project.optional-dependencies]
 dev = [
+  "server_common[epics]",
   "ruff>=0.8",
   "pyright",
   "pytest",

src/server_common/channel_access.py

@@ -24,83 +24,13 @@
 # Number of threads to serve caputs
 NUMBER_OF_CAPUT_THREADS = 20
 
-try:
-    from genie_python.channel_access_exceptions import (
-        ReadAccessException,
-        UnableToConnectToPVException,
-    )  # noqa: F401
-except ImportError:
-
-    class UnableToConnectToPVException(IOError):  # noqa: N818
-        """
-        The system is unable to connect to a PV for some reason.
-        """
-
-        def __init__(self, pv_name: str, err: Exception) -> None:
-            super(UnableToConnectToPVException, self).__init__(
-                "Unable to connect to PV {0}: {1}".format(pv_name, err)
-            )
-
-    class ReadAccessException(IOError):  # noqa: N818
-        """
-        PV exists but its value is unavailable to read.
-        """
-
-        def __init__(self, pv_name: str) -> None:
-            super(ReadAccessException, self).__init__(
-                "Read access denied for PV {}".format(pv_name)
-            )
-
-
-try:
-    # noinspection PyUnresolvedReferences
-    from genie_python.genie_cachannel_wrapper import EXIST_TIMEOUT, CaChannelWrapper
-except ImportError:
-    print("ERROR: No genie_python on the system can not import CaChannelWrapper!")
-
-try:
-    from genie_python.genie_cachannel_wrapper import AlarmCondition as AlarmStatus
-    from genie_python.genie_cachannel_wrapper import AlarmSeverity
-except ImportError:
-    from enum import IntEnum
-
-    class AlarmSeverity(IntEnum):
-        """
-        Enum for severity of alarm
-        """
-
-        No = 0
-        Minor = 1
-        Major = 2
-        Invalid = 3
-
-    class AlarmStatus(IntEnum):
-        """
-        Enum for status of alarm
-        """
-
-        BadSub = 16
-        Calc = 12
-        Comm = 9
-        Cos = 8
-        Disable = 18
-        High = 4
-        HiHi = 3
-        HwLimit = 11
-        Link = 14
-        Lolo = 5
-        Low = 6
-        No = 0
-        Read = 1
-        ReadAccess = 20
-        Scam = 13
-        Simm = 19
-        Soft = 15
-        State = 7
-        Timeout = 10
-        UDF = 17
-        Write = 2
-        WriteAccess = 21
+from genie_python.channel_access_exceptions import (
+    ReadAccessException,
+    UnableToConnectToPVException,
+)
+from genie_python.genie_cachannel_wrapper import EXIST_TIMEOUT, CaChannelWrapper
+from genie_python.genie_cachannel_wrapper import AlarmCondition as AlarmStatus # noqa: F401
+from genie_python.genie_cachannel_wrapper import AlarmSeverity # noqa: F401
 
 
 def _create_caput_pool() -> ThreadPoolExecutor:
@@ -138,7 +68,7 @@
         ChannelAccess.thread_pool = _create_caput_pool()
 
     @staticmethod
     def caget(name: str, as_string: bool = False, timeout: float | None = None):
         """Uses CaChannelWrapper from genie_python to get a pv value. We import CaChannelWrapper
          when used as this means the tests can run without having genie_python installed
 
@@ -162,9 +92,9 @@
             return None
 
     @staticmethod
     def caput(
         name: str,
         value,
         wait: bool = False,
         set_pv_value: Callable | None = None,
         safe_not_quick: bool = True,
@@ -212,7 +142,7 @@
 
     @staticmethod
     def caput_retry_on_fail(
         pv_name: str, value, retry_count: int = 5, safe_not_quick: bool = True
     ) -> None:
         """
         Write to a pv and check the value is set, retry if not; raise if run out of retries
@@ -294,7 +224,7 @@
     Exception to be thrown if manager mode was required, but not enabled, for an operation.
     """
 
     def __init__(self, *args, **kwargs) -> None:
         super(ManagerModeRequiredError, self).__init__(*args, **kwargs)
 
 

src/server_common/helpers.py

@@ -1,43 +1,7 @@
 import json
 import os
-import sys
 from typing import Dict
 
-from genie_python import genie as g
-from genie_python.mysql_abstraction_layer import SQLAbstraction
-
-from server_common.ioc_data_source import IocDataSource
-from server_common.utilities import SEVERITY, print_and_log
-
-
-def register_ioc_start(
-    ioc_name: str,
-    pv_database: Dict[str, Dict[str, str]] = None,
-    prefix: str | None = None,
-) -> None:
-    """
-    A helper function to register the start of an ioc.
-    Args:
-        ioc_name: name of the ioc to start
-        pv_database: doctionary of pvs in the iov
-        prefix: prefix of pvs in this ioc
-    """
-    try:
-        exepath = sys.argv[0]
-        if pv_database is None:
-            pv_database = {}
-        if prefix is None:
-            prefix = "none"
-
-        ioc_data_source = IocDataSource(SQLAbstraction("iocdb", "iocdb", "$iocdb"))
-        ioc_data_source.insert_ioc_start(ioc_name, os.getpid(), exepath, pv_database, prefix)
-    except Exception as e:
-        print_and_log(
-            "Error registering ioc start: {}: {}".format(e.__class__.__name__, e),
-            SEVERITY.MAJOR,
-        )
-
-
 def get_macro_values() -> Dict[str, str]:
     """
     Parse macro environment JSON into dict. To make this work use the icpconfigGetMacros program.
@@ -50,9 +14,6 @@ def get_macro_values() -> Dict[str, str]:
     return macros
 
 
-motor_in_set_mode = g.adv.motor_in_set_mode
-
-
 def _get_env_var(name: str) -> str:
     try:
         return os.environ[name]

src/server_common/ioc_startup.py

@@ -0,0 +1,35 @@
+import os
+import sys
+from typing import Dict
+
+from genie_python.mysql_abstraction_layer import SQLAbstraction
+from server_common.ioc_data_source import IocDataSource
+from server_common.utilities import SEVERITY, print_and_log
+
+
+def register_ioc_start(
+    ioc_name: str,
+    pv_database: Dict[str, Dict[str, str]] = None,
+    prefix: str | None = None,
+) -> None:
+    """
+    A helper function to register the start of an ioc.
+    Args:
+        ioc_name: name of the ioc to start
+        pv_database: doctionary of pvs in the iov
+        prefix: prefix of pvs in this ioc
+    """
+    try:
+        exepath = sys.argv[0]
+        if pv_database is None:
+            pv_database = {}
+        if prefix is None:
+            prefix = "none"
+
+        ioc_data_source = IocDataSource(SQLAbstraction("iocdb", "iocdb", "$iocdb"))
+        ioc_data_source.insert_ioc_start(ioc_name, os.getpid(), exepath, pv_database, prefix)
+    except Exception as e:
+        print_and_log(
+            "Error registering ioc start: {}: {}".format(e.__class__.__name__, e),
+            SEVERITY.MAJOR,
+        )

src/server_common/utilities.py

@@ -24,6 +24,7 @@
 import threading
 import time
 import zlib
+from typing import Any, Iterable
 from xml.etree import ElementTree
 
 from server_common.common_exceptions import MaxAttemptsExceededException
@@ -113,6 +114,14 @@ def dehex_and_decompress(value):
     )
     return zlib.decompress(binascii.unhexlify(value))
 
+def dehex_decompress_and_dejson(value: bytes) -> Any:  # noqa: ANN401
+    """
+    Convert string from zipped hexed json to a python representation
+    :param value: value to convert
+    :return: python representation of json
+    """
+    return json.loads(dehex_and_decompress(value))
+
 
 def dehex_and_decompress_waveform(value):
     """Decompresses the inputted waveform, assuming it is a array of integers representing characters (null terminated).
@@ -157,7 +166,7 @@ def convert_from_json(value):
     return json.loads(value)
 
 
-def parse_boolean(string):
+def parse_boolean(string: str) -> bool:
     """Parses an xml true/false value to boolean
 
     Args:
@@ -265,19 +274,21 @@ def parse_xml_removing_namespace(file_path):
     return it.root
 
 
-def waveform_to_string(data):
+def waveform_to_string(data: Iterable[int | str]) -> str:
     """
     Args:
         data: waveform as null terminated string
 
     Returns: waveform as a sting
-
     """
-    output = str()
+    output = ""
     for i in data:
         if i == 0:
             break
-        output += chr(i)
+        if isinstance(i, str):
+            output += i
+        else:
+            output += str(chr(i))
     return output
 
 
@@ -294,7 +305,7 @@ def ioc_restart_pending(ioc_pv, channel_access):
     return channel_access.caget(ioc_pv + ":RESTART", as_string=True) == "Busy"
 
 
-def retry(max_attempts, interval, exception):
+def retry(max_attempts: int, interval: int, exception: BaseException):
     """
     Attempt to perform a function a number of times in specified intervals before failing.
 
@@ -327,7 +338,7 @@ def _wrapper(*args, **kwargs):
     return _tags_decorator
 
 
-def remove_from_end(string, text_to_remove):
+def remove_from_end(string: str, text_to_remove: str) -> str:
     """
     Remove a String from the end of a string if it exists
     Args:
@@ -342,7 +353,7 @@ def remove_from_end(string, text_to_remove):
     return string
 
 
-def lowercase_and_make_unique(in_list):
+def lowercase_and_make_unique(in_list: list[str]) -> set[str]:
     """
     Takes a collection of strings, and returns it with all strings lowercased and with duplicates removed.
 

src/server_common/test_modules/test_file_path_manager.py → tests/test_file_path_manager.py

@@ -35,7 +35,7 @@
 class TestFilePathManagerSequence(unittest.TestCase):
     def setUp(self):
         # Find the schema directory
-        dir = os.path.join(".")
+        dir = os.path.join("../src/server_common/test_modules")
 
         self.config_path = os.path.abspath(CONFIG_PATH)
         self.script_path = os.path.abspath(SCRIPT_PATH)