Refactor for data organization

pyproject.toml

@@ -35,6 +35,7 @@ packages = ["src/backscattering_analyzer"]
 
 [tool.ruff]
 line-length = 72
+builtins = ["_"]
 
 [tool.ruff.format]
 quote-style = "double"

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"))

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

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(
+class Data:
-    folder: Path, bench: str, date: str
+    def __init__(self, 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(
+            _("loading data for {bench} in {date}").format(
                 bench=bench, date=date
             )
         )
-    excited_file = folder / "{bench}_{date}_dat.csv".format(
+        folders: dict[str, Path] = {}  # list of folder
-        bench=bench,
+        files: dict[str, Path] = {}  # list of file
-        date=date,
+
-    )
+        folders["experiment"] = folder / "experiment_{date}".format(
-    reference_file = folder / "{bench}_{date}_ref.csv".format(
-        bench=bench,
             date=date,
         )
+        files["metadata"] = folders["experiment"] / "metadata.toml"
+
         try:
-        excited_data = loadtxt(excited_file).T
+            with open(files["metadata"], "rb") as file:
-        reference_data = loadtxt(reference_file).T
+                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(
                         _(
-                "files successfully loaded:\n{excited}\n{reference}"
+                            "time between metadata and data does not match: "
-            ).format(excited=excited_file, reference=reference_file)
+                            + "{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 file cannot be read: {error}").format(error=e)
+                _(
+                    "some files could not 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]),
-    }
 
+        logger.debug(_("reference data loaded and checked"))
 
-def load_movements(
+        self.excited = Acquisition(type = AcquisitionType(1))
-    folder: Path, bench: str, date: str
+
-) -> dict[str, Signal]:
+        if excited:
-    """
+            files["exc_bench"] = folders[
-    Load excited movement of the bench and the nearest mirror of an
+                "excited"
-    experiment
+            ] / "{channel}.csv".format(channel=channel_names["bench"])
-    """
+            files["exc_mirror"] = folders[
-    logger.info(
+                "excited"
-        _("loading movements for {bench} in {date}").format(
+            ] / "{channel}.csv".format(channel=channel_names["mirror"])
-            bench=bench, date=date
+            files["exc_sensibility"] = folders[
-        )
+                "excited"
-    )
+            ] / "{channel}.csv".format(
-    bench_file = folder / "{bench}_{date}_ben.csv".format(
+                channel=channel_names["sensibility"]
-        bench=bench,
-        date=date,
-    )
-    mirror_file = folder / "{bench}_{date}_mir.csv".format(
-        bench=bench,
-        date=date,
             )
             try:
-        bench_movement = loadtxt(bench_file).T
+                self.excited.movements = Movements(
-        mirror_movement = loadtxt(mirror_file).T
+                    bench=to_signal(loadtxt(files["exc_bench"]).T),
-        bench_movement[1] = bench_movement[1] * 1e-6  # um -> m
+                    mirror=to_signal(loadtxt(files["exc_mirror"]).T),
-        mirror_movement[1] = mirror_movement[1] * 1e-6  # um -> m
+                )
-        logger.info(
+                self.excited_sensibility = to_signal(
-            _("files successfully loaded:\n{bench}\n{mirror}").format(
+                    loadtxt(files["exc_sensibility"]).T
-                bench=bench_file, mirror=mirror_file
+                )
+                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 file cannot be read: {error}").format(error=e)
+                    _(
+                        "some files could not be read: {error}".format(
+                            error=e
                         )
-        raise e
+                    )
-    return {
+                )
-        "bench": Signal(bench_movement[0], bench_movement[1]),
+            logger.debug(_("excited data loaded and checked"))
-        "mirror": Signal(mirror_movement[0], mirror_movement[1]),
+
+
+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(

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"))

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
             )
         )
-        try:
         self.calibrations: dict[str, float] = {
+            "bench": 1.0,
+            "mirror": 1.0,
+        }
+        try:
+            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"],
             )
         )
-        try:
         self.factors: dict[str, float | None] = {
+            "pre": None,
+            "true": None,
+        }
+        try:
+            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"],
             )
         )
-        try:
         self.powers: dict[str, float] = {
+            "laser": 23,
+            "dark_fringe": 8e-3,
+        }
+        try:
+            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.data = Data(data_folder, bench, date)
-            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"))
 
-        try:
+        self.reference_movement: None | Signal = process_movement(
-            self.movements: dict[str, Signal] = load_movements(
+            self.data.reference.movements.bench,
-                data_folder, bench, date
+            self.data.reference.movements.mirror,
-            )
-            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.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

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