diff --git a/src/server_common/schema/blocks.xsd b/src/server_common/schema/blocks.xsd index fea0e51..6b8f7d0 100644 --- a/src/server_common/schema/blocks.xsd +++ b/src/server_common/schema/blocks.xsd @@ -26,6 +26,11 @@ + + + + + @@ -42,4 +47,8 @@ + + + + diff --git a/src/server_common/utilities.py b/src/server_common/utilities.py index d31d3ab..2136d2c 100644 --- a/src/server_common/utilities.py +++ b/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 @@ class SEVERITY(object): 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 @@ def char_waveform(length): 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,22 +68,24 @@ def set_logger(logger): 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: @@ -90,15 +93,18 @@ def compress_and_hex(value): 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,15 +113,16 @@ def dehex_and_decompress(value): 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 @@ -123,9 +130,9 @@ def dehex_and_decompress_waveform(value): 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) @@ -133,7 +140,7 @@ def dehex_and_decompress_waveform(value): 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 @@ def convert_to_json(value): 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 @@ 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: @@ -177,14 +184,18 @@ def parse_boolean(string): 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 @@ def value_list_to_xml(value_list, grp, group_tag, item_tag): 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,12 +216,16 @@ def check_pv_name_valid(name): 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: @@ -218,7 +233,7 @@ def create_pv_name(name, current_pvs, default_pv, limit=6, allow_colon=False): 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 @@ def create_pv_name(name, current_pvs, default_pv, limit=6, allow_colon=False): 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 +280,7 @@ def parse_xml_removing_namespace(file_path): return it.root -def waveform_to_string(data): +def waveform_to_string(data: Any) -> str: """ Args: data: waveform as null terminated string @@ -273,7 +288,7 @@ def waveform_to_string(data): Returns: waveform as a sting """ - output = str() + output = "" for i in data: if i == 0: break @@ -281,7 +296,7 @@ def waveform_to_string(data): 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 @@ 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: Any) -> Any: """ Attempt to perform a function a number of times in specified intervals before failing. @@ -308,10 +323,10 @@ def retry(max_attempts, interval, exception): """ - 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 @@ def _wrapper(*args, **kwargs): 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 @@ 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. + 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 @@ def lowercase_and_make_unique(in_list): 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 @@ def parse_date_time_arg_exit_on_fail(date_arg, error_code=1): """ 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)})")