# -*- coding: utf-8 -*-
import logging
from astropy.time import Time, TimeDelta
import numpy as np
logger = logging.getLogger(__name__)
def mjd_to_icd(mjd):
"""
Convert MJD seconds into ICD timestamp information
"""
times = Time(mjd / 86400, format='mjd').unix
time_fractions, times = np.modf(times)
times = times.astype('<u4')
time_fractions = (time_fractions * 2 ** 32).astype('<u4')
return times, time_fractions
def icd_to_mjd(times, time_fractions):
"""
Convert ICD timestamp into MJD seconds.
ICD payloads store time in two integer fields: full seconds since the
UNIX epoch, and a fractional second count in 1/2**32 units.
"""
ti = Time(times, format='unix')
tf = TimeDelta(time_fractions / 2 ** 32, format='sec')
mjd = (ti + tf).to_value('mjd')
return mjd * 86400.0 # there might be a precision problem
[docs]class Payload(object):
"""A payload as specified by the ICD"""
def __init__(self):
self._baseline_count = 0
self._channel_count = 0
self._channel_id = 0
self._correlated_data_fraction = []
self._hardware_id = 0
self._phase_bin_id = 0
self._phase_bin_count = 0
self._polarisation_id = 0
self._scan_id = 0
self._time_centroid_indices = []
self._timestamp_count = 0
self._timestamp_fraction = 0
self._visibilities = []
@property
def baseline_count(self):
return self._baseline_count
@baseline_count.setter
def baseline_count(self, baseline_count):
self._baseline_count = baseline_count
@property
def channel_count(self):
"""The number of channels contained in this payload"""
return self._channel_count
@channel_count.setter
def channel_count(self, channel_count):
self._channel_count = channel_count
@property
def channel_id(self):
"""The ID of the first channel of this payload"""
return self._channel_id
@channel_id.setter
def channel_id(self, channel_id):
self._channel_id = channel_id
@property
def correlated_data_fraction(self):
"""The fraction of data on this payload that was correlated"""
return self._correlated_data_fraction
@correlated_data_fraction.setter
def correlated_data_fraction(self, correlated_data_fraction):
self._correlated_data_fraction = correlated_data_fraction
@property
def hardware_id(self):
"""The ID of the hardware source of this payload"""
return self._hardware_id
@hardware_id.setter
def hardware_id(self, hardware_id):
self._hardware_id = hardware_id
@property
def phase_bin_id(self):
"""The ID of the first phase bin of this payload"""
return self._phase_bin_id
@phase_bin_id.setter
def phase_bin_id(self, phase_bin_id):
self._phase_bin_id = phase_bin_id
@property
def phase_bin_count(self):
"""The number of phase bins of this payload"""
return self._phase_bin_count
@phase_bin_count.setter
def phase_bin_count(self, phase_bin_count):
self._phase_bin_count = phase_bin_count
@property
def polarisation_id(self):
"""The ID of the polarisation of this payload"""
return self._polarisation_id
@polarisation_id.setter
def polarisation_id(self, polarisation_id):
self._polarisation_id = polarisation_id
@property
def scan_id(self):
"""The ID of the scan of this payload"""
return self._scan_id
@scan_id.setter
def scan_id(self, scan_id):
self._scan_id = scan_id
@property
def time_centroid_indices(self):
"""The time centroids for each visibility of this payload"""
return self._time_centroid_indices
@time_centroid_indices.setter
def time_centroid_indices(self, time_centroids):
self._time_centroid_indices = time_centroids
@property
def timestamp_count(self):
"""The timestamp of the visibilities, as (integer) seconds since UNIX epoch"""
return self._timestamp_count
@timestamp_count.setter
def timestamp_count(self, timestamp_count):
self._timestamp_count = timestamp_count
@property
def timestamp_fraction(self):
"""The fractional timestamp of the visibilities, as an integer with units
of 1/2**32 seconds"""
return self._timestamp_fraction
@timestamp_fraction.setter
def timestamp_fraction(self, timestamp_fraction):
self._timestamp_fraction = timestamp_fraction
@property
def visibilities(self):
"""The correlator visibilities of this payload"""
return self._visibilities
@visibilities.setter
def visibilities(self, visibilities):
self._visibilities = visibilities
@property
def mjd_time(self) -> float:
"""The timestamp of the payload in MJD seconds"""
return icd_to_mjd(self.timestamp_count, self.timestamp_fraction)
@property
def unix_time(self) -> float:
"""The timestamp as fractional seconds since UNIX epoch"""
return self.timestamp_count + self.timestamp_fraction / 2 ** 32