Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 119 additions & 0 deletions src/accml_lib/core/model/utils/tango_resource_locator.py
Original file line number Diff line number Diff line change
@@ -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])
52 changes: 3 additions & 49 deletions src/accml_lib/custom/soleil/manager_setup.py
Original file line number Diff line number Diff line change
@@ -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[
Expand All @@ -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
70 changes: 70 additions & 0 deletions tests/test_core/test_tango_trl.py
Original file line number Diff line number Diff line change
@@ -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
Loading