Detector Utilities

from gwpy.detector import Channel

class Channel:
    def __init__(self, name, sample_rate=None, unit=None, type=None, **kwargs):
        """
        Create a detector channel object.
        
        Parameters:
        - name: str, channel name (e.g., 'H1:LSC-DARM_ERR_DBL_DQ')
        - sample_rate: float, sampling rate in Hz
        - unit: str or Unit, physical unit of the data
        - type: str, channel data type
        """
    
    @classmethod
    def query(cls, name, **kwargs):
        """
        Query channel information from detector databases.
        
        Parameters:
        - name: str, channel name to query
        - **kwargs: additional query parameters
        
        Returns:
        Channel object with metadata
        """
    
    @property
    def name(self):
        """Channel name string."""
    
    @property
    def ifo(self):
        """Interferometer code (e.g., 'H1', 'L1', 'V1')."""
    
    @property
    def system(self):
        """Detector subsystem (e.g., 'LSC', 'ASC', 'CAL')."""
    
    @property
    def subsystem(self):
        """Detector subsystem component."""
    
    @property
    def signal(self):  
        """Signal name within subsystem."""
    
    @property
    def sample_rate(self):
        """Channel sampling rate in Hz."""
    
    @property
    def unit(self):
        """Physical unit of channel data."""
    
    @property
    def type(self):
        """Channel data type."""
    
    def __str__(self):
        """String representation of channel."""
    
    def __repr__(self):
        """Detailed string representation."""

from gwpy.detector import ChannelList

class ChannelList(list):
    def __init__(self, channels=None):
        """
        Create a list of detector channels.
        
        Parameters:
        - channels: iterable, initial Channel objects
        """
    
    @classmethod
    def query(cls, names, **kwargs):
        """
        Query multiple channels from detector databases.
        
        Parameters:
        - names: list, channel names to query
        
        Returns:
        ChannelList with metadata for all channels
        """
    
    def find(self, name):
        """
        Find channel by name.
        
        Parameters:
        - name: str, channel name to find
        
        Returns:
        Channel object or None if not found
        """
    
    def sieve(self, **kwargs):
        """
        Filter channels by criteria.
        
        Parameters:
        - sample_rate: float, filter by sampling rate
        - unit: str, filter by unit
        - ifo: str, filter by interferometer
        
        Returns:
        Filtered ChannelList
        """

from gwpy.detector import get_timezone, get_timezone_offset, TIMEZONE

def get_timezone(ifo):
    """
    Get timezone object for interferometer.
    
    Parameters:
    - ifo: str, interferometer code ('H1', 'L1', 'V1', 'G1', 'C1')
    
    Returns:
    datetime.timezone object for the interferometer location
    
    Raises:
    ValueError: if interferometer code is not recognized
    """

def get_timezone_offset(ifo, dt=None):
    """
    Get timezone offset for interferometer.
    
    Parameters:
    - ifo: str, interferometer code
    - dt: datetime, specific date/time for offset calculation
    
    Returns:
    float, timezone offset in seconds from UTC
    """

# Timezone mapping constant
TIMEZONE = {
    'C1': 'US/Pacific',      # Caltech 40m prototype
    'G1': 'Europe/Berlin',   # GEO600 
    'H1': 'US/Pacific',      # LIGO Hanford
    'L1': 'US/Central',      # LIGO Livingston  
    'V1': 'Europe/Rome'      # Virgo
}

from gwpy.detector import Channel, ChannelList

# Create channel object
strain_channel = Channel('H1:DCS-CALIB_STRAIN_C02')

# Access channel properties
print(f"Interferometer: {strain_channel.ifo}")
print(f"System: {strain_channel.system}")
print(f"Signal: {strain_channel.signal}")

# Query channel metadata from databases
try:
    channel_info = Channel.query('H1:LSC-DARM_ERR_DBL_DQ')
    print(f"Sample rate: {channel_info.sample_rate} Hz")
    print(f"Unit: {channel_info.unit}")
    print(f"Type: {channel_info.type}")
except Exception as e:
    print(f"Could not query channel metadata: {e}")

# Work with channel names
channels = ['H1:LSC-DARM_ERR_DBL_DQ',
           'H1:CAL-DELTAL_EXTERNAL_DQ',
           'H1:ASC-AS_A_RF45_I_ERR_DQ']

channel_list = ChannelList([Channel(name) for name in channels])
print(f"Number of channels: {len(channel_list)}")

# Filter channels by interferometer
h1_channels = channel_list.sieve(ifo='H1')
print(f"H1 channels: {len(h1_channels)}")

# Define channels for multi-detector analysis
detector_channels = {
    'H1': ['H1:DCS-CALIB_STRAIN_C02',
           'H1:LSC-DARM_ERR_DBL_DQ'],
    'L1': ['L1:DCS-CALIB_STRAIN_C02', 
           'L1:LSC-DARM_ERR_DBL_DQ'],
    'V1': ['V1:Hrec_hoft_16384Hz',
           'V1:LSC_DARM_ERR_DQ']
}

# Create channel objects for each detector
all_channels = ChannelList()
for ifo, names in detector_channels.items():
    for name in names:
        all_channels.append(Channel(name))

# Find specific channels
strain_channels = []
for channel in all_channels:
    if 'STRAIN' in channel.name or 'hoft' in channel.name:
        strain_channels.append(channel)

print(f"Found {len(strain_channels)} strain channels")
for ch in strain_channels:
    print(f"  {ch.name}")

from gwpy.detector import get_timezone, get_timezone_offset, TIMEZONE
from datetime import datetime
import pytz

# Show timezone information for all detectors
print("Detector Timezones:")
for ifo, tz_name in TIMEZONE.items():
    timezone = get_timezone(ifo)
    print(f"  {ifo}: {tz_name}")

# Get timezone offsets
reference_time = datetime(2015, 9, 14, 9, 50, 45)  # GW150914 UTC time

print(f"\nTimezone offsets for {reference_time} UTC:")
for ifo in ['H1', 'L1', 'V1', 'G1']:
    offset_sec = get_timezone_offset(ifo, reference_time)
    offset_hours = offset_sec / 3600
    print(f"  {ifo}: {offset_hours:+.1f} hours")

# Convert GPS time to local times
gps_time = 1126259462.4  # GW150914 GPS time
utc_time = datetime.utcfromtimestamp(gps_time - 315964800)  # GPS to UTC

print(f"\nGW150914 local times:")
print(f"  GPS: {gps_time}")
print(f"  UTC: {utc_time}")

for ifo in ['H1', 'L1', 'V1']:
    tz = get_timezone(ifo)
    local_time = utc_time.replace(tzinfo=pytz.UTC).astimezone(tz)
    print(f"  {ifo} local: {local_time.strftime('%Y-%m-%d %H:%M:%S %Z')}")

# Common gravitational-wave analysis channels
analysis_channels = {
    'strain': ['H1:DCS-CALIB_STRAIN_C02', 'L1:DCS-CALIB_STRAIN_C02'],
    'calibration': ['H1:CAL-DELTAL_EXTERNAL_DQ', 'L1:CAL-DELTAL_EXTERNAL_DQ'],
    'darm_error': ['H1:LSC-DARM_ERR_DBL_DQ', 'L1:LSC-DARM_ERR_DBL_DQ'],
    'alignment': ['H1:ASC-AS_A_RF45_I_ERR_DQ', 'L1:ASC-AS_A_RF45_I_ERR_DQ'],
    'environment': ['H1:PEM-BSC5_MIC_SEIS_Z_DQ', 'L1:PEM-EX_MIC_SEIS_Z_DQ']
}

# Create organized channel list
organized_channels = {}
for category, names in analysis_channels.items():
    organized_channels[category] = ChannelList([Channel(name) for name in names])
    
print("Channel Categories:")
for category, channels in organized_channels.items():
    print(f"  {category}: {len(channels)} channels")
    for ch in channels:
        print(f"    {ch.ifo}: {ch.system}-{ch.signal}")

from gwpy.timeseries import TimeSeries, TimeSeriesDict

# Read data using channel objects
start_time = 1126259446
end_time = 1126259478

