"""Archiver Module"""
import logging
import tango
from ska_ser_logging import configure_logging
from . import AttributeConfig, get_attribute_configs
configure_logging()
[docs]
def parse_archiver_attribute(line: str) -> tuple[str, AttributeConfig]:
"""
Parse out parameters stored in the archiver AttributeList property.
AFAIK there is only strategy; all other config stored in event settings
of the device.
"""
attr_params = AttributeConfig()
try:
attr, *params = line.split(";")
except ValueError:
attr = line
else:
for param in params:
name, value = param.split("=")
if name == "strategy":
attr_params["archive_strategy"] = value
return attr.lower(), attr_params
[docs]
def update_or_remove(d1, d2: AttributeConfig):
"""Update a dict, but remove Nones and 0."""
for key, value in d2.items():
if not value: # None or 0 means no setting!
d1.pop(key, None)
else:
d1[key] = value
# pylint: disable=unused-argument
[docs]
def get_current_attributes(
archiver: str,
delay: float = 0,
db: tango.Database = None,
get_device_proxy=tango.DeviceProxy,
) -> dict[str, AttributeConfig]:
"""
Return the attributes currently configured, and their configs.
Also return any attributes that could not be checked
(e.g. device not running)
"""
try:
# Note that this works even if the archiver is offline!
archiver_proxy = get_device_proxy(archiver)
archiver_attr_list = archiver_proxy.get_property("AttributeList")[
"AttributeList"
]
except tango.DevFailed as e:
logging.error(
"Could not get properties from archiver %r: %s",
archiver,
e.args[-1].desc,
)
raise RuntimeError("Unable to get archiver config") from e
if db is None:
db = tango.Database()
attributes = dict(
parse_archiver_attribute(attr) for attr in archiver_attr_list
)
archived_attributes = get_attribute_configs(db, list(attributes))
for attr, config in archived_attributes.items():
logging.debug("Checking attr %s in %s", attr, attributes)
if attr in attributes:
config.update(attributes[attr])
return archived_attributes