From 2dea7090e3df434bddebb73d094972eae1d78a66 Mon Sep 17 00:00:00 2001 From: linarphy Date: Tue, 11 Jun 2024 18:30:29 +0200 Subject: [PATCH] Refactor for data organization --- pyproject.toml | 1 + src/backscattering_analyzer/__init__.py | 61 ---- src/backscattering_analyzer/acquisition.py | 22 ++ src/backscattering_analyzer/data.py | 362 +++++++++++++++++---- src/backscattering_analyzer/display.py | 63 ++++ src/backscattering_analyzer/experiment.py | 103 +++--- src/backscattering_analyzer/movements.py | 12 + 7 files changed, 431 insertions(+), 193 deletions(-) create mode 100644 src/backscattering_analyzer/acquisition.py create mode 100644 src/backscattering_analyzer/display.py create mode 100644 src/backscattering_analyzer/movements.py diff --git a/pyproject.toml b/pyproject.toml index 1e6d6d3..568af9d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ packages = ["src/backscattering_analyzer"] [tool.ruff] line-length = 72 +builtins = ["_"] [tool.ruff.format] quote-style = "double" diff --git a/src/backscattering_analyzer/__init__.py b/src/backscattering_analyzer/__init__.py index 4ccecfc..33270a8 100644 --- a/src/backscattering_analyzer/__init__.py +++ b/src/backscattering_analyzer/__init__.py @@ -5,64 +5,3 @@ from gettext import install install(__name__) logger = getLogger(__name__) - - -def show_projection( - experiment: "Experiment", - start_frequency: float | None = None, - end_frequency: float | None = None, -) -> None: - """ - Show projection data with matplotlib - """ - logger.debug(_("showing experiment result")) - from matplotlib.pyplot import ( - loglog, - show, - legend, - close, - vlines, - title, - xlabel, - ylabel, - ) - - excited = experiment.signals["excited"].psd().sqrt() - reference = experiment.signals["reference"].psd().sqrt() - loglog( - experiment.projection.x, - experiment.projection.y, - label="projection", - ) # type: ignore[ReportUnusedCallResult] - loglog(excited.x, excited.y, label=_("excited bench")) # type: ignore[ReportUnusedCallResult] - loglog(reference.x, reference.y, label=_("reference bench")) # type: ignore[ReportUnusedCallResult] - loglog( - (experiment.projection + reference).x, - (experiment.projection + reference).y, - label=_("sum reference + excited"), - ) # type: ignore[ReportUnusedCallResult] - if start_frequency is not None: - vlines( - [ - start_frequency, - ], - min(experiment.projection.y), - max(experiment.projection.y), - color="k", - ) # type: ignore[ReportUnusedCallResult] - if end_frequency is not None: - vlines( - [ - end_frequency, - ], - min(experiment.projection.y), - max(experiment.projection.y), - color="k", - ) # type: ignore[ReportUnusedCallResult] - legend() # type: ignore[ReportUnusedCallResult] - title(experiment.bench) # type: ignore[ReportUnusedCallResult] - xlabel(_("frequency (Hz)")) # type: ignore[ReportUnusedCallResult] - ylabel(_("stray ($\\frac{1} {\\sqrt {Hz}}$)")) # type: ignore[ReportUnusedCallResult] - show() - close() - logger.debug(_("experiment result showed")) diff --git a/src/backscattering_analyzer/acquisition.py b/src/backscattering_analyzer/acquisition.py new file mode 100644 index 0000000..263f244 --- /dev/null +++ b/src/backscattering_analyzer/acquisition.py @@ -0,0 +1,22 @@ +from enum import Enum +from dataclasses import dataclass +from science_signal.signal import Signal +from backscattering_analyzer.movements import Movements + + +class AcquisitionType(Enum): + REFERENCE = 0 + EXCITED = 1 + + +@dataclass +class Acquisition: + """ + A data acquisition with a specific time and duration + """ + + movements: Movements + sensibility: Signal + type: AcquisitionType + time: float + duration: float diff --git a/src/backscattering_analyzer/data.py b/src/backscattering_analyzer/data.py index 76f02d4..16c4f95 100644 --- a/src/backscattering_analyzer/data.py +++ b/src/backscattering_analyzer/data.py @@ -1,87 +1,305 @@ from pathlib import Path +from tomllib import load from numpy.lib.npyio import loadtxt +from science_signal import to_signal from science_signal.signal import Signal from scipy.io.matlab import loadmat from backscattering_analyzer import logger +from backscattering_analyzer.acquisition import Acquisition, AcquisitionType +from backscattering_analyzer.movements import Movements -def load_signals( - folder: Path, bench: str, date: str -) -> dict[str, Signal]: - """ - Load excited and reference signal of an experiment - """ - logger.info( - _("loading signals for {bench} in {date}").format( - bench=bench, date=date - ) - ) - excited_file = folder / "{bench}_{date}_dat.csv".format( - bench=bench, - date=date, - ) - reference_file = folder / "{bench}_{date}_ref.csv".format( - bench=bench, - date=date, - ) - try: - excited_data = loadtxt(excited_file).T - reference_data = loadtxt(reference_file).T +class Data: + def __init__(self, folder: Path, bench: str, date: str): logger.info( - _( - "files successfully loaded:\n{excited}\n{reference}" - ).format(excited=excited_file, reference=reference_file) - ) - except OSError as e: - logger.critical( - _("some file cannot be read: {error}").format(error=e) - ) - raise e - return { - "excited": Signal(excited_data[0], excited_data[1]), - "reference": Signal(reference_data[0], reference_data[1]), - } - - -def load_movements( - folder: Path, bench: str, date: str -) -> dict[str, Signal]: - """ - Load excited movement of the bench and the nearest mirror of an - experiment - """ - logger.info( - _("loading movements for {bench} in {date}").format( - bench=bench, date=date - ) - ) - bench_file = folder / "{bench}_{date}_ben.csv".format( - bench=bench, - date=date, - ) - mirror_file = folder / "{bench}_{date}_mir.csv".format( - bench=bench, - date=date, - ) - try: - bench_movement = loadtxt(bench_file).T - mirror_movement = loadtxt(mirror_file).T - bench_movement[1] = bench_movement[1] * 1e-6 # um -> m - mirror_movement[1] = mirror_movement[1] * 1e-6 # um -> m - logger.info( - _("files successfully loaded:\n{bench}\n{mirror}").format( - bench=bench_file, mirror=mirror_file + _("loading data for {bench} in {date}").format( + bench=bench, date=date ) ) - except OSError as e: - logger.critical( - _("some file cannot be read: {error}").format(error=e) + folders: dict[str, Path] = {} # list of folder + files: dict[str, Path] = {} # list of file + + folders["experiment"] = folder / "experiment_{date}".format( + date=date, ) - raise e - return { - "bench": Signal(bench_movement[0], bench_movement[1]), - "mirror": Signal(mirror_movement[0], mirror_movement[1]), + files["metadata"] = folders["experiment"] / "metadata.toml" + + try: + with open(files["metadata"], "rb") as file: + metadata = load(file) + logger.info( + _("successfully load {metadata}").format( + metadata=files["metadata"], + ) + ) + except OSError as error: + logger.critical( + _("cannot load {metadata}: {error}").format( + metadata=files["metadata"], + error=error, + ) + ) + raise error + + try: + available_benches: list[str] = metadata["general"][ + "benches" + ] + except KeyError: + logger.critical( + _("malformed metadata, no benches provided") + ) + raise Exception( + _("malformed metadata, no benches provided") + ) + + if bench not in available_benches: + logger.critical( + _("no data on {bench} in {date}").format( + bench=bench, + data=date, + ) + ) + raise Exception( + _("no data on {bench} in {date}").format( + bench=bench, + data=date, + ) + ) + + description: str | None = None + try: + description = metadata["general"]["description"] + except KeyError: + logger.warning( + _("malformed metadata, no description provided") + ) + + logger.info( + "data description: {description}".format( + description=description + ) + ) + metadata_reference: dict[str, float] + metadata_excited: dict[str, float] + excited = False + try: + metadata_reference = metadata["benches"][bench]["reference"] + folders["reference"] = folders[ + "experiment" + ] / "{bench}/reference".format( + bench=bench, + ) + logger.debug( + _("there are multiple reference for each benches") + ) + except KeyError: + try: + metadata_reference = metadata["reference"] + folders["reference"] = ( + folders["experiment"] / "reference" + ) + logger.debug( + _("there is one reference for every benches") + ) + except KeyError: + logger.critical( + _("malformed metadata, no reference provided") + ) + raise Exception( + _("malformed metadata, no reference provided") + ) + + try: + metadata_excited = metadata["benches"][bench]["excited"] + folders["excited"] = folders[ + "experiment" + ] / "{bench}/excited".format( + bench=bench, + ) + excited = True + logger.debug(_("excited data have been taken")) + except KeyError: + logger.debug(_("no excited data have been taken")) + + logger.debug(_("metadata fully read, starting to load data")) + + channel_names = channel_name_mapping(bench) + + files["ref_bench"] = folders[ + "reference" + ] / "{channel}.csv".format(channel=channel_names["bench"]) + files["ref_mirror"] = folders[ + "reference" + ] / "{channel}.csv".format(channel=channel_names["mirror"]) + files["ref_sensibility"] = folders[ + "reference" + ] / "{channel}.csv".format(channel=channel_names["sensibility"]) + + self.reference = Acquisition(type = AcquisitionType(0)) + + try: + self.reference.movements = Movements( + bench=to_signal(loadtxt(files["ref_bench"]).T), + mirror=to_signal(loadtxt(files["ref_mirror"]).T), + ) + self.reference.sensibility = to_signal( + loadtxt(files["ref_sensibility"]).T + ) + self.reference.time = self.reference.sensibility[0][0] + self.reference.duration = ( + self.reference.sensibility[0][-1] + - self.reference.sensibility[0][0] + ) + logger.debug(_("successfully loaded reference data")) + try: + self.reference.time = metadata_reference["time"] + self.reference.duration = metadata_reference["duration"] + if ( + self.reference.sensibility[0][0] + != self.reference.time + ): + logger.warning( + _( + "time between metadata and data does not match: " + + "{data} != {metadata}" + ).format( + data=self.reference.sensibility[0][0], + metadata=self.reference.time, + ) + ) + if ( + self.reference.sensibility[0][-1] + - self.reference.sensibility[0][0] + != self.reference.duration + ): + logger.warning( + _( + "duration between metadata and data does not " + + "match: {data} != {metadata}" + ).format( + data=( + self.reference.sensibility[0][-1] + - self.reference.sensibility[0][0] + ), + metadata=self.reference.duration, + ) + ) + except KeyError: + logger.warning( + _( + "malformed metadata, each reference entry should have " + + "its own time and duration. Defining it with data" + ) + ) + except OSError as e: + logger.critical( + _( + "some files could not be read: {error}".format( + error=e + ) + ) + ) + raise e + + logger.debug(_("reference data loaded and checked")) + + self.excited = Acquisition(type = AcquisitionType(1)) + + if excited: + files["exc_bench"] = folders[ + "excited" + ] / "{channel}.csv".format(channel=channel_names["bench"]) + files["exc_mirror"] = folders[ + "excited" + ] / "{channel}.csv".format(channel=channel_names["mirror"]) + files["exc_sensibility"] = folders[ + "excited" + ] / "{channel}.csv".format( + channel=channel_names["sensibility"] + ) + try: + self.excited.movements = Movements( + bench=to_signal(loadtxt(files["exc_bench"]).T), + mirror=to_signal(loadtxt(files["exc_mirror"]).T), + ) + self.excited_sensibility = to_signal( + loadtxt(files["exc_sensibility"]).T + ) + self.excited.time = self.excited.sensibility[0][0] + self.excited.duration = ( + self.excited.sensibility[0][-1] + - self.excited.sensibility[0][0] + ) + logger.debug(_("successfully loadd excited data")) + try: + self.excited.time = metadata_excited["time"] + self.excited.duration = metadata_excited["duration"] + if ( + self.excited.sensibility[0][0] + != self.excited.time + ): + logger.warning( + _( + "time between metadata and data does not " + + "match: {data} != {metadata}" + ).format( + data=self.excited.sensibility[0][0], + metadata=self.excited.time, + ) + ) + if ( + self.excited.sensibility[0][-1] + - self.excited.sensibility[0][0] + != self.excited.duration + ): + logger.warning( + _( + "duration between metadata and data does " + + "not match: {data} != {metadata}" + ).format( + data=( + self.excited.sensibility[0][-1] + - self.excited.sensibility[0][0] + ), + metadata=self.excited.duration, + ) + ) + except KeyError: + logger.warning( + _( + "malformed metadata, each excited entry should " + + "have its own time and duration. Defining it " + + "with data" + ) + ) + except OSError as e: + logger.critical( + _( + "some files could not be read: {error}".format( + error=e + ) + ) + ) + logger.debug(_("excited data loaded and checked")) + + +def channel_name_mapping(bench: str) -> dict[str, str]: + names = { + "sensibility": "h", + "bench": bench, } + if bench in ["SDB1", "SDB2"]: + names["mirror"] = "SR" + elif bench in ["SNEB", "SWEB"]: + names["mirror"] = bench[1:3] + elif bench == "SIB2": + names["mirror"] = "MC" + elif bench == "SPRB": + names["mirror"] = "SR" # not good, choice between MC, SR and PR + else: + raise ValueError(_("Unknow bench {bench}".format(bench=bench))) + return names def load_coupling( diff --git a/src/backscattering_analyzer/display.py b/src/backscattering_analyzer/display.py new file mode 100644 index 0000000..4afe2c3 --- /dev/null +++ b/src/backscattering_analyzer/display.py @@ -0,0 +1,63 @@ +from backscattering_analyzer import logger +from backscattering_analyzer.experiment import Experiment + + +def show_projection( + experiment: Experiment, + start_frequency: float | None = None, + end_frequency: float | None = None, +) -> None: + """ + Show projection data with matplotlib + """ + logger.debug(_("showing experiment result")) + from matplotlib.pyplot import ( + loglog, + show, + legend, + close, + vlines, + title, + xlabel, + ylabel, + ) + + excited = experiment.signals["excited"].psd().sqrt() + reference = experiment.signals["reference"].psd().sqrt() + loglog( + experiment.projection.x, + experiment.projection.y, + label="projection", + ) # type: ignore[ReportUnusedCallResult] + loglog(excited.x, excited.y, label=_("excited bench")) # type: ignore[ReportUnusedCallResult] + loglog(reference.x, reference.y, label=_("reference bench")) # type: ignore[ReportUnusedCallResult] + loglog( + (experiment.projection + reference).x, + (experiment.projection + reference).y, + label=_("sum reference + excited"), + ) # type: ignore[ReportUnusedCallResult] + if start_frequency is not None: + vlines( + [ + start_frequency, + ], + min(experiment.projection.y), + max(experiment.projection.y), + color="k", + ) # type: ignore[ReportUnusedCallResult] + if end_frequency is not None: + vlines( + [ + end_frequency, + ], + min(experiment.projection.y), + max(experiment.projection.y), + color="k", + ) # type: ignore[ReportUnusedCallResult] + legend() # type: ignore[ReportUnusedCallResult] + title(experiment.bench) #  type: ignore[ReportUnusedCallResult] + xlabel(_("frequency (Hz)")) # type: ignore[ReportUnusedCallResult] + ylabel(_("stray ($\\frac{1} {\\sqrt {Hz}}$)")) #  type: ignore[ReportUnusedCallResult] + show() + close() + logger.debug(_("experiment result showed")) diff --git a/src/backscattering_analyzer/experiment.py b/src/backscattering_analyzer/experiment.py index cdae089..91750f7 100644 --- a/src/backscattering_analyzer/experiment.py +++ b/src/backscattering_analyzer/experiment.py @@ -8,9 +8,8 @@ from science_signal import interpolate from science_signal.signal import Signal from backscattering_analyzer import logger from backscattering_analyzer.data import ( + Data, load_coupling, - load_movements, - load_signals, ) from backscattering_analyzer.utils import ( compute_light, @@ -71,8 +70,12 @@ class Experiment: wavelength=self.wavelength ) ) + self.calibrations: dict[str, float] = { + "bench": 1.0, + "mirror": 1.0, + } try: - self.calibrations: dict[str, float] = { + self.calibrations = { "bench": values["benches"][bench]["calib"]["bench"], "mirror": values["benches"][bench]["calib"]["mirror"], } @@ -84,10 +87,6 @@ class Experiment: + "wrong: {error}" ).format(file=values_file, error=e) ) - self.calibrations: dict[str, float] = { - "bench": 1.0, - "mirror": 1.0, - } logger.debug( _( "value of calibrations (bench, mirror): ({bench}, " @@ -97,8 +96,12 @@ class Experiment: mirror=self.calibrations["mirror"], ) ) + self.factors: dict[str, float | None] = { + "pre": None, + "true": None, + } try: - self.factors: dict[str, float | None] = { + self.factors = { "pre": 1e6 * values["benches"][bench]["scatter_factor"][0], "true": values["benches"][bench]["scatter_factor"][0], @@ -111,10 +114,6 @@ class Experiment: + "be searched with a fit" ).format(file=values_file) ) - self.factors: dict[str, float | None] = { - "pre": None, - "true": None, - } logger.debug( _( "values of scattering factor (outscale, real): ({pre}, " @@ -124,8 +123,12 @@ class Experiment: true=self.factors["true"], ) ) + self.powers: dict[str, float] = { + "laser": 23, + "dark_fringe": 8e-3, + } try: - self.powers: dict[str, float] = { + self.powers = { "laser": values["dates"][date]["laser"], "dark_fringe": values["dates"][date]["dark_fringe"], } @@ -137,10 +140,6 @@ class Experiment: + "wrong: {error}" ).format(file=values_file, error=e) ) - self.powers: dict[str, float] = { - "laser": 23, - "dark_fringe": 8e-3, - } logger.debug( _( "values of powers (laser, dark_fringe): ({laser}, " @@ -168,8 +167,9 @@ class Experiment: file=self.modelisation_file, ) ) + data_folder_name: str = "/home/demagny/data" try: - data_folder_name: str = values["general"]["data_folder"] + data_folder_name = values["general"]["data_folder"] except KeyError: logger.error( _( @@ -178,7 +178,6 @@ class Experiment: + "set, if its doesn't exist the program will crash" ).format(file=values_file) ) - data_folder_name = "/home/demagny/data" logger.debug( _("data folder containing signal values: {folder}").format( folder=data_folder_name, @@ -187,41 +186,33 @@ class Experiment: data_folder = Path(data_folder_name) - try: - self.signals: dict[str, Signal] = load_signals( - data_folder, bench, date - ) - logger.debug( - _("excited and reference signal successfully loaded") - ) - except OSError: - logger.critical(_("cannot load signals")) - raise Exception(_("cannot load signals")) + self.data = Data(data_folder, bench, date) - try: - self.movements: dict[str, Signal] = load_movements( - data_folder, bench, date - ) - logger.debug( - _( - "movements of the bench and the mirror " - + "successfully loaded" - ) - ) - except OSError: - logger.critical(_("cannot load movements")) - raise Exception(_("cannot load movements")) - - self.movements["true"] = process_movement( - self.movements["bench"], - self.movements["mirror"], + self.reference_movement: None | Signal = process_movement( + self.data.reference.movements.bench, + self.data.reference.movements.mirror, self.calibrations, vibration_frequency, ) + self.excited_movement: None | Signal = None + try: + self.excited_movement = process_movement( + self.data.excited.movements.bench, + self.data.excited.movements.mirror, + self.calibrations, + vibration_frequency, + ) + except ValueError: + pass logger.debug(_("movement processed")) + model_powers: dict[str, float] = { + "laser": 0, + "dark_fringe": 0, + } + correct_power: bool = False try: - correct_power = bool(values["dates"][date]["correct-power"]) + correct_power = values["dates"][date]["correct-power"] if correct_power: model_powers = { "laser": values["dates"][date]["coupling"]["laser"], @@ -229,11 +220,6 @@ class Experiment: "dark_fringe" ], } - else: - model_powers = { - "laser": 0, - "dark_fringe": 0, - } except KeyError as error: print(error) logger.warning( @@ -244,11 +230,6 @@ class Experiment: + "applied" ) ) - correct_power = False - model_powers = { - "laser": 0, - "dark_fringe": 0, - } try: self.coupling = load_coupling( @@ -285,9 +266,11 @@ class Experiment: Get factor to compute the projection for a given scatter factor """ phase = 4 * pi / self.wavelength - factor_n = (self.movements["true"] * phase).sin().psd().sqrt() + factor_n = (self.excited_movement * phase).sin().psd().sqrt() coupling_n = self.coupling[0] - factor_d = (self.movements["true"] * phase).cos().psd().sqrt() + factor_d = ( + (self.excited_movement["true"] * phase).cos().psd().sqrt() + ) coupling_d = self.coupling[1] if start is None: @@ -330,7 +313,7 @@ class Experiment: start_frequency: float = 15, end_frequency: float = 100, precision: int = 3, - ) -> dict[str, float]: + ) -> dict[str, None | float]: """ Find the best factor (first order only) to get the projection that best match the excited signal diff --git a/src/backscattering_analyzer/movements.py b/src/backscattering_analyzer/movements.py new file mode 100644 index 0000000..5f6c3c1 --- /dev/null +++ b/src/backscattering_analyzer/movements.py @@ -0,0 +1,12 @@ +from science_signal.signal import Signal +from dataclasses import dataclass + + +@dataclass +class Movements: + """ + Container for the movement of the bench and the mirror + """ + + bench: str + mirror: str