# Single channel read
strain_ch = Channel('H1:DCS-CALIB_STRAIN_C02')
try:
    strain_data = TimeSeries.get(strain_ch.name, start=start_time, end=end_time)
    print(f"Read {strain_data.duration} seconds of {strain_ch.name}")
except Exception as e:
    print(f"Could not read {strain_ch.name}: {e}")

# Multi-channel read using ChannelList
aux_channels = ChannelList([
    Channel('H1:LSC-DARM_ERR_DBL_DQ'),
    Channel('H1:CAL-DELTAL_EXTERNAL_DQ'),
    Channel('H1:ASC-AS_A_RF45_I_ERR_DQ')
])

# Read all auxiliary channels
aux_data = {}
for channel in aux_channels:
    try:
        data = TimeSeries.get(channel.name, start=start_time, end=end_time)
        aux_data[channel.name] = data
        print(f"Successfully read {channel.name}")
    except Exception as e:
        print(f"Failed to read {channel.name}: {e}")

print(f"Successfully read {len(aux_data)} auxiliary channels")

# Define standard detector configurations
DETECTOR_CONFIG = {
    'LIGO': {
        'sites': ['H1', 'L1'],
        'strain_channels': {
            'H1': 'H1:DCS-CALIB_STRAIN_C02',
            'L1': 'L1:DCS-CALIB_STRAIN_C02'
        },
        'sample_rate': 16384,
        'timezone_mapping': {
            'H1': 'US/Pacific',
            'L1': 'US/Central'
        }
    },
    'Virgo': {
        'sites': ['V1'],
        'strain_channels': {
            'V1': 'V1:Hrec_hoft_16384Hz'
        },
        'sample_rate': 16384,
        'timezone_mapping': {
            'V1': 'Europe/Rome'
        }
    }
}

def get_detector_config(detector_name):
    """Get configuration for a specific detector."""
    return DETECTOR_CONFIG.get(detector_name, {})

def get_all_strain_channels():
    """Get all strain channels across detectors."""
    channels = []
    for config in DETECTOR_CONFIG.values():
        for ifo, channel_name in config['strain_channels'].items():
            channels.append(Channel(channel_name))
    return ChannelList(channels)

# Use configuration
ligo_config = get_detector_config('LIGO')
print(f"LIGO sites: {ligo_config['sites']}")

all_strain = get_all_strain_channels()
print(f"All strain channels: {[ch.name for ch in all_strain]}")

def parse_channel_name(channel_name):
    """Parse channel name into components."""
    try:
        parts = channel_name.split(':')
        if len(parts) != 2:
            raise ValueError("Invalid channel name format")
        
        ifo = parts[0]
        signal_parts = parts[1].split('-')
        
        return {
            'ifo': ifo,
            'system': signal_parts[0] if signal_parts else None,
            'subsystem': signal_parts[1] if len(signal_parts) > 1 else None,
            'signal': '-'.join(signal_parts[2:]) if len(signal_parts) > 2 else None,
            'full_name': channel_name
        }
    except Exception as e:
        print(f"Error parsing channel name '{channel_name}': {e}")
        return None

def validate_channel_names(channel_names):
    """Validate a list of channel names."""
    valid_channels = []
    invalid_channels = []
    
    for name in channel_names:
        parsed = parse_channel_name(name)
        if parsed and parsed['ifo'] in TIMEZONE:
            valid_channels.append(name)
        else:
            invalid_channels.append(name)
    
    return valid_channels, invalid_channels

# Example usage
test_channels = [
    'H1:DCS-CALIB_STRAIN_C02',
    'L1:LSC-DARM_ERR_DBL_DQ', 
    'InvalidChannel',
    'V1:Hrec_hoft_16384Hz',
    'X1:FAKE-CHANNEL'  # X1 is not a real detector
]

valid, invalid = validate_channel_names(test_channels)
print(f"Valid channels: {valid}")
print(f"Invalid channels: {invalid}")

# Parse valid channels
for name in valid:
    info = parse_channel_name(name)
    print(f"{name}:")
    print(f"  IFO: {info['ifo']}")
    print(f"  System: {info['system']}")
    print(f"  Signal: {info['signal']}")