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