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

Merge branch 'implement_get_generic_circuit_type' into 'dev'

implement get_generic_circuit_type

See merge request !391
parents 7d6e9bb5 fb5cb80f
No related branches found
No related tags found
2 merge requests!391implement get_generic_circuit_type,!381Merge dev into master
Pipeline #8200547 passed with stage
in 7 minutes and 4 seconds
......@@ -68,6 +68,20 @@ class CircuitType(str, enum.Enum):
return str.__str__(self)
class GenericCircuitType(str, enum.Enum):
A60 = "60A"
A600 = "600A"
A80_120 = "80-120A"
IPD = "IPD"
IPQ = "IPQ"
IT = "IT"
RB = "RB"
RQ = "RQ"
def __str__(self) -> str:
return str.__str__(self)
_cache: Dict[str, Dict[str, Union[Dict, str]]] = {}
......@@ -158,7 +172,7 @@ def get_circuit_types() -> List[str]:
def get_circuit_names(
circuit_type: Union[str, List[str]], timestamp: Optional[Union[int, str, datetime]] = None
circuit_type: Optional[Union[str, List[str]]] = None, timestamp: Optional[Union[int, str, datetime]] = None
) -> List[str]:
"""
Args:
......@@ -168,6 +182,9 @@ def get_circuit_names(
Returns:
A list of circuit names for given circuit type(s) valid at the given point in time.
"""
if circuit_type is None:
circuit_type = get_circuit_types()
circuit_types = utils.vectorize(circuit_type)
circuit_names: List[str] = []
for circuit_type_ in circuit_types:
......@@ -491,6 +508,29 @@ def get_circuit_type_for_circuit_name(circuit_name: str) -> str:
raise KeyError(f"Circuit name {circuit_name} does not map to internal metadata.")
def get_generic_circuit_type_for_circuit_name(circuit_name: str) -> GenericCircuitType:
"""Returns the generic circuit type for a given circuit name, and raises an error if the circuit name is not found."""
if circuit_name not in get_circuit_names():
raise ValueError(f"Circuit name {circuit_name} not present in the metadata.")
circuit_type_mapping = {
is_60a: GenericCircuitType.A60,
is_80_120a: GenericCircuitType.A80_120,
is_600a: GenericCircuitType.A600,
is_inner_triplet: GenericCircuitType.IT,
is_ipd: GenericCircuitType.IPD,
is_ipq: GenericCircuitType.IPQ,
is_main_dipole: GenericCircuitType.RB,
is_main_quadrupole: GenericCircuitType.RQ,
}
for check_func, circuit_type in circuit_type_mapping.items():
if check_func(circuit_name):
return circuit_type
raise NotImplementedError(f"Circuit name '{circuit_name}' does not map to a generic circuit type.")
def get_family_name_for_600A(circuit_name: str) -> str:
"""Parses name of the 600A circuit to get its family name:
- RCD-RCO for RCD, RCO circuits
......@@ -611,7 +651,7 @@ def _get_rq_magnet_names_without_u_diode(circuit_name: str) -> List[str]:
mask_magnets_with_nan = (magnet_to_qps_crate_df["Circuit"] == circuit_name) & (
magnet_to_qps_crate_df["Crate U_DIODE_RQx"].isna()
)
return magnet_to_qps_crate_df[mask_magnets_with_nan]["Magnet"].values
return magnet_to_qps_crate_df[mask_magnets_with_nan]["Magnet"].values.tolist()
def get_signal_name(
......@@ -735,14 +775,11 @@ def has_ee(circuit_name: str) -> bool:
Raises:
ValueError: If the circuit name is not present in the metadata
"""
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
circuit_type = get_generic_circuit_type_for_circuit_name(circuit_name)
if circuit_type in (CircuitType.RB, CircuitType.RQ):
if circuit_type in (GenericCircuitType.RB, GenericCircuitType.RQ):
return True
if circuit_type == CircuitType.A600:
if circuit_type == GenericCircuitType.A600:
with open(pathlib.Path(__file__).parent / "circuit" / "circuit_types_location_acctesting.csv") as f:
for row in csv.DictReader(f):
if row["CIRCUIT_NAME"] == circuit_name:
......@@ -859,11 +896,9 @@ def _get_renaming_date_of_fgc_pm_class_names(circuit_name: str) -> int:
Raises:
ValueError: If the circuit name is not present in the metadata.
"""
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
if circuit_type == CircuitType.A60:
circuit_type = get_generic_circuit_type_for_circuit_name(circuit_name)
if circuit_type == GenericCircuitType.A60:
return 1483228800000000000 # 2017-01-01 00:00:00
else:
return 1514764800000000000 # 2018-01-01 00:00:00
......@@ -1083,14 +1118,11 @@ def get_hwc_summary(path: str | pathlib.Path | None = None) -> pd.DataFrame:
def get_cells_for_circuit(circuit_name: str) -> set[str]:
"""Returns the cell names for a given RB or RQ circuit name."""
if is_main_dipole(circuit_name):
file_name = "RB_MagnetCellQpscrateSector.csv"
elif is_main_quadrupole(circuit_name):
file_name = "RQ_MagnetCellQpscrateSector.csv"
else:
raise ValueError(f"{circuit_name} is not a main quadrupole or main dipole circuit.")
circuit_type = get_generic_circuit_type_for_circuit_name(circuit_name)
if circuit_type not in (GenericCircuitType.RB, GenericCircuitType.RQ):
raise ValueError(f"{circuit_name} is not a main dipole or main quadrupole circuit.")
df = pd.read_csv(pathlib.Path(__file__).parent / "magnet" / file_name)
df = pd.read_csv(pathlib.Path(__file__).parent / "magnet" / f"{circuit_type}_MagnetCellQpscrateSector.csv")
return set(df[df["Circuit"] == circuit_name]["Cell"].tolist())
......
......@@ -4,6 +4,7 @@ import warnings
import pytest
from lhcsmapi.metadata import signal_metadata
from lhcsmapi.metadata.signal_metadata import GenericCircuitType
def test_get_metadata():
......@@ -1437,6 +1438,32 @@ def test_get_circuit_type_for_circuit_name_error():
assert "'Circuit name ABC does not map to internal metadata.'" == str(exception.value)
@pytest.mark.parametrize(
("circuit_name", "expected_circuit_type"),
[
("RCBH11.L1B2", GenericCircuitType.A60),
("RCBCH10.L1B1", GenericCircuitType.A80_120),
("RCBXH1.L1", GenericCircuitType.A600),
("RQX.R1", GenericCircuitType.IT),
("RD1.L2", GenericCircuitType.IPD),
("RD2.L1", GenericCircuitType.IPD),
("RQ4.L2", GenericCircuitType.IPQ),
("RQ4.L1", GenericCircuitType.IPQ),
("RQ5.L1", GenericCircuitType.IPQ),
("RB.A12", GenericCircuitType.RB),
("RQD.A12", GenericCircuitType.RQ),
],
)
def test_get_generic_circuit_type_for_circuit_name(circuit_name: str, expected_circuit_type: GenericCircuitType):
assert signal_metadata.get_generic_circuit_type_for_circuit_name(circuit_name) == expected_circuit_type
assert signal_metadata.get_generic_circuit_type_for_circuit_name(circuit_name) == expected_circuit_type.value
def test_get_generic_circuit_type_for_circuit_name_invalid_circuit():
with pytest.raises(ValueError, match=re.escape("Circuit name ABC not present in the metadata.")):
signal_metadata.get_generic_circuit_type_for_circuit_name("ABC")
_NXCALS_SYSTEMS = [("RB", "QDS", "U_QS0", "CMW"), ("RB", ["LEADS_EVEN", "LEADS_ODD"], ["U_RES", "U_HTS"], "CMW")]
......@@ -1535,11 +1562,9 @@ def test_has_ee_valid_circuit_name():
def test_has_ee_invalid_circuit_name():
with pytest.raises(ValueError) as e:
with pytest.raises(ValueError, match=re.escape("Circuit name ABC not present in the metadata.")):
signal_metadata.has_ee("ABC")
assert str(e.value) == "Circuit name ABC not present in the metadata."
def test_has_crowbar_valid_circuit_name():
assert signal_metadata.has_crowbar("RQS.L1B2")
......@@ -1575,15 +1600,15 @@ def test_get_fgc_pm_class_name(circuit_name: str, timestamp: str, expected: str)
def test_get_fgc_pm_class_name_invalid_origin():
with pytest.raises(ValueError) as exception:
with pytest.raises(
ValueError, match=re.escape("The origin of the FGC dump could be only 'self' or 'ext' - provided origin.")
):
signal_metadata.get_fgc_pm_class_name("RCBH11.L1B2", "2016-07-01 00:00:00", "origin")
assert str(exception.value) == "The origin of the FGC dump could be only 'self' or 'ext' - provided origin."
def test_get_fgc_pm_class_name_invalid_circuit_name():
with pytest.raises(ValueError) as exception:
with pytest.raises(ValueError, match=re.escape("Circuit name ABC not present in the metadata.")):
signal_metadata.get_fgc_pm_class_name("ABC", "2016-07-01 00:00:00", "self")
assert str(exception.value) == "Circuit name ABC not present in the metadata."
def test_is_ipq():
......
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