Skip to content
Snippets Groups Projects
Commit 5090dab3 authored by Simon Mazenoux's avatar Simon Mazenoux
Browse files

Merge branch 'SIGMON-761_implement_get_dbf_method' into 'dev'

[SIGMON-761] implement get_dfb_names method

Closes SIGMON-761

See merge request !375
parents 83ac58f6 36538f3f
No related branches found
No related tags found
2 merge requests!377removed new formatting function, put the functionality in the old one,...,!375[SIGMON-761] implement get_dfb_names method
Pipeline #7580970 passed
......@@ -102,13 +102,13 @@ def _get_metadata(
def _get_metadata_version(circuit_type: str, timestamp: Optional[Union[int, str, datetime]]) -> str:
"""Returns the metadata version based on the provided arguments.
Metadata of all circuits but 'IPD2' have 2 possible versions:
- _V1 valid before _RUN_3_START (inclusive)
- _V2 valid after _RUN_3_START
- _V1 valid before _RUN_3_START
- _V2 valid after _RUN_3_START (inclusive)
Metadata of 'IPD2' have additional version _V3 valid from _RUN_3_QPS_UPGRADE_START
(see: https://its.cern.ch/jira/browse/SIGMON-477):
"""
timestamp = Time.to_unix_timestamp(timestamp) if timestamp else None
if timestamp and timestamp <= _RUN_3_START:
if timestamp and timestamp < _RUN_3_START:
return _V1
if circuit_type.upper() == "IPD2" and (not timestamp or timestamp > _RUN_3_QPS_UPGRADE_START):
return _V3
......@@ -280,6 +280,67 @@ def get_ee_names(circuit_name: str, timestamp: int | str | datetime) -> list[str
)
def get_dfb_names(circuit_name: str, timestamp: int | str | datetime) -> list[str]:
"""Returns the name of the DFB for a given circuit name and timestamp
Args:
circuit_name: circuit name
timestamp: timestamp
Returns:
DFB name for input circuit type(s) narrowed to the provided circuit names(s) valid
at the given point in time.
Raises:
ValueError: If the circuit name is not present in the metadata
Examples:
>>> from lhcsmapi.metadata import signal_metadata
>>> signal_metadata.get_dfb_names("RQ5.L1", 1679052360000000000)
['DFLCS.RR13.RQ5.L1']
"""
try:
circuit_type = get_circuit_type_for_circuit_name(circuit_name)
except KeyError as e:
raise ValueError(f"Circuit name {circuit_name} not present in the metadata.") from e
metadata = _get_metadata(circuit_type, timestamp)
if not isinstance(metadata["CIRCUITS"], dict):
raise TypeError(f"Metadata for {circuit_type} is not a dictionary")
if circuit_name not in metadata["CIRCUITS"]:
raise ValueError(f"Circuit name {circuit_name} not present in metadata for {circuit_type}.")
circuit_metadata = metadata["CIRCUITS"][circuit_name]
if circuit_type in (CircuitType.A60, CircuitType.A80_120):
return []
if circuit_type == CircuitType.A600:
if "LEADS_IT" in circuit_metadata and "LEADS" in circuit_metadata:
raise NotImplementedError(
f"600A circuit {circuit_name} both has a LEADS and a LEADS_IT section. "
"The implementation of get_dfb_names should be reconsidered."
)
return (
[f"{circuit_metadata['LEADS_IT']['%DFB%']}.{circuit_name}"]
if "LEADS_IT" in circuit_metadata
else [f"{circuit_metadata['LEADS']['%DFB%']}.{circuit_name}"]
)
if circuit_type == CircuitType.RB:
return [circuit_metadata["LEADS_ODD"]["%DFB%"], circuit_metadata["LEADS_EVEN"]["%DFB%"]]
if circuit_type in (CircuitType.RQ):
return [circuit_metadata["LEADS"]["%DFB%"]]
if circuit_type in (CircuitType.IT, CircuitType.IPD2, CircuitType.IPD2_B1B2):
return [f"{circuit_metadata['LEADS']['%DFB%']}.{circuit_name}"]
if circuit_type in (CircuitType.IPQ2, CircuitType.IPQ4, CircuitType.IPQ8):
if circuit_metadata["LEADS_B1"]["%DFB%"] != circuit_metadata["LEADS_B2"]["%DFB%"]:
raise NotImplementedError(
f"DFB names for B1 and B2 are different for {circuit_name}. "
"The implementation of get_dfb_names should be reconsidered."
)
return [f"{circuit_metadata['LEADS_B1']['%DFB%']}.{circuit_name}"]
raise NotImplementedError(f"get_dfb_names method does not support circuit type {circuit_type}.")
def get_system_types_per_circuit_name(
circuit_type: str, circuit_name: str, timestamp: Optional[Union[int, str, datetime]] = None
) -> List[str]:
......
......@@ -179,19 +179,29 @@ def test_get_pc_names(circuit_type, circuit_name, expected_pcs):
"circuit_name, timestamp, expected_fgcs",
[
("RCBH11.L1B2", signal_metadata._RUN_3_START - 1, ["RPLA.12L1.RCBH11.L1B2"]), # 60A
("RCBH11.L1B2", signal_metadata._RUN_3_START + 1, ["RPLA.12L1.RCBH11.L1B2"]), # 60A
("RCBH11.L1B2", signal_metadata._RUN_3_START, ["RPLA.12L1.RCBH11.L1B2"]), # 60A
("RCBCH10.L1B1", signal_metadata._RUN_3_START - 1, ["RPLB.RR13.RCBCH10.L1B1"]), # 80-120A
("RCBCH10.L1B1", signal_metadata._RUN_3_START + 1, ["RPLBB.RR13.RCBCH10.L1B1"]), # 80-120A
("RCBCH10.L1B1", signal_metadata._RUN_3_START, ["RPLBB.RR13.RCBCH10.L1B1"]), # 80-120A
("RCBXH1.L1", signal_metadata._RUN_3_START - 1, ["RPMBB.UL14.RCBXH1.L1"]), # 600A
("RCBXH1.L1", signal_metadata._RUN_3_START + 1, ["RPMBB.UL14.RCBXH1.L1"]), # 600A
("RCBXH1.L1", signal_metadata._RUN_3_START, ["RPMBB.UL14.RCBXH1.L1"]), # 600A
(
"RQX.L1",
signal_metadata._RUN_3_START - 1,
["RPHFC.UL14.RQX.L1", "RPMBC.UL14.RTQX1.L1", "RPHGC.UL14.RTQX2.L1"],
), # IT
(
"RQX.L1",
signal_metadata._RUN_3_START,
["RPHFC.UL14.RQX.L1", "RPMBC.UL14.RTQX1.L1", "RPHGC.UL14.RTQX2.L1"],
), # IT
("RD1.L2", signal_metadata._RUN_3_START - 1, ["RPHF.UA23.RD1.L2"]), # IPD
("RD1.L2", signal_metadata._RUN_3_START + 1, ["RPHF.UA23.RD1.L2"]), # IPD
("RD1.L2", signal_metadata._RUN_3_START, ["RPHF.UA23.RD1.L2"]), # IPD
("RQ5.L1", signal_metadata._RUN_3_START - 1, ["RPHGB.RR13.RQ5.L1B1", "RPHGB.RR13.RQ5.L1B2"]), # IPQ
("RQ5.L1", signal_metadata._RUN_3_START + 1, ["RPHSB.RR13.RQ5.L1B1", "RPHSB.RR13.RQ5.L1B2"]), # IPQ
("RQ5.L1", signal_metadata._RUN_3_START, ["RPHSB.RR13.RQ5.L1B1", "RPHSB.RR13.RQ5.L1B2"]), # IPQ
("RB.A12", signal_metadata._RUN_3_START - 1, ["RPTE.UA23.RB.A12"]), # RB
("RB.A12", signal_metadata._RUN_3_START + 1, ["RPTE.UA23.RB.A12"]), # RB
("RB.A12", signal_metadata._RUN_3_START, ["RPTE.UA23.RB.A12"]), # RB
("RQD.A12", signal_metadata._RUN_3_START - 1, ["RPHE.UA23.RQD.A12"]), # RQ
("RQD.A12", signal_metadata._RUN_3_START + 1, ["RPHE.UA23.RQD.A12"]), # RQ
("RQD.A12", signal_metadata._RUN_3_START, ["RPHE.UA23.RQD.A12"]), # RQ
],
)
def test_get_fgc_names_valid_circuit(circuit_name: str, timestamp: int, expected_fgcs: list[str]):
......@@ -209,21 +219,23 @@ def test_get_fgc_names_invalid_circuit():
("circuit_name", "timestamp", "expected_ees"),
[
("RCBH11.L1B2", signal_metadata._RUN_3_START - 1, []), # 60A
("RCBH11.L1B2", signal_metadata._RUN_3_START + 1, []), # 60A
("RCBH11.L1B2", signal_metadata._RUN_3_START, []), # 60A
("RCBCH10.L1B1", signal_metadata._RUN_3_START - 1, []), # 80-120A
("RCBCH10.L1B1", signal_metadata._RUN_3_START + 1, []), # 80-120A
("RCBCH10.L1B1", signal_metadata._RUN_3_START, []), # 80-120A
("RCBXH1.L1", signal_metadata._RUN_3_START - 1, []), # 600A without EE
("RCBXH1.L1", signal_metadata._RUN_3_START + 1, []), # 600A without EE
("RCBXH1.L1", signal_metadata._RUN_3_START, []), # 600A without EE
("RCD.A12B1", signal_metadata._RUN_3_START - 1, ["UA23.RCD.A12B1"]), # 600A with EE
("RCD.A12B1", signal_metadata._RUN_3_START + 1, ["UA23.RCD.A12B1"]), # 600A with EE
("RCD.A12B1", signal_metadata._RUN_3_START, ["UA23.RCD.A12B1"]), # 600A with EE
("RQX.L1", signal_metadata._RUN_3_START - 1, []), # IT
("RQX.L1", signal_metadata._RUN_3_START, []), # IT
("RD1.L2", signal_metadata._RUN_3_START - 1, []), # IPD
("RD1.L2", signal_metadata._RUN_3_START + 1, []), # IPD
("RD1.L2", signal_metadata._RUN_3_START, []), # IPD
("RQ5.L1", signal_metadata._RUN_3_START - 1, []), # IPQ
("RQ5.L1", signal_metadata._RUN_3_START + 1, []), # IPQ
("RQ5.L1", signal_metadata._RUN_3_START, []), # IPQ
("RB.A12", signal_metadata._RUN_3_START - 1, ["RR17.RB.A12", "UA23.RB.A12"]), # RB
("RB.A12", signal_metadata._RUN_3_START + 1, ["RR17.RB.A12", "UA23.RB.A12"]), # RB
("RB.A12", signal_metadata._RUN_3_START, ["RR17.RB.A12", "UA23.RB.A12"]), # RB
("RQD.A12", signal_metadata._RUN_3_START - 1, ["UA23.RQD.A12"]), # RQ
("RQD.A12", signal_metadata._RUN_3_START + 1, ["UA23.RQD.A12"]), # RQ
("RQD.A12", signal_metadata._RUN_3_START, ["UA23.RQD.A12"]), # RQ
],
)
def test_get_ee_names_valid_circuit(circuit_name: str, timestamp: int, expected_ees: list[str]):
......@@ -235,6 +247,50 @@ def test_get_ee_names_invalid_circuit():
signal_metadata.get_ee_names("ABC", 0)
@pytest.mark.parametrize(
("circuit_name", "timestamp", "expected_dfbs"),
[
("RCBH11.L1B2", signal_metadata._RUN_3_START - 1, []), # 60A
("RCBH11.L1B2", signal_metadata._RUN_3_START, []), # 60A
("RCBCH10.L1B1", signal_metadata._RUN_3_START - 1, []), # 80-120A
("RCBCH10.L1B1", signal_metadata._RUN_3_START, []), # 80-120A
("RCBXH1.L1", signal_metadata._RUN_3_START - 1, ["DFLBS.3L1.RCBXH1.L1"]), # 600A
("RCBXH1.L1", signal_metadata._RUN_3_START, ["DFLBS.3L1.RCBXH1.L1"]), # 600A
("RQX.L1", signal_metadata._RUN_3_START - 1, ["DFLX.3L1.RQX.L1"]), # IT
("RQX.L1", signal_metadata._RUN_3_START, ["DFLX.3L1.RQX.L1"]), # IT
("RD1.L2", signal_metadata._RUN_3_START - 1, ["DFLCS.3L2.RD1.L2"]), # IPD
("RD1.L2", signal_metadata._RUN_3_START, ["DFLCS.3L2.RD1.L2"]), # IPD
("RQ5.L1", signal_metadata._RUN_3_START - 1, ["DFLCS.RR13.RQ5.L1"]), # IPQ
("RQ5.L1", signal_metadata._RUN_3_START, ["DFLCS.RR13.RQ5.L1"]), # IPQ
("RB.A12", signal_metadata._RUN_3_START - 1, ["DFLAS.7R1.RB.A12", "DFLAS.7L2.RB.A12"]), # RB
("RB.A12", signal_metadata._RUN_3_START, ["DFLAS.7R1.RB.A12", "DFLAS.7L2.RB.A12"]), # RB
("RQD.A12", signal_metadata._RUN_3_START - 1, ["DFLAS.7L2.RQD.A12"]), # RQ
("RQD.A12", signal_metadata._RUN_3_START, ["DFLAS.7L2.RQD.A12"]), # RQ
],
)
def test_get_dfb_names_valid_circuit(circuit_name: str, timestamp: int, expected_dfbs: list[str]):
assert signal_metadata.get_dfb_names(circuit_name, timestamp) == expected_dfbs
@pytest.mark.parametrize(
("circuit_name"),
[
circuit_name
for circuit_type in signal_metadata.get_circuit_types()
for circuit_name in signal_metadata.get_circuit_names(circuit_type)
],
)
def test_get_dfb_names_no_exceptions(circuit_name):
"""Check that no circuit name raises an exception in the edge cases"""
signal_metadata.get_dfb_names(circuit_name, signal_metadata._RUN_3_START - 1)
signal_metadata.get_dfb_names(circuit_name, signal_metadata._RUN_3_START)
def test_get_dfb_names_invalid_circuit():
with pytest.raises(ValueError, match=re.escape("Circuit name ABC not present in the metadata.")):
signal_metadata.get_dfb_names("ABC", 0)
_SIGNAL_NAMES = [
(
"RB",
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment