tessl/pypi-gwpy
A python package for gravitational-wave astrophysics
—
Pending
segments.mddocs/
Segments and Data Quality
Data quality flag handling and time segment management for identifying valid analysis periods and detector operational states. These tools are essential for gravitational-wave data analysis to ensure only high-quality data is used in scientific analyses.
Capabilities
Segment - Time Interval Representation
Represents a single time interval with start and end times, following the convention [start, end).
class Segment:
def __init__(self, start, end):
"""
Create a time segment.
Parameters:
- start: float, segment start time (GPS)
- end: float, segment end time (GPS)
"""
def __contains__(self, other):
"""
Test if another segment/time is contained within this segment.
Parameters:
- other: Segment, float, or other time-like object
Returns:
bool, True if contained
"""
def intersects(self, other):
"""
Test if this segment intersects another.
Parameters:
- other: Segment, another time segment
Returns:
bool, True if segments intersect
"""
def protract(self, x):
"""
Expand segment by amount x on both sides.
Parameters:
- x: float, amount to expand (seconds)
Returns:
New expanded Segment
"""
def contract(self, x):
"""
Contract segment by amount x on both sides.
Parameters:
- x: float, amount to contract (seconds)
Returns:
New contracted Segment
"""
@property
def start(self):
"""Start time of segment."""
@property
def end(self):
"""End time of segment."""
@property
def duration(self):
"""Duration of segment in seconds."""SegmentList - Collection of Time Segments
List of time segments with set-like operations for combining and manipulating segment collections.
class SegmentList(list):
def __init__(self, segments=None):
"""
Create a list of segments.
Parameters:
- segments: iterable, initial segments
"""
def coalesce(self):
"""
Merge overlapping and adjacent segments.
Returns:
New coalesced SegmentList
"""
def intersects(self, other):
"""
Find intersection with another SegmentList.
Parameters:
- other: SegmentList or Segment
Returns:
SegmentList with intersecting segments
"""
def union(self, other):
"""
Find union with another SegmentList.
Parameters:
- other: SegmentList
Returns:
SegmentList with combined segments
"""
def __sub__(self, other):
"""
Remove segments (set difference).
Parameters:
- other: SegmentList or Segment
Returns:
SegmentList with segments removed
"""
def protract(self, x):
"""
Expand all segments.
Parameters:
- x: float, expansion amount
Returns:
SegmentList with expanded segments
"""
def contract(self, x):
"""
Contract all segments.
Parameters:
- x: float, contraction amount
Returns:
SegmentList with contracted segments
"""
@property
def extent(self):
"""
Total extent from earliest start to latest end.
Returns:
Segment spanning the full extent
"""
@property
def livetime(self):
"""
Total duration of all segments.
Returns:
float, total time in seconds
"""DataQualityFlag - Data Quality Flag Management
Represents a data quality flag with active segments (when flag is set) and valid segments (when data exists).
class DataQualityFlag:
def __init__(self, name=None, active=None, valid=None, **kwargs):
"""
Create a data quality flag.
Parameters:
- name: str, flag name (e.g., 'H1:DMT-ANALYSIS_READY:1')
- active: SegmentList, times when flag is active/True
- valid: SegmentList, times when data is valid/available
"""
@classmethod
def query(cls, flag, start, end, **kwargs):
"""
Query data quality flag from segment database.
Parameters:
- flag: str, flag name to query
- start: float, start time (GPS)
- end: float, end time (GPS)
- url: str, segment server URL
Returns:
DataQualityFlag object
"""
@classmethod
def read(cls, source, flag=None, **kwargs):
"""
Read flag from file.
Parameters:
- source: str, file path
- flag: str, specific flag name
- format: str, file format
Returns:
DataQualityFlag object
"""
def write(self, target, **kwargs):
"""
Write flag to file.
Parameters:
- target: str, output file path
- format: str, output format
"""
def plot(self, **kwargs):
"""
Plot the data quality flag.
Returns:
Plot showing active and valid segments
"""
def __and__(self, other):
"""
Logical AND with another flag.
Parameters:
- other: DataQualityFlag
Returns:
Combined DataQualityFlag
"""
def __or__(self, other):
"""
Logical OR with another flag.
Parameters:
- other: DataQualityFlag
Returns:
Combined DataQualityFlag
"""
def __invert__(self):
"""
Logical NOT (invert flag).
Returns:
Inverted DataQualityFlag
"""
@property
def livetime(self):
"""Total active time in seconds."""
@property
def efficiency(self):
"""Efficiency: active livetime / valid livetime."""DataQualityDict - Collection of Data Quality Flags
Dictionary container for multiple data quality flags with batch operations.
class DataQualityDict(dict):
def __init__(self, *args, **kwargs):
"""
Dictionary of DataQualityFlag objects.
"""
@classmethod
def query(cls, flags, start, end, **kwargs):
"""
Query multiple flags from segment database.
Parameters:
- flags: list, flag names to query
- start: float, start time
- end: float, end time
Returns:
DataQualityDict with all flags
"""
@classmethod
def read(cls, source, **kwargs):
"""
Read multiple flags from file.
Returns:
DataQualityDict with all flags from file
"""
def plot(self, **kwargs):
"""
Plot all flags in a multi-panel figure.
Returns:
Plot with separate panels for each flag
"""
def intersection(self):
"""
Find intersection of all flags.
Returns:
DataQualityFlag representing intersection
"""
def union(self):
"""
Find union of all flags.
Returns:
DataQualityFlag representing union
"""Usage Examples
Basic Segment Operations
from gwpy.segments import Segment, SegmentList
# Create individual segments
seg1 = Segment(1000, 1100) # 100 second segment
seg2 = Segment(1050, 1150) # Overlapping segment
seg3 = Segment(1200, 1300) # Non-overlapping segment
# Create segment list
segments = SegmentList([seg1, seg2, seg3])
# Coalesce overlapping segments
coalesced = segments.coalesce()
print(f"Original: {len(segments)} segments")
print(f"Coalesced: {len(coalesced)} segments")
# Calculate total livetime
total_time = segments.livetime
print(f"Total livetime: {total_time} seconds")
# Find extent
extent = segments.extent
print(f"Data spans from {extent.start} to {extent.end}")Data Quality Flag Analysis
from gwpy.segments import DataQualityFlag
# Query standard analysis-ready flag for LIGO Hanford
start_time = 1126259446
end_time = 1126259478
analysis_ready = DataQualityFlag.query('H1:DMT-ANALYSIS_READY:1',
start=start_time,
end=end_time)
print(f"Analysis ready efficiency: {analysis_ready.efficiency:.2%}")
print(f"Active livetime: {analysis_ready.livetime} seconds")
# Plot the flag
plot = analysis_ready.plot()
plot.set_title('H1 Analysis Ready Flag')
plot.set_xlabel('Time [GPS]')
plot.show()Multi-Flag Analysis
from gwpy.segments import DataQualityDict
# Query multiple data quality flags
flags = ['H1:DMT-ANALYSIS_READY:1',
'H1:DMT-CALIBRATED:1',
'H1:DMT-UP:1',
'H1:LSC-DARM_LOCKED:1']
dq_flags = DataQualityDict.query(flags,
start=start_time,
end=end_time)
# Plot all flags
plot = dq_flags.plot(figsize=(12, 8))
plot.set_title('H1 Data Quality Flags')
plot.show()
# Find intersection (when all flags are active)
science_time = dq_flags.intersection()
print(f"Science-quality livetime: {science_time.livetime} seconds")
# Calculate individual efficiencies
for name, flag in dq_flags.items():
print(f"{name}: {flag.efficiency:.2%} efficient")Segment-Based Data Selection
from gwpy.timeseries import TimeSeries
# Get science-quality segments
science_segments = science_time.active
# Read data only during science time
science_data = []
for segment in science_segments:
if segment.duration >= 64: # Only use segments ≥64s
data = TimeSeries.read('data.gwf', 'H1:STRAIN',
start=segment.start,
end=segment.end)
science_data.append(data)
# Join all science segments
if science_data:
full_science_data = TimeSeries.concatenate(science_data)
print(f"Total science data: {full_science_data.duration} seconds")Custom Flag Creation
# Create custom segments based on analysis criteria
loud_segments = SegmentList()
# Find times when PSD is anomalous
strain = TimeSeries.fetch_open_data('H1', start_time, end_time)
spec = strain.spectrogram(stride=60, fftlength=4)
# Identify loud times (simplified example)
for i, time in enumerate(spec.times):
if spec[i].max() > threshold:
loud_segments.append(Segment(time-30, time+30))
# Create custom data quality flag
loud_flag = DataQualityFlag(name='LOUD_TIMES',
active=loud_segments.coalesce(),
valid=SegmentList([Segment(start_time, end_time)]))
# Remove loud times from science segments
clean_science = science_time - loud_flag
print(f"Clean science time: {clean_science.livetime} seconds")Working with Vetoes
# Query category 1 (hardware) vetoes
cat1_vetoes = DataQualityDict.query(['H1:DMT-ETMY_ESD_DAC_OVERFLOW:1',
'H1:DMT-ETMX_ESD_DAC_OVERFLOW:1'],
start=start_time, end=end_time)
# Apply vetoes to remove bad data
vetoed_science = science_time
for veto_flag in cat1_vetoes.values():
vetoed_science = vetoed_science - veto_flag
print(f"Science time after vetoes: {vetoed_science.livetime} seconds")
print(f"Veto efficiency: {(science_time.livetime - vetoed_science.livetime) / science_time.livetime:.2%}")Segment I/O Operations
# Save segments to file
analysis_ready.write('analysis_ready_segments.xml', format='ligolw')
# Read segments from file
loaded_flag = DataQualityFlag.read('analysis_ready_segments.xml')
# Export to different formats
segments_only = analysis_ready.active
segments_only.write('segments.txt', format='segwizard')
# Read from SegWizard format
segwiz_segments = SegmentList.read('segments.txt', format='segwizard')Install with Tessl CLI
npx tessl i tessl/pypi-gwpy