linarphy/backscattering_experimental_data
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Fix a typo & add option to be faster

Browse source
This commit is contained in:
linarphy 2025-07-11 19:54:02 +02:00
parent 7a645f6572
commit a3cc8ce1d5
Signed by: linarphy
GPG key ID: 0CBF02E039B4FFFF
4 changed files with 119 additions and 57 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
src/backscattering_experimental_data/factory/component.py
View file

@ -17,6 +17,7 @@ class ComponentFactory:
start_time: Time,
duration: Quantity[time],
metadata: MetadataComponent,
force_virgotools: bool = False,
) -> Component:
"""
Fetch a component from online data
@ -31,11 +32,18 @@ class ComponentFactory:
Duration
metadata : MetadataComponent
Component's metadata
force_virgotools : bool
Force the use of the virgotools library in place of gwpy
first to be faster. Default to False
Returns
-------
Component
"""
sensor = download_timeseries(
channel="V1:Sc_{}_MIR_Z".format(metadata.name),
start_time=start_time,
duration=duration,
unit=uV, # in um, but the calibration factor is considered as m/V
force_virgotools=force_virgotools,
)
return Component(metadata=metadata, sensor=sensor)

30
src/backscattering_experimental_data/factory/measurement.py
View file

@ -36,6 +36,7 @@ class MeasurementFactory:
start_time: Time,
duration: Quantity[time],
initializer_metadata: list[InitializerScatterer],
force_virgotools: bool = False,
) -> Measurement:
"""
Fetch a measurement from online data
@ -55,6 +56,32 @@ class MeasurementFactory:
initializer_metadata : list[InitializerScatterer]
List of component's and scatterer's pre-initialized
metadata that will be used in the created measurement
force_virgotools : bool
Force the use of the virgotools library in place of gwpy
first to be faster. Default to False
Examples
--------
>>> data = MeasurementFactory.fetch(
... name="a name",
... description="a detailed description",
... start_time=Time(
... "2025-01-01T00:00:00.000000000",
... format="isot",
... scale="utc",
... ),
... duration=10 * s,
... initializer_metadata=[
... InitializerScatterer(
... name="SDB1",
... calibration_factor=1.0 * m / V,
... associated_component=InitializerComponent(
... name="SR",
... calibration_factor=1.0 * m / V,
... ),
... injection=None,
... ),
... ],
... )
Returns
-------
Measurement : constructed Measurement from these metadata
@ -70,6 +97,7 @@ class MeasurementFactory:
duration=duration,
units=list(channels_map.values()),
frame="trend",
force_virgotools=force_virgotools,
)
if sum(data["V1:META_ITF_LOCK_index"].diff().abs()) != 0:
raise Exception(
@ -101,6 +129,7 @@ class MeasurementFactory:
name=initializer_scatterer.associated_component.name,
calibration_factor=initializer_scatterer.associated_component.calibration_factor,
),
force_virgotools=force_virgotools,
)
components.append(component)
scatterers.append(
@ -120,6 +149,7 @@ class MeasurementFactory:
start_time=start_time,
duration=duration,
unit=None,
force_virgotools=force_virgotools,
)
return Measurement(
metadata=metadata,

8
src/backscattering_experimental_data/factory/scatterer.py
View file

@ -17,6 +17,7 @@ class ScattererFactory:
start_time: Time,
duration: Quantity[time],
metadata: MetadataScatterer,
force_virgotools: bool = False,
) -> Scatterer:
"""
Fetch a component from online data
@ -31,11 +32,18 @@ class ScattererFactory:
Duration
metadata : MetadataComponent
Component's metadata
force_virgotools : bool
Force the use of the virgotools library in place of gwpy
first to be faster. Default to False
Returns
-------
Scatterer
"""
sensor = download_timeseries(
channel="V1:{}_LC_Z".format(metadata.name),
start_time=start_time,
duration=duration,
unit=uV, # in um, but the calibration factor is considered as m/V
force_virgotools=force_virgotools,
)
return Scatterer(metadata=metadata, sensor=sensor)

130
src/backscattering_experimental_data/utils.py
View file

@ -11,9 +11,9 @@ from gwpy.timeseries import TimeSeries, TimeSeriesDict
try:
from virgotools.frame_lib import FrameFile, FrVect # pyright: ignore[reportMissingImports]
virgotool = True
virgotools = True
except ImportError:
virgotool = False
virgotools = False
pass
@ -51,6 +51,7 @@ def download_timeseries(
duration: Quantity[time],
unit: Unit | None = None,
frame: Literal["raw"] | Literal["trend"] = "raw",
force_virgotools: bool = False,
) -> TimeSeries:
"""
Download a TimeSeries from an online channel
@ -66,37 +67,44 @@ def download_timeseries(
Unit of the TimeSeries (default ot None)
frame : raw|trend
Which frame to take the data from. Default to raw
force_virgotools : bool
Force the use of the virgotools library in place of gwpy first
to be faster. Default to False
Notes
-----
It first try to use gwpy, and fallback to virgotools
"""
try:
return TimeSeries.get(
channel=channel,
start=to_gps(start_time),
end=to_gps(start_time + duration),
)
except HTTPError:
if virgotool:
vect: FrVect | None = None # pyright: ignore[reportPossiblyUnboundVariable]
with FrameFile(frame) as ffl: # pyright: ignore[reportPossiblyUnboundVariable]
vect = ffl.getChannel( # pyright: ignore[reportPossiblyUnboundVariable]
channel, to_gps(start_time), duration.to(s)
)
if vect is None: # pyright: ignore[reportPossiblyUnboundVariable]
raise RuntimeError(
"cannot get data vector from online data"
)
return TimeSeries(
data=vect.data, # pyright: ignore[reportPossiblyUnboundVariable]
unit=unit,
t0=start_time,
sample_rate=vect.fsample * Hz, # pyright: ignore[reportPossiblyUnboundVariable]
if not force_virgotools:
try:
return TimeSeries.get(
channel=channel,
start=to_gps(start_time),
end=to_gps(start_time + duration),
)
else:
except HTTPError:
pass
if virgotools:
vect: FrVect | None = None # pyright: ignore[reportPossiblyUnboundVariable]
with FrameFile(frame) as ffl: # pyright: ignore[reportPossiblyUnboundVariable]
vect = ffl.getChannel( # pyright: ignore[reportPossiblyUnboundVariable]
channel,
to_gps(start_time),
duration.to(s).value,
)
if vect is None: # pyright: ignore[reportPossiblyUnboundVariable]
raise RuntimeError(
"virgotools not available in this environment, cannot download online Virgo data"
"cannot get data vector from online data"
)
return TimeSeries(
data=vect.data, # pyright: ignore[reportPossiblyUnboundVariable]
unit=unit,
t0=start_time,
sample_rate=vect.fsample * Hz, # pyright: ignore[reportPossiblyUnboundVariable]
)
else:
raise RuntimeError(
"virgotools not available in this environment, cannot download online Virgo data"
)
def download_dicttimeseries(
@ -105,6 +113,7 @@ def download_dicttimeseries(
duration: Quantity[time],
units: list[Unit | None] | None = None,
frame: Literal["raw"] | Literal["trend"] = "raw",
force_virgotools: bool = False,
) -> TimeSeriesDict:
"""
Download multiple TimeSeries at the same time to be more efficient
@ -121,39 +130,46 @@ def download_dicttimeseries(
channels. Default to None
frame : raw|trend
Which frame to take the data from. Default to raw
force_virgotools : bool
Force the use of the virgotools library in place of gwpy first
to be faster. Default to False
Notes
-----
It first try to use gwpy, and fallback to virgotools
"""
try:
return TimeSeriesDict.get(
channels=channels,
start=to_gps(start_time),
end=to_gps(start_time + duration),
)
except HTTPError:
if virgotool:
vects: list[FrVect] = [] # pyright: ignore[reportPossiblyUnboundVariable]
with FrameFile(frame) as ffl: # pyright: ignore[reportPossiblyUnboundVariable]
for channel in channels:
vects.append(
ffl.getChannel(
channel, to_gps(start_time), duration.to(s)
)
)
timeseries_dict: dict[str, TimeSeries] = dict()
for i in range(len(vects)):
unit = None
if units is not None:
unit = units[i]
timeseries_dict[channels[i]] = TimeSeries(
data=vects[i].data,
unit=unit,
t0=start_time,
sample_rate=vects[i].fsample * Hz,
)
return TimeSeriesDict(timeseries_dict)
else:
raise RuntimeError(
"virgotools not available in this environment, cannot download online Virgo data"
if not force_virgotools:
try:
return TimeSeriesDict.get(
channels=channels,
start=to_gps(start_time),
end=to_gps(start_time + duration),
)
except HTTPError:
pass
if virgotools:
vects: list[FrVect] = [] # pyright: ignore[reportPossiblyUnboundVariable]
with FrameFile(frame) as ffl: # pyright: ignore[reportPossiblyUnboundVariable]
for channel in channels:
vects.append(
ffl.getChannel(
channel,
to_gps(start_time),
duration.to(s).value,
)
)
timeseries_dict: dict[str, TimeSeries] = dict()
for i in range(len(vects)):
unit = None
if units is not None:
unit = units[i]
timeseries_dict[channels[i]] = TimeSeries(
data=vects[i].data,
unit=unit,
t0=start_time,
sample_rate=vects[i].fsample * Hz,
)
return TimeSeriesDict(timeseries_dict)
else:
raise RuntimeError(
"virgotools not available in this environment, cannot download online Virgo data"
)