Merge branch 'SIGMON-326_ccc_notebook_analysis' into 'dev'

[SIGMON-326] ccc notebook analysis

See merge request !168

ccc.py

+from lhcsmapi.api.analysis.qh import QHCCCAnalysis
+from nxcals.spark_session_builder import get_or_create, Flavor
+
+# spark = get_or_create(flavor=Flavor.YARN_MEDIUM)
+
+qhccc = QHCCCAnalysis('qh', '2021-10-18 07:00:00', '2021-10-20 23:01:00', False)
+# qhccc.set_spark(spark)
+
+qhccc.query()
+print(len(qhccc._source_timestamp_qds_df))
+print(qhccc._source_timestamp_qds_df)
+
+qhccc.analyze()
+print(qhccc.get_analysis_output())
+
+# import enum
+
+# class CircuitTypes(enum.Enum):
+#     RB = 1
+#     RQ = 2
+#     IPQ = 3
+#     IPD = 4
+#     IT = 5
+
+# _DETAILED_CIRCUIT_TYPES_MAP = {
+#     'RB': ['RB'],
+#     'RQ': ['RQ'],
+#     'IPQ': ['IPQ2', 'IPQ4', 'IPQ8'],
+#     'IPD': ['IPD2', 'IPD2_B1B2'],
+#     'IT': ['IT']
+# }
+
+# print(str(CircuitTypes.RB.name))
+# print(_DETAILED_CIRCUIT_TYPES_MAP.get(CircuitTypes.RB.name, None))
\ No newline at end of file

lhcsmapi/metadata/signal_metadata.py

@@ -10,6 +10,7 @@ The circuits are divided into main families:
 - 60A: circuits with rated current of 60A
 """
 import copy
+import enum
 import json
 import os
 import re
@@ -37,9 +38,40 @@ _PATHS = {
     '80-120A': 'circuit/80-120A_METADATA.json'
 }
 
+
+class CircuitTypes(enum.Enum):
+    RB = 'RB'
+    RQ = 'RQ'
+    IPQ = 'IPQ'
+    IPD = 'IPD'
+    IT = 'IT'
+
+
+_DETAILED_CIRCUIT_TYPES_MAP = {
+    'RB': ['RB'],
+    'RQ': ['RQ'],
+    'IPQ': ['IPQ2', 'IPQ4', 'IPQ8'],
+    'IPD': ['IPD2', 'IPD2_B1B2'],
+    'IT': ['IT']
+}
+
 _cache = {}
 
 
+def get_detailed_circuit_types(circuit_type: CircuitTypes) -> Union[None, List[str]]:
+    """Returns a list of the detailed circuit types for a specific circuit type.
+    For example, if IPQ is passed as input, ['IPQ2', 'IPQ4', 'IPQ8'] is returned.
+
+    Args:
+        circuit_type: a circuit type (RB, RQ, IPQ, IPD, IT)
+
+    Returns:
+        A list of detailed circuit types as strings
+    
+    """
+    return _DETAILED_CIRCUIT_TYPES_MAP.get(circuit_type.value, None)
+
+
 def _get_metadata(circuit_type: str, timestamp: (Union[int, str, datetime]) = None) -> Dict[str, Union[Dict, str]]:
     """Loads metadata for provided circuit_type valid at the given point in time.
      If no timestamp is provided, the most recent data is returned.