Add tool to create data from cascina or use it localy
This commit is contained in:
parent
ab3d840bd3
commit
2ba73c617a
GPG key ID:
434A18DA831BC610
1 changed files with 169 additions and 0 deletions
169
src/backscattering_analyzer/measurementfactory.py
Normal file
169
src/backscattering_analyzer/measurementfactory.py
Normal file
|
|
@ -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
|
||||||
Loading…
Reference in a new issue