"""Shared methods in package."""
import datetime
import hashlib
import logging
import time
from dataclasses import dataclass
from typing import Any, Optional
from astropy.time import Time
from asyncua import Node, ua
from ska_mid_dish_steering_control import SteeringControlUnit
from tango import InfoIt
from ska_mid_dish_ds_manager.models.constants import (
MAX_AZIMUTH,
MAX_ELEVATION,
MIN_AZIMUTH,
MIN_ELEVATION,
OBFUSCATED_ID_HASH_LEN,
SKA_EPOCH,
)
[docs]def get_tai_timestamp_from_unix_s(unix_s: float) -> float:
"""Calculate atomic time in seconds from unix time in seconds.
:param unix_s: Unix time in seconds
:return: atomic time (tai) in seconds
"""
unix_time = Time(unix_s, format="unix")
ska_epoch_tai = Time(SKA_EPOCH, scale="utc").unix_tai
return unix_time.unix_tai - ska_epoch_tai
[docs]def get_tai_timestamp_from_datetime(datetime_obj: datetime.datetime) -> float:
"""Convert a datetime object into a TAI timestamp."""
source_timestamp_unix = datetime_obj.timestamp()
source_timestamp_tai = Time(source_timestamp_unix, format="unix").unix_tai
ska_epoch_tai = Time(SKA_EPOCH, scale="utc").unix_tai
return source_timestamp_tai - ska_epoch_tai
[docs]def get_current_tai_timestamp() -> float:
"""Get the current time as a TAI timestamp."""
return get_tai_timestamp_from_unix_s(time.time())
[docs]def is_point_within_dish_limits(pos_az: float, pos_el: float) -> bool:
"""Check if the given point is within the dishes mechanical limits."""
az_param_valid = MAX_AZIMUTH >= pos_az >= MIN_AZIMUTH
el_param_valid = MAX_ELEVATION >= pos_el >= MIN_ELEVATION
return az_param_valid and el_param_valid
[docs]def are_points_within_dish_limits(points: list[tuple[float, float]]) -> bool:
"""Check if the given points are within the dishes limits."""
for pos_az, pos_el in points:
if not is_point_within_dish_limits(pos_az, pos_el):
return False
return True
[docs]def generate_tango_name_from_opcua_path(opcua_node_path: str) -> str:
"""Generate a tango attribute name.
:param opcua_node_path: OPCUA Node
:type opcua_node_path: Node
:return: generated name
:rtype: str
"""
list_of_node_path = opcua_node_path.split(".")
# Ideally use 3 of the node paths for descriptive and unique names
number_of_node_names_to_use = min(3, len(list_of_node_path))
return "_".join(list_of_node_path[-number_of_node_names_to_use:])
[docs]def generate_component_state_key_from_tango_attr(tango_name: str) -> str:
"""Generate a component state key.
:param tango_name: Tango attribute name
:type tango_name: str
:return: generated key
:rtype: str
"""
return tango_name.lower()
[docs]def generate_component_state_key_from_opcua_path(opcua_node_path: str) -> str:
"""Generate a component state key.
:param opcua_node_path: OPCUA Node
:type opcua_node_path: Node
:return: generated key
:rtype: str
"""
tango_name = generate_tango_name_from_opcua_path(opcua_node_path)
return generate_component_state_key_from_tango_attr(tango_name)
[docs]def convert_opcua_type_to_python_type(opcua_type: str) -> type:
"""Get the python data type for the given opcua type."""
dtype_mapping = {
"Float": float,
"Double": float,
"Boolean": bool,
"String": str,
"Int16": int,
"UInt16": int,
"Int32": int,
"UInt32": int,
"Int64": int,
"UInt64": int,
"Enumeration": int,
}
# TODO: Create tango enum type?
# if attribute_datatype_string_list[0] == "Enumeration":
# enum_values: list[str] = attribute_datatype_string_list[1:]
return dtype_mapping.get(opcua_type, str)
[docs]def get_attribute_data_type_as_python_type(
scu: SteeringControlUnit, opcua_node_path: str
) -> Optional[type]:
"""Get the python data type for an opcua attribute."""
attribute_datatype_string_list = scu.get_attribute_data_type(opcua_node_path)
return convert_opcua_type_to_python_type(attribute_datatype_string_list[0])
[docs]def safe_string_casting(value: str, python_dtype: type) -> Any:
"""Cast a string value to the given python data type."""
if python_dtype is bool:
if value.lower() in ("true", "1"):
converted_arg = True
elif value.lower() in ("false", "0"):
converted_arg = False
else:
raise ValueError()
else:
converted_arg = python_dtype(value)
return converted_arg
[docs]def obfuscate_string(input_string: str) -> str:
"""Obfuscate a string using SHA-256 hashing."""
return hashlib.sha256(input_string.encode()).hexdigest()[:OBFUSCATED_ID_HASH_LEN]
[docs]def get_immutable_user_id(tango_trl: str, logger: logging.Logger | None = None) -> str:
"""Get the unique ID which is used for taking and retaking authority from the DSC.
Input of the form trl="mid-dish/ds-manager/SKA001"
Output has pattern LMC-SKA001-obfuscated(trl)
"""
dish_id = tango_trl.rsplit("/", maxsplit=1)[-1]
if logger is not None:
if not dish_id.startswith("SKA") or len(dish_id) != 6:
logger.warning(
"Unexpected tango trl format: %s, expected format is mid-dish/ds-manager/SKA001",
tango_trl,
)
return f"LMC-{dish_id}-{obfuscate_string(tango_trl)}"
[docs]class BaseInfoIt(InfoIt):
"""Update Tango InfoIt to not truncate the result."""
@staticmethod
def _LogIt__compact_elem(v: Any, **_: Any) -> str:
"""Just return the item as is."""
v = repr(v)
return v
[docs]@dataclass
class SubscriptionEvent:
"""Convert the event to a dataclass for better ergonomics."""
name: str
node: Node
value: Any
source_timestamp: datetime.datetime
server_timestamp: float
data: ua.DataChangeNotification