"""Define a class to create the proper :class:`.IMeasurementPoint`."""
import logging
from collections.abc import Sequence
from itertools import cycle
import matplotlib.pyplot as plt
import pandas as pd
from multipac_testbench.instruments.factory import InstrumentFactory
from multipac_testbench.measurement_point.global_diagnostics import (
GlobalDiagnostics,
)
from multipac_testbench.measurement_point.i_measurement_point import (
IMeasurementPoint,
)
from multipac_testbench.measurement_point.pick_up import PickUp
[docs]
class IMeasurementPointFactory:
"""Class to create the proper :class:`.GlobalDiagnostics` :class:`.PickUp`.
It infers the proper type, position of instruments as well as the measured
data from the configuration ``TOML`` file and the measurements ``CSV``
file.
"""
def __init__(
self,
is_raw: bool = False,
create_virtual_instruments: bool = True,
commented_lines: Sequence[str] | None = None,
**kwargs,
) -> None:
"""Instantiate the class with its :class:`.InstrumentFactory`.
Parameters
----------
is_raw :
If set to ``True``, input data files is considered to be raw, ie to
contain acquisition voltages instead of physical quantities.
create_virtual_instruments :
If virtual instruments should be created.
commented_lines :
Lines from a power step ``CSV`` file header, stripped of their
comment character. Will be ``None`` in the context of
:class:`.MultipactorTest`, but will be set within
:class:`.PowerStep`.
kwargs :
Keyword arguments that are directly passed down to the
:class:`.InstrumentFactory`.
"""
self.instrument_factory = InstrumentFactory(
is_raw=is_raw,
create_virtual_instruments=create_virtual_instruments,
commented_lines=commented_lines,
**kwargs,
)
[docs]
def run_single(
self,
config_key: str,
config_value: dict,
df_data: pd.DataFrame,
color: tuple[float, float, float],
) -> IMeasurementPoint | None:
"""Create a single measurement point.
Parameters
----------
config_key :
A key from the ``TOML`` file. If 'global' keyword is in the key,
we return a :class:`.GlobalDiagnostics`. Else, we return a
:class:`.PickUp`.
config_value :
Values from the ``TOML`` file corresponding to ``config_key``,
which will passed down to the created :class:`.IMeasurementPoint`.
df_data :
Full data from the ``CSV`` file.
Returns
-------
A :class:`.GlobalDiagnostics` or :class:`.PickUp`.
"""
if "global" in config_key:
return GlobalDiagnostics(
name=config_key,
df_data=df_data,
instrument_factory=self.instrument_factory,
**config_value,
)
if "step_constants" in config_key:
# Handled in PowerSet
return
return PickUp(
name=config_key,
df_data=df_data,
instrument_factory=self.instrument_factory,
color=color,
**config_value,
)
[docs]
def run(
self,
config: dict[str, dict],
df_data: pd.DataFrame,
verbose: bool = False,
) -> tuple[GlobalDiagnostics | None, list[PickUp]]:
"""Create all the measurement points."""
colors = cycle(plt.rcParams["axes.prop_cycle"].by_key()["color"])
measurement_points = []
for key, val in config.items():
mp = self.run_single(
key,
val,
df_data,
color=(0, 0, 0) if "global" in key else next(colors),
)
if mp is None:
continue
measurement_points.append(mp)
global_diagnostics = self._filter_global_diagnostics(
measurement_points, verbose
)
pick_ups = self._filter_pick_ups(measurement_points, verbose)
return global_diagnostics, pick_ups
[docs]
def _filter_global_diagnostics(
self,
measurement_points: list[IMeasurementPoint],
verbose: bool = False,
) -> GlobalDiagnostics | None:
"""Ensure that we have only one :class:`.GlobalDiagnostics` object."""
global_diagnostics = [
x for x in measurement_points if isinstance(x, GlobalDiagnostics)
]
if len(global_diagnostics) == 0:
if verbose:
logging.info("No global diagnostic defined.")
return
if len(global_diagnostics) == 1:
if verbose:
logging.info(
"1 set of global diagnostics defined:\n\t"
f"{global_diagnostics[0]}"
)
return global_diagnostics[0]
raise OSError(
"Several global diagnostics were found! It means that several "
"entries in the ``TOML`` file have the word 'global' in their "
"entry. Please gather them."
)
[docs]
def _filter_pick_ups(
self,
measurement_points: list[IMeasurementPoint],
verbose: bool = False,
) -> list[PickUp]:
"""Print information on the created pick-ups."""
pick_ups = [x for x in measurement_points if isinstance(x, PickUp)]
n_pick_ups = len(pick_ups)
if len(pick_ups) == 0:
raise OSError("No pick-up was defined.")
if verbose:
logging.info(f"{n_pick_ups} pick-ups created:")
for pick_up in pick_ups:
logging.info(f"\t{pick_up}")
return pick_ups