Ticket8780 alarms config

src/server_common/schema/blocks.xsd

@@ -26,6 +26,11 @@
         <xs:element ref="blk:log_deadband" minOccurs="0"/>
         <xs:element ref="blk:set_block" minOccurs="0"/>
         <xs:element ref="blk:set_block_val" minOccurs="0"/>
+        <xs:element ref="blk:alarm_enabled" minOccurs="0"/>
+		<xs:element ref="blk:alarm_latched" minOccurs="0"/>
+        <xs:element ref="blk:alarm_delay" minOccurs="0"/>
+        <xs:element ref="blk:alarm_guidance" minOccurs="0"/>
+
       </xs:sequence>
     </xs:complexType>
   </xs:element>
@@ -42,4 +47,8 @@
   <xs:element name="log_deadband" type="xs:double"/>
   <xs:element name="set_block" type="xs:string"/>
   <xs:element name="set_block_val" type="xs:string"/>
+  <xs:element name="alarm_enabled" type="xs:string"/>
+  <xs:element name="alarm_latched" type="xs:string"/>
+  <xs:element name="alarm_delay" type="xs:double"/>
+  <xs:element name="alarm_guidance" type="xs:string"/>
 </xs:schema>

src/server_common/utilities.py

@@ -24,11 +24,13 @@
 import threading
 import time
 import zlib
+from typing import Any
 from xml.etree import ElementTree
 
 from server_common.common_exceptions import MaxAttemptsExceededException
 from server_common.loggers.logger import Logger
 
+# ruff: noqa: ANN401
 # Default to base class - does not actually log anything
 LOGGER = Logger()
 _LOGGER_LOCK = threading.RLock()  # To prevent message interleaving between different threads.
@@ -44,7 +46,7 @@
     MAJOR = "MAJOR"
 
 
-def char_waveform(length):
+def char_waveform(length: object) -> dict[str, str | list[int] | Any]:
     """
     Helper function for creating a char waveform PV.
 
@@ -57,7 +59,7 @@
     return {"type": "char", "count": length, "value": [0]}
 
 
-def set_logger(logger):
+def set_logger(logger: Logger) -> None:
     """Sets the logger used by the print_and_log function.
 
     Args:
@@ -67,13 +69,15 @@
     LOGGER = logger
 
 
-def print_and_log(message, severity=SEVERITY.INFO, src="BLOCKSVR"):
+def print_and_log(
+    message: str | object, severity: str = SEVERITY.INFO, src: str = "BLOCKSVR"
+) -> None:
     """Prints the specified message to the console and writes it to the log.
 
     Args:
         message (string|exception): The message to log
-        severity (string, optional): Gives the severity of the message. Expected serverities are MAJOR, MINOR and INFO.
-                                    Default severity is INFO.
+        severity (string, optional): Gives the severity of the message. Expected severities are
+                                    MAJOR, MINOR and INFO. Default severity is INFO.
         src (string, optional): Gives the source of the message. Default source is BLOCKSVR.
     """
     with _LOGGER_LOCK:
@@ -82,23 +86,26 @@
         LOGGER.write_to_log(message, severity, src)
 
 
-def compress_and_hex(value):
+def compress_and_hex(value: str) -> bytes:
     """Compresses the inputted string and encodes it as hex.
 
     Args:
         value (str): The string to be compressed
     Returns:
         bytes : A compressed and hexed version of the inputted string
     """
