Commit daad3394 authored by Aleksandra Mnich's avatar Aleksandra Mnich
Browse files

Merge branch 'SIGMON-356_gui_pc_refactoring' into 'dev'

[SIGMON-356] `gui.pc` refactoring

See merge request !173
parents 17e6ce2e c2492b59
Pipeline #4155230 failed with stages
in 7 minutes and 47 seconds
from enum import Enum
class FgcBrowserAction(Enum):
FIND_BUTTON_CLICKED = 1
EVENT_SELECTION_CHANGE = 2
TIME_RANGE_CHANGE = 3
CIRCUIT_NAME_DROPDOWN_CHANGE = 4
import functools
from abc import ABC
from typing import List
from ipywidgets import Dropdown
from lhcsmapi.gui import pc
from lhcsmapi.gui.BaseModule import BaseModule
from lhcsmapi.gui.pc.FgcBrowserAction import FgcBrowserAction
from lhcsmapi.metadata import signal_metadata
class FgcPmCircuitNameDropdownBaseModule(BaseModule, ABC):
class FgcPmCircuitNameDropdownBaseModule(BaseModule):
def __init__(self, circuit_type):
super().__init__()
......@@ -27,8 +27,9 @@ class FgcPmCircuitNameDropdownBaseModule(BaseModule, ABC):
self.widget = self.select_circuit_name_or_prefix_dropdown
def handle_circuit_name_dropdown_change(d, self):
self.mediator.notify(self, FgcBrowserAction.CIRCUIT_NAME_DROPDOWN_CHANGE)
@staticmethod
def handle_circuit_name_dropdown_change(_, self):
self.mediator.notify(self, pc.FgcBrowserAction.CIRCUIT_NAME_DROPDOWN_CHANGE)
def get_circuit_name_or_family(self):
return self.select_circuit_name_or_prefix_dropdown.value
......@@ -47,44 +48,36 @@ def get_circuit_names_or_prefixes_to_display(circuit_type):
:type circuit_type: str
:return: list of circuit names or types
"""
if circuit_type in ['RB', 'IT']:
return signal_metadata.get_circuit_names(circuit_type)
elif circuit_type == 'IPD':
return signal_metadata.get_circuit_names(['IPD2', 'IPD2_B1B2'])
if circuit_type in ['RB', 'IT', 'IPD']:
detailed_types = signal_metadata.get_detailed_circuit_types(signal_metadata.CircuitTypes(circuit_type))
return signal_metadata.get_circuit_names(detailed_types)
elif circuit_type == '60A':
r60A_names = signal_metadata.get_circuit_names('60A')
r60A_names_prefix = list(set([r60A_name.split('.')[0] for r60A_name in r60A_names]))
return sorted(r60A_names_prefix)
return _get_prefixes(signal_metadata.get_circuit_names('60A'))
elif circuit_type == '80-120A':
r80_120A_names = signal_metadata.get_circuit_names('80-120A')
r80_120A_names_prefix = list(
set([signal_metadata.get_family_name_for_80_120A(r80_120A_name) for r80_120A_name in r80_120A_names]))
return sorted(r80_120A_names_prefix)
names = signal_metadata.get_circuit_names('80-120A')
return _get_prefixes([signal_metadata.get_family_name_for_80_120A(name) for name in names])
elif circuit_type == 'IPQ':
ipq_names = signal_metadata.get_circuit_names(['IPQ2', 'IPQ4', 'IPQ8'])
ipqs_names_prefix = list(set([ipq_name.split('.')[0] for ipq_name in ipq_names]))
return sorted(ipqs_names_prefix)
return _get_prefixes(signal_metadata.get_circuit_names(['IPQ2', 'IPQ4', 'IPQ8']))
elif circuit_type == 'RQ':
circuit_names = signal_metadata.get_circuit_names('RQ')
circuit_names_rqd = list(filter(lambda name: 'RQD' in name, circuit_names))
circuit_names_rqd_set = {circuit_name.replace('RQD', 'RQ') for circuit_name in circuit_names_rqd}
return sorted(list(circuit_names_rqd_set))
names = signal_metadata.get_circuit_names('RQ')
return _filter_and_replace(names, 'RQD', 'RQ')
elif circuit_type == '600A':
circuit_names = signal_metadata.get_circuit_names('600A')
circuit_names_without_rcbx = list(filter(lambda name: 'RCBX' not in name, circuit_names))
circuit_names_without_rcbx_rcd = list(filter(lambda name: 'RCD' not in name, circuit_names_without_rcbx))
circuit_names_without_rcbx_rcd_rco = list(filter(lambda name: 'RCO' not in name,
circuit_names_without_rcbx_rcd))
return sorted(list({circuit_name.split('.')[0] for circuit_name in circuit_names_without_rcbx_rcd_rco}))
names = signal_metadata.get_circuit_names('600A')
return _get_prefixes([name for name in names if not any(p in name for p in ['RCBX', 'RCD', 'RCO'])])
elif circuit_type == '600A_RCBXHV':
circuit_names = signal_metadata.get_circuit_names('600A')
circuit_names_rcbxh = list(filter(lambda name: 'RCBXH' in name, circuit_names))
circuit_names_rcbxh_set = {circuit_name.replace('RCBXH', 'RCBX') for circuit_name in circuit_names_rcbxh}
return sorted(list(circuit_names_rcbxh_set))
names = signal_metadata.get_circuit_names('600A')
return _filter_and_replace(names, 'RCBXH', 'RCBX')
elif circuit_type == '600A_RCDO':
circuit_names = signal_metadata.get_circuit_names('600A')
circuit_names_rcd = list(filter(lambda name: 'RCD' in name, circuit_names))
circuit_names_rcd_set = {circuit_name.replace('RCD', 'RC') for circuit_name in circuit_names_rcd}
return sorted(list(circuit_names_rcd_set))
names = signal_metadata.get_circuit_names('600A')
return _filter_and_replace(names, 'RCD', 'RC')
else:
raise KeyError(f'Circuit type {circuit_type} is not supported')
def _get_prefixes(circuit_names):
return sorted(list({name.split('.')[0] for name in circuit_names}))
def _filter_and_replace(circuit_names: List[str], old: str, new: str) -> List[str]:
return sorted(list({name.replace(old, new) for name in circuit_names if old in name}))
import warnings
from ipywidgets import HBox, VBox
from IPython.display import display, clear_output
from lhcsmapi.gui.AnalysisModeDoneByModule import AnalysisModeDoneByModule
from lhcsmapi.gui.pc.FgcPmCircuitNameDropdownBaseModule import FgcPmCircuitNameDropdownBaseModule
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModuleFactory import FgcPmEventSelectBaseModuleFactory
from lhcsmapi.gui.pc.fgc_pm_find_button.FgcPmFindButtonBaseModuleFactory import FgcPmFindButtonBaseModuleFactory
from lhcsmapi.gui.pc import select, find
from lhcsmapi.gui.ModuleMediator import ModuleMediator
from lhcsmapi.gui.pc.FgcBrowserAction import FgcBrowserAction
from lhcsmapi.gui import pc
from lhcsmapi.gui.DateTimeBaseModule import DateTimeBaseModule
......@@ -30,13 +28,13 @@ class FgcPmSearchModuleMediator(ModuleMediator):
second_row = HBox(children=self.analysis_mode_done_by_module.widget)
# Third row - find button + progress text display
self.fgc_pm_find_button = FgcPmFindButtonBaseModuleFactory.create(circuit_type)
self.fgc_pm_find_button = find.get_module_for_circuit_type(circuit_type)
self.fgc_pm_find_button.mediator = self
third_row = HBox(children=self.fgc_pm_find_button.widget)
# Fourth row - FGC PM event Select
self.fgc_pm_event_select = FgcPmEventSelectBaseModuleFactory.create(circuit_type)
self.fgc_pm_event_select = select.get_module_for_circuit_type(circuit_type)
self.fgc_pm_event_select.mediator = self
fourth_row = self.fgc_pm_event_select.widget
......@@ -55,15 +53,15 @@ class FgcPmSearchModuleMediator(ModuleMediator):
circuit_name_or_family = self.get_circuit_name_or_family()
if DateTimeBaseModule.check_date_time(start_date_time, end_date_time):
if event == FgcBrowserAction.FIND_BUTTON_CLICKED:
if event == pc.FgcBrowserAction.FIND_BUTTON_CLICKED:
self._query_fgc_pm_events_update_widget(circuit_name_or_family, end_date_time, start_date_time)
elif event in [FgcBrowserAction.EVENT_SELECTION_CHANGE, FgcBrowserAction.TIME_RANGE_CHANGE]:
print('The analysis is executed in %s mode.' % ('Automatic' if self.is_automatic_mode() else 'Manual'))
elif event in [pc.FgcBrowserAction.EVENT_SELECTION_CHANGE, pc.FgcBrowserAction.TIME_RANGE_CHANGE]:
print(f'The analysis is executed in {"Automatic" if self.is_automatic_mode() else "Manual"} mode.')
self.fgc_pm_event_select.update_event_selection()
elif event == FgcBrowserAction.CIRCUIT_NAME_DROPDOWN_CHANGE:
elif event == pc.FgcBrowserAction.CIRCUIT_NAME_DROPDOWN_CHANGE:
self._update_circuit_name_selection()
else:
raise KeyError('Action %s not supported!' % event)
raise KeyError(f'Action {event} not supported!')
def _query_fgc_pm_events_update_widget(self, circuit_name, end_date_time, start_date_time):
self.fgc_pm_event_select.fgc_pm_events = []
......@@ -76,7 +74,7 @@ class FgcPmSearchModuleMediator(ModuleMediator):
options = self.fgc_pm_event_select.convert_fgc_pm_events_to_options(self.fgc_pm_event_select.fgc_pm_events)
self.fgc_pm_event_select.fgc_pm_event_sel.options = options
else:
print('No FGC PM events for %s from %s to %s' % (circuit_name, start_date_time, end_date_time))
print(f'No FGC PM events for {circuit_name} from {start_date_time} to {end_date_time}')
def _update_circuit_name_selection(self):
self.fgc_pm_event_select.fgc_pm_events = []
......@@ -95,16 +93,7 @@ class FgcPmSearchModuleMediator(ModuleMediator):
return self.analysis_mode_done_by_module.get_author()
def get_fgc_timestamp(self):
if self.fgc_pm_event_select.get_fgc_pm_event() is None:
raise ValueError('The list of FGC PM Entries should contain at least one element. '
'Please adjust circuit name as well as start and end date.')
return self.fgc_pm_event_select.get_selected_timestamp(self.fgc_pm_event_select.get_fgc_pm_event())
return self.fgc_pm_event_select.get_selected_timestamp()
def get_fgc_circuit(self):
if self.fgc_pm_event_select.get_fgc_pm_event() is None:
raise ValueError('The list of FGC PM Entries should contain at least one element. '
'Please adjust circuit name as well as start and end date.')
return self.fgc_pm_event_select.get_selected_circuit_name(self.fgc_pm_event_select.get_fgc_pm_event())
return self.fgc_pm_event_select.get_selected_circuit_name()
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import List, Tuple, Union
class FgcBase(ABC):
""" Base class for the FGC event"""
@abstractmethod
def get_all_circuits(self) -> List[str]:
pass
@abstractmethod
def get_all_timestamps(self) -> List[int]:
pass
@dataclass
class Fgc(FgcBase):
""" Represents an FGC event associated with a single circuit."""
circuit_name: str
timestamp: int
def get_all_circuits(self) -> List[str]:
return [self.circuit_name]
def get_all_timestamps(self) -> List[int]:
return [self.timestamp]
class FgcCompound(FgcBase, ABC):
""" Represents an FGC event associated with many correlated circuits."""
def get_all_circuits(self) -> List[str]:
return [pm.circuit_name for pm in self.get_all_fgcs()]
def get_all_timestamps(self) -> List[int]:
return [pm.timestamp for pm in self.get_all_fgcs()]
@abstractmethod
def get_all_fgcs(self) -> List[Fgc]:
pass
@dataclass
class RqFgc(FgcCompound):
"""RQ specific FGC event."""
rqd: Fgc
rqf: Fgc
def get_all_fgcs(self) -> List[Fgc]:
return [self.rqd, self.rqf]
@dataclass
class RcdoFgc(FgcCompound):
"""RCDO specific FGC event."""
rcd: Fgc
rco: Fgc
def get_all_fgcs(self) -> List[Fgc]:
return [self.rcd, self.rco]
@dataclass
class RcbxhvFgc(FgcCompound):
"""RCBXHV specific FGC event."""
rcbxh: Fgc
rcbxv: Fgc
def get_all_fgcs(self) -> List[Fgc]:
return [self.rcbxh, self.rcbxv]
class FgcBrowserAction(Enum):
FIND_BUTTON_CLICKED = 1
EVENT_SELECTION_CHANGE = 2
TIME_RANGE_CHANGE = 3
CIRCUIT_NAME_DROPDOWN_CHANGE = 4
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModule import FgcPmEventSelectBaseModule
from lhcsmapi.gui.pc.fgc_pm_event_select.IpdFgcPmEventSelectBaseModule import IpdFgcPmEventSelectBaseModule
from lhcsmapi.gui.pc.fgc_pm_event_select.IpqFgcPmEventSelectBaseModule import IpqFgcPmEventSelectBaseModule
from lhcsmapi.gui.pc.fgc_pm_event_select.ItFgcPmEventSelectBaseModule import ItFgcPmEventSelectBaseModule
from lhcsmapi.gui.pc.fgc_pm_event_select.RbFgcPmEventSelectBaseModule import RbFgcPmEventSelectBaseModule
from lhcsmapi.gui.pc.fgc_pm_event_select.RqFgcPmEventSelectBaseModule import RqFgcPmEventSelectBaseModule
from lhcsmapi.gui.pc.fgc_pm_event_select.R600AFgcPmEventSelectBaseModule import \
R600ARcdoFgcPmEventSelectBaseModule, R600ARcbxhvFgcPmEventSelectBaseModule
class FgcPmEventSelectBaseModuleFactory(object):
""" Class builds objects for selecting a FGC PM event and performing actions depending on the circuit type.
"""
@staticmethod
def create(circuit_type: str):
""" Method creating a FgcPmEventSelectBaseModule object for a FGC PM query of a given circuit type
:param circuit_type: circuit type to initiate a FgcPmEventSelectBaseModule object
:return: a FgcPmEventSelectBaseModule object for a given circuit type
"""
if circuit_type == 'RB':
return RbFgcPmEventSelectBaseModule(circuit_type)
elif circuit_type == 'RQ':
return RqFgcPmEventSelectBaseModule(circuit_type)
elif circuit_type == 'IT':
return ItFgcPmEventSelectBaseModule(circuit_type)
elif circuit_type == 'IPD':
return IpdFgcPmEventSelectBaseModule(circuit_type)
elif circuit_type == 'IPQ':
return IpqFgcPmEventSelectBaseModule(circuit_type)
elif circuit_type == '600A_RCDO':
return R600ARcdoFgcPmEventSelectBaseModule(circuit_type)
elif circuit_type == '600A_RCBXHV':
return R600ARcbxhvFgcPmEventSelectBaseModule(circuit_type)
elif circuit_type in ['600A', '80-120A', '60A']:
return FgcPmEventSelectBaseModule(circuit_type)
else:
raise ValueError('Circuit type {} not supported!'.format(circuit_type))
from typing import List, Tuple
from IPython.display import display, Image
from lhcsmapi.Time import Time
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModule import FgcPmEventSelectBaseModule
from lhcsmapi.metadata.MappingMetadata import MappingMetadata
class IpdFgcPmEventSelectBaseModule(FgcPmEventSelectBaseModule):
_repo_url = "https://gitlab.cern.ch/LHCData/lhc-sm-hwc/-/raw/master"
_circuit_class_to_url = {
'MBX': ("%s/figures/ipd/IPD_MBX_D1.png" % _repo_url),
'MBRC': ("%s/figures/ipd/IPD_MBRC_D2_MBRB_D4.png" % _repo_url),
'MBRS': ("%s/figures/ipd/IPD_MBRS_D3.png" % _repo_url),
'MBRB': ("%s/figures/ipd/IPD_MBRC_D2_MBRB_D4.png" % _repo_url)
}
def display_qps_circuit_schematic(self, circuit_name: str) -> None:
circuit_class = MappingMetadata.get_qps_schematic_naming('IPD', circuit_name)
display(Image(url=self._circuit_class_to_url[circuit_class]))
def convert_fgc_pm_events_to_options(self, fgc_pm_events: List[Tuple[str, int]]) -> List[str]:
""" Method converting a list of source and timestamp tuples into a list of strings to be displayed
:param fgc_pm_events: a list of FGC PM source and timestamp event tuples
:return: list of strings with FGC PM source and timestamp
"""
return [Time.to_string(el[1]) for el in fgc_pm_events]
from typing import List, Tuple
from IPython.display import display, Image
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModule import FgcPmEventSelectBaseModule
from lhcsmapi.metadata.MappingMetadata import MappingMetadata
class IpqFgcPmEventSelectBaseModule(FgcPmEventSelectBaseModule):
_repo_url = "https://gitlab.cern.ch/LHCData/lhc-sm-hwc/-/raw/master"
_circuit_class_to_url = {
'2MQM': ("%s/figures/ipq/2MQM_QPS.png" % _repo_url),
'2x2MQM': ("%s/figures/ipq/2x2MQM_QPS.png" % _repo_url),
'2MQML': ("%s/figures/ipq/2MQML_QPS.png" % _repo_url),
'2MQM+2MQML': ("%s/figures/ipq/2MQM_2MQML_QPS.png" % _repo_url),
'2MQM+2MQMC': ("%s/figures/ipq/2MQM_2MQMC_QPS.png" % _repo_url),
'2MQY': ("%s/figures/ipq/2MQY_QPS.png" % _repo_url),
'2x2MQY': ("%s/figures/ipq/2x2MQY_QPS.png" % _repo_url)
}
def display_qps_circuit_schematic(self, circuit_name: str) -> None:
circuit_class = MappingMetadata.get_qps_schematic_naming('IPQ', circuit_name)
display(Image(url=self._circuit_class_to_url[circuit_class]))
def convert_fgc_pm_events_to_options(self, fgc_pm_events: List[Tuple[str, int]]) -> List[str]:
""" Method converting a list of source and timestamp tuples into a list of strings to be displayed
:param fgc_pm_events: a list of FGC PM source and timestamp event tuples
:return: list of strings with FGC PM source and timestamp
"""
return [self.create_pm_circuit_name_timestamp_text(el) for el in fgc_pm_events]
from typing import List, Tuple, Union
from lhcsmapi.Time import Time
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModule import FgcPmEventSelectBaseModule
class ItFgcPmEventSelectBaseModule(FgcPmEventSelectBaseModule):
""" Specific class for search of IT FGC PM events by respective notebook browsers
"""
def convert_fgc_pm_events_to_options(self, fgc_pm_events: List[List[Union[str, int]]]) -> List[str]:
""" Method converting a list of source and timestamp tuples into a list of strings to be displayed
:param fgc_pm_events: a list of FGC PM source and timestamp event tuples
:return: list of strings with FGC PM source and timestamp
"""
return [Time.to_string(el[1]) for el in fgc_pm_events]
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModule import FgcPmEventSelectBaseModule
class R600ARcdoFgcPmEventSelectBaseModule(FgcPmEventSelectBaseModule):
""" Specific class for search of 600A RCD-RCO FGC PM events by respective notebook browsers
"""
def get_selected_circuit_name(self, fgc_pm_event):
""" Method returning circuit name for a selected element in the list. The order is RCD followed by RCO.
:param fgc_pm_event: a source and timestamp FGC PM tuples
:return: a selected circuit name(s)
"""
if len(fgc_pm_event) == 2:
if 'RCD' in fgc_pm_event[0]:
return fgc_pm_event[0], fgc_pm_event[0].replace('RCD', 'RCO')
else:
return fgc_pm_event[0].replace('RCO', 'RCD'), fgc_pm_event[0]
else:
if 'RCD' in fgc_pm_event[0]:
return fgc_pm_event[0], fgc_pm_event[2]
else:
return fgc_pm_event[2], fgc_pm_event[0]
def get_selected_timestamp(self, fgc_pm_event):
""" Method returning a tuple of timestamps for a selected element in the list. The order is RCD followed by RCO.
:param fgc_pm_event: a source and timestamp FGC PM tuples
:return: a selected timestamp
"""
if len(fgc_pm_event) == 2:
if 'RCD' in fgc_pm_event[0]:
return fgc_pm_event[1], float('nan')
else:
return float('nan'), fgc_pm_event[1]
else:
if 'RCD' in fgc_pm_event[0]:
return fgc_pm_event[1], fgc_pm_event[3]
else:
return fgc_pm_event[3], fgc_pm_event[1]
class R600ARcbxhvFgcPmEventSelectBaseModule(FgcPmEventSelectBaseModule):
""" Specific class for search of 600A RCBXH/V FGC PM events by respective notebook browsers
"""
def get_selected_circuit_name(self, fgc_pm_event):
""" Method returning circuit name for a selected element in the list. The order is RCBXH followed by RCBXV.
:param fgc_pm_event: a source and timestamp FGC PM tuples
:return: a selected circuit name(s)
"""
if len(fgc_pm_event) == 2:
if 'RCBXH' in fgc_pm_event[0]:
return fgc_pm_event[0], fgc_pm_event[0].replace('RCBXH', 'RCBXV')
else:
return fgc_pm_event[0].replace('RCBXV', 'RCBXH'), fgc_pm_event[0]
else:
if 'RCBXH' in fgc_pm_event[0]:
return fgc_pm_event[0], fgc_pm_event[2]
else:
return fgc_pm_event[2], fgc_pm_event[0]
def get_selected_timestamp(self, fgc_pm_event):
""" Method returning a tuple of timestamps for a selected element in the list. The order is RCBXH followed by
RCBXV.
:param fgc_pm_event: a source and timestamp FGC PM tuples
:return: a selected timestamp
"""
if len(fgc_pm_event) == 2:
if 'RCBXH' in fgc_pm_event[0]:
return fgc_pm_event[1], float('nan')
else:
return float('nan'), fgc_pm_event[1]
else:
if 'RCBXH' in fgc_pm_event[0]:
return fgc_pm_event[1], fgc_pm_event[3]
else:
return fgc_pm_event[3], fgc_pm_event[1]
from typing import List, Tuple, Union
from lhcsmapi.Time import Time
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModule import FgcPmEventSelectBaseModule
class RbFgcPmEventSelectBaseModule(FgcPmEventSelectBaseModule):
""" Specific class for search of RB FGC PM events by respective notebook browsers
"""
def convert_fgc_pm_events_to_options(self, fgc_pm_events: List[List[Union[str, int]]]) -> List[str]:
""" Method converting a list of source and timestamp tuples into a list of strings to be displayed
:param fgc_pm_events: a list of FGC PM source and timestamp event tuples
:return: list of strings with FGC PM source and timestamp
"""
return [Time.to_string(el[1]) for el in fgc_pm_events]
from typing import List, Tuple, Union
from lhcsmapi.gui.pc.fgc_pm_event_select.FgcPmEventSelectBaseModule import FgcPmEventSelectBaseModule
class RqFgcPmEventSelectBaseModule(FgcPmEventSelectBaseModule):
""" Specific class for search of RQ FGC PM events by respective notebook browsers
"""
def convert_fgc_pm_events_to_options(self, fgc_pm_events: List[Tuple[str, int, str, int]]) -> List[str]:
""" Method converting a list of source and timestamp tuples into a list of strings to be displayed
:param fgc_pm_events: a list of FGC PM source and timestamp event tuples
:return: list of strings with FGC PM source and timestamp
"""
return [self.create_pm_circuit_name_timestamp_text(x) for x in fgc_pm_events]
def get_selected_circuit_name(self, fgc_pm_event: List[Union[str, int]]) -> Tuple[str, str]:
""" Method returning circuit name(s) of a selected event. In case of an event with two circuits,
RQD is returned first (RQF is the second name).
:param fgc_pm_event: a circuit name/timestamp tuple
:return: a list of two circuit names associated with an FGC event.
"""
if len(fgc_pm_event) == 2 and 'RQD' in fgc_pm_event[0]:
return fgc_pm_event[0], fgc_pm_event[0].replace('RQD', 'RQF')
if len(fgc_pm_event) == 2 and 'RQF' in fgc_pm_event[0]:
return fgc_pm_event[0].replace('RQF', 'RQD'), fgc_pm_event[0]
else:
return fgc_pm_event[0], fgc_pm_event[2]
def get_selected_timestamp(self, fgc_pm_event: List[Union[str, int]]) -> Tuple[float, float]:
""" Method returning circuit name(s) of a selected event. In case of an event with two circuits,
RQD is returned first (RQF is the second name).
:param fgc_pm_event: a circuit name/timestamp tuple
:return: a list of two circuit names associated with an FGC event. NaN if only one circuit is concerned.
"""
if len(fgc_pm_event) == 2 and 'RQD' in fgc_pm_event[0]:
return fgc_pm_event[1], float('nan')
if len(fgc_pm_event) == 2 and 'RQF' in fgc_pm_event[0]:
return float('nan'), fgc_pm_event[1]
else:
return fgc_pm_event[1], fgc_pm_event[3]
import functools
from abc import ABC, abstractmethod
from ipywidgets import Button, Label, IntProgress, Layout, Box
from lhcsmapi.gui.BaseModule import BaseModule
from lhcsmapi.gui.pc.FgcBrowserAction import FgcBrowserAction
def convert_source_timestamp_to_circuit_name_timestamp(source_timestamp_df):
output = []
for index, row in source_timestamp_df.iterrows():
pc_name = row['source']
pc_name_split = pc_name.split('.')
circuit_name = '%s.%s' % (pc_name_split[-2], pc_name_split[-1])
output.append((circuit_name, row['timestamp']))
return output
class FgcPmFindButtonBaseModule(BaseModule, ABC):
def __init__(self, circuit_type):
super().__init__()
self.circuit_type = circuit_type
# Widget
self.find_btn = Button(
description='Find FGC PM entries',
tooltip='Click to find FGC PM entries between indicated dates.',
button_style='', # 'success', 'info', 'warning', 'danger' or ''
)
self.find_btn.on_click(functools.partial(FgcPmFindButtonBaseModule.handle_find_button_event, self=self))