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,6 +24,7 @@
 import threading
 import time
 import zlib
+from typing import Any
 from xml.etree import ElementTree
 
 from server_common.common_exceptions import MaxAttemptsExceededException
@@ -34,7 +35,7 @@
 _LOGGER_LOCK = threading.RLock()  # To prevent message interleaving between different threads.
 
 
-class SEVERITY(object):
+class SEVERITY:
     """
     Standard message severities.
     """
@@ -44,7 +45,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 +58,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,38 +68,43 @@
     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:
-        message = "[{}] {}: {}".format(datetime.datetime.now(), severity, message)
+        message = f"[{datetime.datetime.now()}] {severity}: {message}"  # ruff: ignore [DTZ005]
         print(message)
         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"
+            f"Argument was type {value.__class__.__name__} with value {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,33 +113,34 @@
     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)
+        f"Argument was type {value.__class__.__name__} with value {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)
+        f"Argument was type {value.__class__.__name__} with value {value}"
     )
 
     unicode_rep = waveform_to_string(value)
     bytes_rep = unicode_rep.encode("ascii")
     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 +152,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 +164,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 +184,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 +207,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:
@@ -205,20 +216,24 @@
     Returns:
         bool : True if text conforms to standard, False otherwise
     """
-    if re.match(r"[A-Za-z0-9_]*", name) is None:
-        return False
-    return True
+    return re.match(r"[A-Za-z0-9_]*", name) is not None
 
 
-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 +264,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,23 +280,23 @@
     return it.root
 
 
-def waveform_to_string(data):
+def waveform_to_string(data: Any) -> 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)
     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 +309,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,10 +323,10 @@
 
     """
 
-    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))
+            last_exception = ValueError(f"Max attempts should be > 0, it is {max_attempts}")
             while attempts < max_attempts:
                 try:
                     return func(*args, **kwargs)
@@ -327,7 +342,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 +357,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 +371,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:
@@ -367,7 +383,25 @@
 
     """
     try:
-        return datetime.datetime.strptime(date_arg, "%Y-%m-%dT%H:%M:%S")
+        return datetime.datetime.strptime(date_arg, "%Y-%m-%dT%H:%M:%S")  # ruff:ignore [DTZ007]
     except (ValueError, TypeError) as ex:
         print(f"Can not interpret date '{date_arg}' error: {ex}")
-        exit(error_code)
+        raise SystemExit(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)})")