-    assert type(value) == str, (
-        "Non-str argument passed to compress_and_hex, maybe Python 2/3 compatibility issue\n"
-        "Argument was type {} with value {}".format(value.__class__.__name__, value)
+    (
+        isinstance(value, str),
+        (
+            "Non-str argument passed to compress_and_hex, maybe Python 2/3 compatibility issue\n"
+            "Argument was type {} with value {}".format(value.__class__.__name__, value)
+        ),
     )
     compr = zlib.compress(bytes(value, "utf-8"))
     return binascii.hexlify(compr)
 
 
-def dehex_and_decompress(value):
+def dehex_and_decompress(value: bytes) -> bytes:
     """Decompresses the inputted string, assuming it is in hex encoding.
 
     Args:
@@ -107,23 +114,24 @@
     Returns:
         bytes : A decompressed version of the inputted string
     """
-    assert type(value) == bytes, (
+    assert isinstance(value, bytes), (
         "Non-bytes argument passed to dehex_and_decompress, maybe Python 2/3 compatibility issue\n"
         "Argument was type {} with value {}".format(value.__class__.__name__, value)
     )
     return zlib.decompress(binascii.unhexlify(value))
 
 
-def dehex_and_decompress_waveform(value):
-    """Decompresses the inputted waveform, assuming it is a array of integers representing characters (null terminated).
+def dehex_and_decompress_waveform(value: list) -> bytes:
+    """Decompresses the inputted waveform, assuming it is an array of integers
+        representing characters (null terminated).
 
     Args:
         value (list[int]): The string to be decompressed
 
     Returns:
         bytes : A decompressed version of the inputted string
     """
-    assert type(value) == list, (
+    assert isinstance(value, list), (
         "Non-list argument passed to dehex_and_decompress_waveform\n"
         "Argument was type {} with value {}".format(value.__class__.__name__, value)
     )
@@ -133,7 +141,7 @@
     return dehex_and_decompress(bytes_rep)
 
 
-def convert_to_json(value):
+def convert_to_json(value: object) -> str:
     """Converts the inputted object to JSON format.
 
     Args:
@@ -145,7 +153,7 @@
     return json.dumps(value)
 
 
-def convert_from_json(value):
+def convert_from_json(value: str) -> object:
     """Converts the inputted string into a JSON object.
 
     Args:
@@ -157,7 +165,7 @@
     return json.loads(value)
 
 
-def parse_boolean(string):
+def parse_boolean(string: str) -> bool:
     """Parses an xml true/false value to boolean
 
     Args:
@@ -177,14 +185,18 @@
         raise ValueError(str(string) + ': Attribute must be "true" or "false"')
 
 
-def value_list_to_xml(value_list, grp, group_tag, item_tag):
+def value_list_to_xml(
+    value_list: dict, grp: ElementTree.Element, group_tag: str, item_tag: str
+) -> None:
     """Converts a list of values to corresponding xml.
 
     Args:
-        value_list (dist[str, dict[object, object]]): The dictionary of names and their values, values are in turn a
-            dictonary of names and value {name: {parameter : value, parameter : value}}
+        value_list (dict[str, dict[object, object]]): The dictionary of names and their values,
+            values are in turn a dictionary of names and value {name: {parameter : value,
+            parameter : value}}
         grp (ElementTree.SubElement): The SubElement object to append the list on to
-        group_tag (string): The tag that corresponds to the group for the items given in the list e.g. macros
+        group_tag (string): The tag that corresponds to the group for the items given in the list
+            e.g. macros
         item_tag (string): The tag that corresponds to each item in the list e.g. macro
     """
     xml_list = ElementTree.SubElement(grp, group_tag)
@@ -196,7 +208,7 @@
                 xml_item.set(str(cn), str(cv))
 
 
-def check_pv_name_valid(name):
+def check_pv_name_valid(name: str) -> bool:
     """Checks that text conforms to the ISIS PV naming standard
 
     Args:
@@ -210,15 +222,17 @@
     return True
 
 
-def create_pv_name(name, current_pvs, default_pv, limit=6, allow_colon=False):
+def create_pv_name(
+    name: str, current_pvs: list, default_pv: str, limit: int = 6, allow_colon: bool = False
+) -> str:
     """Uses the given name as a basis for a valid PV.
 
     Args:
         name (string): The basis for the PV
         current_pvs (list): List of already allocated pvs
         default_pv (string): Basis for the PV if name is unreasonable, must be a valid PV name
         limit (integer): Character limit for the PV
-        allow_colon (bool): If True, pv name is allowed to contain colons; if False, remove the colons
+        allow_colon (bool): If True, pv name is allowed to contain colons; else, remove the colons
 
     Returns:
         string : A valid PV
@@ -249,7 +263,7 @@
     return pv
 
 
-def parse_xml_removing_namespace(file_path):
+def parse_xml_removing_namespace(file_path: str) -> Any:
     """Creates an Element object from a given xml file, removing the namespace.
 
     Args:
@@ -265,7 +279,7 @@
     return it.root
 
 
-def waveform_to_string(data):
+def waveform_to_string(data: Any) -> str:
     """
     Args:
         data: waveform as null terminated string
@@ -281,7 +295,7 @@
     return output
 
 
-def ioc_restart_pending(ioc_pv, channel_access):
+def ioc_restart_pending(ioc_pv: Any, channel_access: Any) -> Any:
     """Check if a particular IOC is restarting. Assumes it has suitable restart PV
 
     Args:
@@ -294,7 +308,7 @@
     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: Any) -> Any:
     """
     Attempt to perform a function a number of times in specified intervals before failing.
 
@@ -308,8 +322,8 @@
 
     """
 
-    def _tags_decorator(func):
-        def _wrapper(*args, **kwargs):
+    def _tags_decorator(func: Any) -> Any:
+        def _wrapper(*args: Any, **kwargs: Any) -> Any:
             attempts = 0
             last_exception = ValueError("Max attempts should be > 0, it is {}".format(max_attempts))
             while attempts < max_attempts:
@@ -327,7 +341,7 @@
     return _tags_decorator
 
 
-def remove_from_end(string, text_to_remove):
+def remove_from_end(string: str | None, text_to_remove: str) -> str | None:
     """
     Remove a String from the end of a string if it exists
     Args:
@@ -342,9 +356,10 @@
     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.
+    Takes a collection of strings, and returns it with all strings lowercased and with duplicates
+        removed.
 
     Args:
         in_list (List[str]): the collection of strings to operate on
@@ -355,7 +370,7 @@
     return {x.lower() for x in in_list}
 
 
-def parse_date_time_arg_exit_on_fail(date_arg, error_code=1):
+def parse_date_time_arg_exit_on_fail(date_arg: str, error_code: int = 1) -> datetime.datetime:
     """
     Parse a date argument and exit the program with an error code if that argument is not a date
     Args:
@@ -371,3 +386,21 @@
     except (ValueError, TypeError) as ex:
         print(f"Can not interpret date '{date_arg}' error: {ex}")
         exit(error_code)
+
+
+def dehex_and_decompress_waveform_value(value: str | bytes) -> str:
+    """Decompresses the inputted waveform, assuming it is available as string.
+
+    Args:
+        value: The string to be decompressed
+
+    Returns:
+        str : A decompressed and unhexed version of the input string
+
+    Raises:
+        ValueError : If the supplied string is not valid hexadecimal characters
+    """
+    if value and len(value) % 2 == 0:
+        return zlib.decompress(binascii.unhexlify(value)).decode("utf-8")
+    else:
+        raise ValueError(f"Invalid hex string: value is empty or has odd length ({len(value)})")