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