"""Define a class to create the proper :class:`.Instrument`."""
import logging
from collections.abc import Sequence
from pprint import pformat
from typing import Any, Literal
import multipac_testbench.instruments as ins
import pandas as pd
from multipac_testbench.instruments.instrument import Instrument
from multipac_testbench.instruments.penning import DiffPenning, Penning
from multipac_testbench.instruments.rpa import RPA
from multipac_testbench.instruments.step_constant import StepConstant
STRING_TO_INSTRUMENT_CLASS = {
"CurrentCalibre": ins.CurrentCalibre,
"CurrentProbe": ins.CurrentProbe,
"DiffPenning": ins.DiffPenning,
"ElectricFieldProbe": ins.FieldProbe,
"FieldProbe": ins.FieldProbe,
"ForwardPower": ins.ForwardPower,
"FrequencySetpoint": ins.FrequencySetpoint,
"PowerSetpoint": ins.PowerSetpoint,
"OpticalFibre": ins.OpticalFibre,
"Penning": ins.Penning,
"PolarizationSetpoint": ins.PolarizationSetpoint,
"PostTrigger": ins.PostTrigger,
"PreTrigger": ins.PreTrigger,
"RPACurrent": ins.RPACurrent,
"RPAPotential": ins.RPAPotential,
"ReflectedPower": ins.ReflectedPower,
"Sync": ins.Sync,
"Trigger": ins.Trigger,
} #:
INSTRUMENT_NAME_T = Literal[
"CurrentCalibre",
"CurrentProbe",
"DiffPenning",
"ElectricFieldProbe",
"FieldProbe",
"ForwardPower",
"FrequencySetpoint",
"OpticalFibre",
"Penning",
"PolarizationSetpoint",
"PostTrigger",
"PowerSetpoint",
"PreTrigger",
"RPACurrent",
"RPAPotential",
"ReflectedPower",
"Sync",
"Trigger",
]
[docs]
class InstrumentFactory:
"""Class to create instruments."""
def __init__(
self,
freq_mhz: float | None = None,
is_raw: bool = False,
create_virtual_instruments: bool = True,
commented_lines: Sequence[str] | None = None,
) -> None:
"""Set user-defined constants to create correspondig instrument.
Parameters
----------
freq_mhz:
Frequency in :unit:`MHz`.
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`.
"""
self.freq_mhz = freq_mhz
self._is_raw = is_raw
self._create_virtual_instruments = create_virtual_instruments
self._commented_lines: Sequence[str] = commented_lines or ()
[docs]
def run(
self,
name: str,
df_data: pd.DataFrame,
class_name: INSTRUMENT_NAME_T,
column_header: str | list[str] | None = None,
header_key: str | None = None,
**instruments_kw: Any,
) -> ins.Instrument | None:
"""Take the proper subclass, instantiate it and return it.
Parameters
----------
name :
Name of the instrument. For clarity, it should match the name of a
column in ``df_data`` when it is possible.
df_data :
Content of the multipactor tests results ``CSV`` file.
class_name :
Name of the instrument class, as given in the ``TOML`` file.
column_header :
Name of the column(s) from which the data of the instrument will
be taken. The default is None, in which case ``column_header`` is
set to ``name``. In general it is not necessary to provide it. An
exception is when several ``CSV`` columns should be loaded in the
instrument.
header_key :
Key to look for in a power step ``CSV`` header. Used to instantiate
:class:`.StepConstant` in the context of :class:`.PowerStep`.
instruments_kw :
Other keyword arguments in the ``TOML`` file.
Returns
-------
Instrument properly subclassed.
"""
constructor = _get_constructor(class_name)
if (
issubclass(constructor, StepConstant)
and header_key is not None
and self._commented_lines
):
return constructor.from_single_csv_header(
name=name,
commented_lines=self._commented_lines,
n_points=len(df_data),
header_key=header_key,
**instruments_kw,
)
if column_header is None:
column_header = name
if column_header not in df_data:
logging.warning(
f"{column_header = } not present in provided file. Skipping "
"associated instrument."
)
return
raw_data = df_data[column_header]
if isinstance(raw_data, pd.DataFrame):
return constructor.from_pd_dataframe(
name, raw_data, **instruments_kw
)
return constructor(
name,
raw_data,
is_raw=self._is_raw,
freq_mhz=self.freq_mhz,
**instruments_kw,
)
[docs]
def run_virtual(
self,
instruments: Sequence[ins.Instrument],
is_global: bool = False,
**kwargs,
) -> list[ins.VirtualInstrument]:
"""Add the implemented :class:`.VirtualInstrument`.
Parameters
----------
instruments :
The :class:`.Instrument` that were already created. They are used
to compute derived quantities, eg :math:`SWR` and :math:`R`.
is_global :
Tells if the :class:`.IMeasurementPoint` from which this method is
called is global. It allows to forbid creation of one
:class:`.Frequency` or one :class:`.SWR` instrument per
:class:`.IMeasurementPoint`.
kwargs :
Other keyword arguments passed to :meth:`._power_related`.
Returns
-------
The created virtual instruments.
"""
if not self._create_virtual_instruments:
return []
virtuals = []
power_related = []
if is_global:
power_related = self._power_related(instruments, **kwargs)
if len(power_related) > 0:
virtuals += power_related
if len(instruments) == 0:
return []
rpa = self._rpa_related(instruments, **kwargs)
if rpa is not None:
virtuals.append(rpa)
diff_pennings = self._pressure_related(instruments, **kwargs)
if diff_pennings is not None:
virtuals += diff_pennings
return virtuals
[docs]
def _get_constructor(class_name: INSTRUMENT_NAME_T) -> type[Instrument]:
"""Get fail-safe proper instrument constructor."""
if class_name not in STRING_TO_INSTRUMENT_CLASS:
raise KeyError(
f"{class_name = } not recognized, allowed values are:\n"
f"{pformat(INSTRUMENT_NAME_T)}\nSee: instruments/factory.py"
)
return STRING_TO_INSTRUMENT_CLASS[class_name]