Skip to content
Snippets Groups Projects
Commit 35b1696b authored by Andreas Verdoner Barba's avatar Andreas Verdoner Barba
Browse files

Merge branch 'add_get_dfb_prefixes' into 'dev'

Add get_dfb_prefixes in signal_metadata

See merge request !401
parents a41d3e43 109c0e55
No related branches found
No related tags found
2 merge requests!404Merge Dev into Master,!401Add get_dfb_prefixes in signal_metadata
Pipeline #8567042 failed
......@@ -20,8 +20,7 @@ import os
import pathlib
import re
import warnings
from datetime import datetime
from typing import Literal
from typing import Any, Literal
import pandas as pd
......@@ -241,7 +240,7 @@ def get_fgc_names(circuit_name: str, timestamp: TimeAlias) -> list[str]:
return get_pc_names(circuit_type, circuit_name, timestamp)
def get_ee_names(circuit_name: str, timestamp: int | str | datetime) -> list[str]:
def get_ee_names(circuit_name: str, timestamp: TimeAlias) -> list[str]:
"""Returns the names of the EE systems for a given circuit name and timestamp
Args:
......@@ -254,6 +253,7 @@ def get_ee_names(circuit_name: str, timestamp: int | str | datetime) -> list[str
Raises:
ValueError: If the circuit name is not present in the metadata
TypeError: If the metadata found is not a dictionary
Examples:
>>> from lhcsmapi.metadata import signal_metadata
......@@ -291,7 +291,7 @@ 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]:
def get_dfb_names(circuit_name: str, timestamp: TimeAlias) -> list[str]:
"""Returns the name of the DFB for a given circuit name and timestamp
Args:
......@@ -304,6 +304,7 @@ def get_dfb_names(circuit_name: str, timestamp: int | str | datetime) -> list[st
Raises:
ValueError: If the circuit name is not present in the metadata
TypeError: If the metadata found is not a dictionary
Examples:
>>> from lhcsmapi.metadata import signal_metadata
......@@ -352,6 +353,49 @@ def get_dfb_names(circuit_name: str, timestamp: int | str | datetime) -> list[st
raise NotImplementedError(f"get_dfb_names method does not support circuit type {circuit_type}.")
def get_dfb_prefixes(circuit_name: str, timestamp: TimeAlias) -> list[str]:
"""Returns the DFB prefixes ("%DFB_PREFIX%") for a given circuit name and timestamp
Args:
circuit_name: circuit name
timestamp: timestamp
Returns:
DFB prefixes 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
TypeError: If the metadata found is not a dictionary
Examples:
>>> from lhcsmapi.metadata import signal_metadata
>>> signal_metadata.get_dfb_prefixes("ROF.A56B2", 1679052360000000000)
['DAJB28_07R5_', 'DAJB27_07R5_']
"""
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}.")
def search_dfb_prefix(node: dict[str, Any]) -> list[str]:
prefixes = node.get("%DFB_PREFIX%", [])
for child in node.values():
if isinstance(child, dict):
prefixes += search_dfb_prefix(child)
return prefixes
return list(set(search_dfb_prefix(metadata["CIRCUITS"][circuit_name])))
def get_system_types_per_circuit_name(
circuit_type: str, circuit_name: str, timestamp: TimeAlias | None = None
) -> list[str]:
......@@ -798,7 +842,7 @@ def has_crowbar(circuit_name: str) -> bool:
raise ValueError(f"Circuit name {circuit_name} not present in the metadata.")
def get_fgc_pm_class_name(circuit_name: str, timestamp: int | str | datetime, origin: Literal["self", "ext"]) -> str:
def get_fgc_pm_class_name(circuit_name: str, timestamp: TimeAlias, origin: Literal["self", "ext"]) -> str:
"""Returns the class name to query FGCs from PM
Args:
......
......@@ -292,6 +292,66 @@ def test_get_dfb_names_invalid_circuit():
signal_metadata.get_dfb_names("ABC", 0)
@pytest.mark.parametrize(
("circuit_name", "timestamp", "expected_prefixes"),
[
("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, ["DXAY11_03L1_", "DXAY12_03L1_"]), # 600A
("RCBXH1.L1", signal_metadata._RUN_3_START, ["DXAY11_03L1_", "DXAY12_03L1_"]), # 600A
(
"RQX.L1",
signal_metadata._RUN_3_START - 1,
["DXAX01_03L1_", "DXAX02_03L1_", "DXAX03_03L1_", "DXAX04_03L1_"],
), # IT
(
"RQX.L1",
signal_metadata._RUN_3_START,
["DXAX01_03L1_", "DXAX02_03L1_", "DXAX03_03L1_", "DXAX04_03L1_"],
), # IT
("RD1.L2", signal_metadata._RUN_3_START - 1, ["DXCX05_03L2_", "DXCX06_03L2_"]), # IPD
("RD1.L2", signal_metadata._RUN_3_START, ["DXCX05_03L2_", "DXCX06_03L2_"]), # IPD
("RQ5.L1", signal_metadata._RUN_3_START - 1, ["DLAC06_RR13_", "DLAC07_RR13_", "DLAC08_RR13_"]), # IPQ
("RQ5.L1", signal_metadata._RUN_3_START, ["DLAC06_RR13_", "DLAC07_RR13_", "DLAC08_RR13_"]), # IPQ
(
"RB.A12",
signal_metadata._RUN_3_START - 1,
["DACA05_07L2_", "DACA06_07L2_", "DABA01_07R1_", "DABA02_07R1_"],
), # RB
(
"RB.A12",
signal_metadata._RUN_3_START,
["DACA05_07L2_", "DACA06_07L2_", "DABA01_07R1_", "DABA02_07R1_"],
), # RB
("RQD.A12", signal_metadata._RUN_3_START - 1, ["DACA01_07L2_", "DACA02_07L2_"]), # RQ
("RQD.A12", signal_metadata._RUN_3_START, ["DACA01_07L2_", "DACA02_07L2_"]), # RQ
],
)
def test_get_dfb_prefixes_valid_circuit(circuit_name: str, timestamp: int, expected_prefixes: list[str]):
assert sorted(signal_metadata.get_dfb_prefixes(circuit_name, timestamp)) == sorted(expected_prefixes)
@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_prefixes_no_exceptions(circuit_name):
"""Check that no circuit name raises an exception in the edge cases"""
signal_metadata.get_dfb_prefixes(circuit_name, signal_metadata._RUN_3_START - 1)
signal_metadata.get_dfb_prefixes(circuit_name, signal_metadata._RUN_3_START)
def test_get_dfb_prefixes_invalid_circuit():
with pytest.raises(ValueError, match=re.escape("Circuit name ABC not present in the metadata.")):
signal_metadata.get_dfb_prefixes("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