From 2ba73c617a8c7cc828f89bdbf3ad21a390239f7e Mon Sep 17 00:00:00 2001
From: linarphy <linarphy@linarphy.net>
Date: Thu, 3 Apr 2025 18:22:51 +0200
Subject: [PATCH] Add tool to create data from cascina or use it localy
---
.../measurementfactory.py | 169 ++++++++++++++++++
1 file changed, 169 insertions(+)
create mode 100644 src/backscattering_analyzer/measurementfactory.py
diff --git a/src/backscattering_analyzer/measurementfactory.py b/src/backscattering_analyzer/measurementfactory.py
new file mode 100644
index 0000000..7a01264
--- /dev/null
+++ b/src/backscattering_analyzer/measurementfactory.py
@@ -0,0 +1,169 @@
+from pathlib import Path
+
+from h5py import File
+
+from gwpy.timeseries import TimeSeries
+from gwpy.detector import Channel
+
+from backscattering_analyzer.package.metadata.package import (
+ Package as PackageMetadata,
+)
+from backscattering_analyzer.package.metadata.measurement import (
+ Measurement as MeasureMetadata,
+)
+from backscattering_analyzer.package.metadata.bench import (
+ Bench as BenchMetadata,
+)
+from backscattering_analyzer.package.metadata.mirror import (
+ Mirror as MirrorMetadata,
+)
+from backscattering_analyzer.package.bench import Bench
+from backscattering_analyzer.package.mirror import Mirror
+from backscattering_analyzer.package.measurement import Measurement
+from backscattering_analyzer.package import Package
+from backscattering_analyzer.measurement import (
+ Measurement as RuntimeMeasurement,
+)
+
+
+class MeasurementFactory:
+ def make(
+ self,
+ start_time: int,
+ duration: int,
+ benches: dict[str, dict[str, float]],
+ name: str,
+ description: str,
+ injection_power: float,
+ dark_fringe_power: float,
+ path: Path,
+ ) -> Package:
+ temp_metadata = PackageMetadata(name, description)
+ temp_measure_metadata = MeasureMetadata(
+ name,
+ description,
+ injection_power,
+ dark_fringe_power,
+ start_time,
+ duration,
+ )
+ temp_benches: list[Bench] = []
+ temp_mirrors: list[Mirror] = []
+ for bench in benches.keys():
+ temp_benches.append(
+ Bench(
+ BenchMetadata(
+ bench,
+ benches[bench]["fsc"],
+ benches[bench]["calib"],
+ self.get_mirror_name(bench),
+ ), # FIXME: add fsc & calibration
+ TimeSeries.get(
+ Channel("V1:{}_LC_Z".format(bench)),
+ start_time,
+ start_time + duration,
+ ),
+ )
+ )
+ temp_mirrors.append(
+ Mirror(
+ MirrorMetadata(
+ temp_benches[-1].metadata.mirror,
+ benches[bench]["calib_mirror"],
+ ), # FIXME: here too, add calibration
+ TimeSeries.get(
+ Channel(
+ "V1:Sc_{}_MIR_Z".format(
+ temp_benches[-1].metadata.mirror
+ )
+ ),
+ start_time,
+ start_time + duration,
+ ),
+ )
+ )
+ return self.save(
+ Package(
+ temp_metadata,
+ Measurement(
+ temp_measure_metadata,
+ TimeSeries.get(
+ Channel("V1:Hrec_hoft_20000Hz"),
+ start_time,
+ start_time + duration,
+ ),
+ temp_mirrors,
+ temp_benches,
+ ),
+ ),
+ path,
+ )
+
+ def run(self, measurement: Measurement):
+ return RuntimeMeasurement(measurement)
+
+ def get_mirror_name(self, bench_name: str):
+ if bench_name in ["SNEB", "SWEB"]:
+ return bench_name[1:4]
+ if bench_name == "SDB1":
+ return "SR"
+ else:
+ raise Exception("Unknown bench {}".format(bench_name))
+
+ def save(self, package: Package, path: Path) -> Package:
+ with File(path, "w") as file:
+ file.attrs["name"] = package.metadata.name
+ file.attrs["description"] = package.metadata.description
+ measure_group = file.create_group("measures").create_group(
+ package.measurement.metadata.name
+ )
+ measure_group.attrs["name"] = (
+ package.measurement.metadata.name
+ )
+ measure_group.attrs["description"] = (
+ package.measurement.metadata.description
+ )
+ measure_group.attrs["start time"] = (
+ package.measurement.metadata.start_time
+ )
+ measure_group.attrs["duration"] = (
+ package.measurement.metadata.duration
+ )
+ measure_group.attrs["injection power"] = (
+ package.measurement.metadata.injection_power
+ )
+ measure_group.attrs["dark fringe power"] = (
+ package.measurement.metadata.dark_fringe_power
+ )
+ measure_group.create_dataset(
+ "sensitivity",
+ data=package.measurement.sensitivity,
+ )
+
+ mirrors_group = measure_group.create_group("mirrors")
+ for mirror in package.measurement.mirrors:
+ mirror_dset = mirrors_group.create_dataset(
+ mirror.metadata.name,
+ mirror.signal,
+ )
+ mirror_dset["name"] = mirror.metadata.name
+ mirror_dset["calibration factor"] = (
+ mirror.metadata.factor_calib
+ )
+
+ benches_group = measure_group.create_group("benches")
+ for bench in package.measurement.benches:
+ bench_dset = benches_group.create_dataset(
+ bench.metadata.name,
+ data=bench.signal,
+ )
+ bench_dset.attrs["name"] = bench.metadata.name
+ bench_dset.attrs["fsc"] = bench.metadata.fsc
+ bench_dset.attrs["calibration factor"] = (
+ bench.metadata.factor_calib
+ )
+ bench_dset.attrs["associated mirror"] = (
+ bench.metadata.mirror
+ )
+
+ return package