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)})")