diff --git a/src/accml_lib/core/model/utils/tango_resource_locator.py b/src/accml_lib/core/model/utils/tango_resource_locator.py new file mode 100644 index 0000000..2eb6d20 --- /dev/null +++ b/src/accml_lib/core/model/utils/tango_resource_locator.py @@ -0,0 +1,119 @@ +"""Support of Tango Resource Locator + +Tango resource locators seem to need a bit extra treatment +e.g. Bluesky expects them to be compatible with json names + +accml needs a bit further investigation to fully support +TRL with all available tools without treating TRL specially +""" +from dataclasses import dataclass + + +@dataclass(frozen=True) +class TangoResourceLocator: + """ Lightweight Tango Resource Locator (TRL) for device names only. + + This class intentionally represents *only* the core Tango device name + consisting of exactly three components: + + domain / family / member + + Todo: + review whole stack to find where trl can not be used as + string + + or to say differently that TRL can be used as strings. + These are only made json compatible where imposed by + (external) modules + + Limitations and design choices + ------------------------------- + * Only the device name part of a TRL is supported. + Full TRL features such as protocol (tango://), host:port, + attributes, properties (->), database selectors (#dbase), + or wildcards are deliberately NOT handled here. + + * The class is designed to be a small, explicit value object and + not a full TRL parser. + + * Input tokens are assumed to be already valid Tango name tokens. + No automatic escaping, sanitization, or normalization is performed + except where explicitly documented. + + * Users are responsible for providing JSON-compatible tokens + when serialization is required. The helper method + `json_compatible()` exists for this purpose but is not lossless. + + * Tango device names are case-insensitive. Equality comparisons + are therefore performed in a case-insensitive manner. + + Rationale + --------- + This restricted scope avoids implicit behavior, keeps the object + predictable, and makes failures explicit. Any logic related to full + TRL parsing or environment-specific resolution should live at a + higher level in the application stack. + """ + + domain: str + family: str + member: str + + @classmethod + def from_trl(cls, trl: str): + tmp = trl.split("/") + assert len(tmp) == 3, ( + "Only simple device TRLs of the form 'domain/family/member'" + f' are supported. I received "{trl}" which does not split in three' + ) + for cnt, token in enumerate(tmp): + assert token != "", f'trl "{trl}" split in "{tmp}", but token {cnt} is empty' + domain, family, member = tmp + return cls(domain=domain, family=family, member=member) + + def as_trl(self) -> str: + r = "/".join([self.domain, self.family, self.member]) + return r + + def json_compatible(self) -> str: + return "__".join(map(clear_token, [self.domain, self.family, self.member])) + + def __str__(self): + return self.as_trl() + + def __eq__(self, other): + if not isinstance(other, TangoResourceLocator): + return NotImplemented + return ( + self.domain.lower(), + self.family.lower(), + self.member.lower(), + ) == ( + other.domain.lower(), + other.family.lower(), + other.member.lower(), + ) + + def __hash__(self): + return hash(( + self.domain.lower(), + self.family.lower(), + self.member.lower(), + )) + + +def clear_token(token): + """ + Todo: + jsons complains on this name + """ + return token.replace(".", "_") + + +def name_from_trl(trl): + """ + Todo: + not used ... remove me + """ + prefix, middle, suffix = map(clear_token, trl.split("/")) + return "__".join([prefix, middle, suffix]) diff --git a/src/accml_lib/custom/soleil/manager_setup.py b/src/accml_lib/custom/soleil/manager_setup.py index ce87d2b..5488f37 100644 --- a/src/accml_lib/custom/soleil/manager_setup.py +++ b/src/accml_lib/custom/soleil/manager_setup.py @@ -1,56 +1,10 @@ -from dataclasses import dataclass from typing import Tuple - -from accml.app.tune.bluesky.tune_correction import tune_correction from accml_lib.core.bl.yellow_pages import YellowPages from accml_lib.core.interfaces.utils.liaison_manager import LiaisonManagerBase from accml_lib.core.interfaces.utils.translator_service import TranslatorServiceBase from accml_lib.core.interfaces.utils.yellow_pages import YellowPagesBase - - -def clear_token(token): - """ - Todo: - jsons complains on this name - """ - return token.replace(".", "_") - - -def name_from_trl(trl): - prefix, middle, suffix = map(clear_token, trl.split("/")) - return "__".join([prefix, middle, suffix]) - - -@dataclass(frozen=True) -class TRL: - """Tango resource locator - - Todo: - need to find the appropriate name for the class - and the entries below - - Move to some tango support library - """ - - domain: str - family: str - member: str - - @classmethod - def from_trl(cls, trl: str): - domain, family, member = trl.split("/") - return cls(domain=domain, family=family, member=member) - - def as_trl(self) -> str: - r = "/".join([self.domain, self.family, self.member]) - return r - - def json_compatible(self) -> str: - return "__".join(map(clear_token, [self.domain, self.family, self.member])) - - def __str__(self): - return self.as_trl() +from accml_lib.core.model.utils.tango_resource_locator import TangoResourceLocator def load_managers() -> Tuple[ @@ -64,13 +18,13 @@ def load_managers() -> Tuple[ quad_names += [f"AN03-AR/EM-QP/{id_}" for id_ in ("QD08.01", "QD08.08", "QD11.04", "QD11.05", "QF09.02", "QF09.07", "QF10.03", "QF10.06")] quad_names += [f"AN04-AR/EM-QP/{id_}" for id_ in ("QD08.01", "QD12.08", "QD11.04", "QD15.05", "QF09.02", "QF10.03", "QF13.07", "QF14.06")] quad_names += [f"AN05-AR/EM-QP/{id_}" for id_ in ("QD12.01", "QD15.04", "QD18.08", "QD21.05", "QF13.02", "QF14.03", "QF19.07", "QF20.06")] - quad_ids = [TRL.from_trl(name) for name in quad_names] + quad_ids = [TangoResourceLocator.from_trl(name) for name in quad_names] yp = YellowPages( dict( quadrupoles=quad_ids, # Todo: need to select the correct quad tune_correction_quadrupoles=quad_ids, - tune=[TRL("simulator", "ringsimulator", "ringsimulator")], + tune=[TangoResourceLocator("simulator", "ringsimulator", "ringsimulator")], ) ) return yp, None, None diff --git a/tests/test_core/test_tango_trl.py b/tests/test_core/test_tango_trl.py new file mode 100644 index 0000000..97f94cd --- /dev/null +++ b/tests/test_core/test_tango_trl.py @@ -0,0 +1,70 @@ +# tests/test_tango_resource_locator.py +import pytest +from dataclasses import FrozenInstanceError + +from accml_lib.core.model.utils.tango_resource_locator import TangoResourceLocator, clear_token + + +def test_from_trl_valid_simple(): + trl = "DOMAIN/Fam/Member1" + obj = TangoResourceLocator.from_trl(trl) + # stored exactly as provided by your class (no normalization in your current impl) + assert obj.domain == "DOMAIN" + assert obj.family == "Fam" + assert obj.member == "Member1" + # as_trl and __str__ should return the same canonical TRL + assert obj.as_trl() == "DOMAIN/Fam/Member1" + assert str(obj) == obj.as_trl() + + +def test_from_trl_invalid_parts_raises(): + bad_values = [ + "", # empty + "too/many/parts/x", # 4 parts + "onlyonepart", # 1 part + "two/parts", # 2 parts + "/leading/slash", # 2 meaningful parts (empty first) + ] + for bad in bad_values: + with pytest.raises(AssertionError): + TangoResourceLocator.from_trl(bad) + + +def test_json_compatible_and_clear_token(): + # tokens contain dots which clear_token should convert to underscores + domain = "a.b" + family = "c.d" + member = "e.f" + obj = TangoResourceLocator(domain=domain, family=family, member=member) + j = obj.json_compatible() + # Each dot replaced by underscore, parts joined by "__" + assert j == "a_b__c_d__e_f" + # direct clear_token check + assert clear_token("one.two.three") == "one_two_three" + # ensure no accidental extra separators + assert "__" in j and j.count("__") == 2 + + +def test_equality_case_insensitive_and_hash(): + a = TangoResourceLocator(domain="Lab", family="Power", member="01") + b = TangoResourceLocator(domain="lab", family="power", member="01") + assert a == b + # identical hash for equal objects + assert hash(a) == hash(b) + # can be used as dict keys and in sets + s = {a: "value"} + assert s[b] == "value" + assert a in set([b]) + + +def test_inequality_with_other_types_returns_false(): + trl = TangoResourceLocator(domain="D", family="F", member="M") + # comparing to unrelated type should not raise, but be False + assert (trl == "D/F/M") is False + assert (trl != "D/F/M") is True + + +def test_frozen_immutable_dataclass(): + trl = TangoResourceLocator(domain="X", family="Y", member="Z") + with pytest.raises(FrozenInstanceError): + trl.domain = "new" # cannot assign because frozen