<commit_before>"""Data accessor class for HDF5 files produced by RTS correlator."""

import logging

import numpy as np
import h5py
import katpoint

from .dataset import DataSet, WrongVersion, BrokenFile, Subarray, SpectralWindow, \
                     DEFAULT_SENSOR_PROPS, DEFAULT_VIRTUAL_SENSORS, _robust_target
from .sensordata import SensorData, SensorCache
from .categorical import CategoricalData
from .lazy_indexer import LazyIndexer, LazyTransform

logger = logging.getLogger(__name__)

# Simplify the scan activities to derive the basic state of the antenna (slewing, scanning, tracking, stopped)
SIMPLIFY_STATE = {'scan_ready': 'slew', 'scan': 'scan', 'scan_complete': 'scan', 'track': 'track', 'slew': 'slew'}

SENSOR_PROPS = dict(DEFAULT_SENSOR_PROPS)
SENSOR_PROPS.update({
    '*activity': {'greedy_values': ('slew', 'stop'), 'initial_value': 'slew',
                  'transform': lambda act: SIMPLIFY_STATE.get(act, 'stop')},
    '*target': {'initial_value': '', 'transform': _robust_target},
    '*ap_indexer_position': {'initial_value': ''},
    '*_serial_number': {'initial_value': 0}
})

SENSOR_ALIASES = {
    'nd_coupler': 'dig_noise_diode',
}


def _calc_azel(cache, name, ant):
    """Calculate virtual (az, el) sensors from actual ones in sensor cache."""
    real_sensor = 'Antennas/%s/%s' % (ant, 'pos_actual_scan_azim' if name.endswith('az') else 'pos_actual_scan_elev')
    cache[name] = sensor_data = katpoint.deg2rad(cache.get(real_sensor))
    return sensor_data

VIRTUAL_SENSORS = dict(DEFAULT_VIRTUAL_SENSORS)
VIRTUAL_SENSORS.update({'Antennas/{ant}/az': _calc_azel, 'Antennas/{ant}/el': _calc_azel})

FLAG_NAMES = ('reserved0', 'static', 'cam', 'reserved3', 'detected_rfi', 'predicted_rfi', 'reserved6', 'reserved7')
FLAG_DESCRIPTIONS = ('reserved - bit 0', 'predefined static flag list', 'flag based on live CAM information',
                     'reserved - bit 3', 'RFI detected in the online system', 'RFI predicted from space based pollutants',
                     'reserved - bit 6', 'reserved - bit 7')
WEIGHT_NAMES = ('precision',)
WEIGHT_DESCRIPTIONS = ('visibility precision (inverse variance, i.e. 1 / sigma^2)',)

# Number of bits in ADC sample counter, used to timestamp correlator data in original SPEAD stream
ADC_COUNTER_BITS = 48

# -------------------------------------------------------------------------------------------------
# --- Utility functions
# -------------------------------------------------------------------------------------------------

def dummy_dataset(name, shape, dtype, value):
    """Dummy HDF5 dataset containing a single value.

    This creates a dummy HDF5 dataset in memory containing a single value. It
    can have virtually unlimited size as the dataset is highly compressed.

    Parameters
    ----------
    name : string
        Name of dataset
    shape : sequence of int
        Shape of dataset
    dtype : :class:`numpy.dtype` object or equivalent
        Type of data stored in dataset
    value : object
        All elements in the dataset will equal this value

    Returns
    -------
    dataset : :class:`h5py.Dataset` object
        Dummy HDF5 dataset

    """
    # It is important to randomise the filename as h5py does not allow two writable file objects with the same name
    # Without this randomness katdal can only open one file requiring a dummy dataset
    random_string = ''.join(['%02x' % (x,) for x in np.random.randint(256, size=8)])
    dummy_file = h5py.File('%s_%s.h5' % (name, random_string), driver='core', backing_store=False)
    return dummy_file.create_dataset(name, shape=shape, maxshape=shape,
                                     dtype=dtype, fillvalue=value, compression='gzip')

# -------------------------------------------------------------------------------------------------
# -- CLASS :  H5DataV3
# -------------------------------------------------------------------------------------------------

class H5DataV3(DataSet):
    """Load HDF5 format version 3 file produced by RTS correlator.

    For more information on attributes, see the :class:`DataSet` docstring.

    Parameters
    ----------
    filename : string
        Name of HDF5 file
    ref_ant : string, optional
        Name of reference antenna, used to partition data set into scans
        (default is first antenna in use)
    time_offset : float, optional
        Offset to add to all correlator timestamps, in seconds
    mode : string, optional
        HDF5 file opening mode (e.g. 'r+' to open file in write mode)
    time_scale : float or None, optional
        Resynthesise timestamps using this scale factor
    time_origin : float or None, optional
        Resynthesise timestamps using this sync time / epoch
    rotate_bls : {False, True}, optional
        Rotate baseline label list to work around early RTS correlator bug
    centre_freq : float or None, optional
        Override centre frequency if provided, in Hz
    squeeze : {False, True}, optional
        Don't force vis / weights / flags to be 3-dimensional
    kwargs : dict, optional
        Extra keyword arguments, typically meant for other formats and ignored

    Attributes
    ----------
    file : :class:`h5py.File` object
        Underlying HDF5 file, exposed via :mod:`h5py` interface

    Notes
    -----
    The timestamps can be resynchronised from the original sample counter
    values by specifying *time_scale* and/or *time_origin*. The basic formula
    is given by::

      timestamp = sample_counter / time_scale + time_origin

    """
    def __init__(self, filename, ref_ant='', time_offset=0.0, mode='r',
                 time_scale=None, time_origin=None, rotate_bls=False,
                 centre_freq=None, squeeze=False, **kwargs):
        DataSet.__init__(self, filename, ref_ant, time_offset)

        # Load file
        self.file, self.version = H5DataV3._open(filename, mode)
        f = self.file

        # Load main HDF5 groups
        data_group, tm_group = f['Data'], f['TelescopeModel']
        # Pick first group with appropriate class as CBF
        cbfs = [comp for comp in tm_group
                if tm_group[comp].attrs.get('class') == 'CorrelatorBeamformer']
        cbf_group = tm_group[cbfs[0]]

        # ------ Extract sensors ------

        # Populate sensor cache with all HDF5 datasets below TelescopeModel group that fit the description of a sensor
        cache = {}
        def register_sensor(name, obj):
            """A sensor is defined as a non-empty dataset with expected dtype."""
            if isinstance(obj, h5py.Dataset) and obj.shape != () and obj.dtype.names == ('timestamp', 'value', 'status'):
                comp_name, sensor_name = name.split('/', 1)
                comp_type = tm_group[comp_name].attrs.get('class')
                # Mapping from specific components to generic sensor groups
                # Put antenna sensors in virtual Antenna group, the rest according to component type
                group_lookup = {'AntennaPositioner': 'Antennas/' + comp_name}
                group_name = group_lookup.get(comp_type, comp_type) if comp_type else comp_name
                name = '/'.join((group_name, sensor_name))
                cache[name] = SensorData(obj, name)
        tm_group.visititems(register_sensor)

        # ------ Extract vis and timestamps ------

        if cbf_group.attrs.keys() == ['class']:
            raise BrokenFile('File contains no correlator metadata')
        # Get SDP L0 dump period if available, else fall back to CBF dump period
        self.dump_period = cbf_group.attrs['int_time']
        if 'sdp' in tm_group:
            self.dump_period = tm_group['sdp'].attrs.get('l0_int_time', self.dump_period)
        # Obtain visibilities and timestamps (load the latter explicitly, but obviously not the former...)
        if 'correlator_data' in data_group:
            self._vis = data_group['correlator_data']
        else:
            raise BrokenFile('File contains no visibility data')
        self._timestamps = data_group['timestamps'][:]
        self._squeeze = squeeze

        # Resynthesise timestamps from sample counter based on a different scale factor or origin
        old_scale = cbf_group.attrs['scale_factor_timestamp']
        old_origin = cbf_group.attrs['sync_time']
        # If no new scale factor or origin is given, just use old ones - timestamps should be identical
        time_scale = old_scale if time_scale is None else time_scale
        time_origin = old_origin if time_origin is None else time_origin
        # Work around wraps in ADC sample counter
        adc_wrap_period = 2 ** ADC_COUNTER_BITS / time_scale
        # Get second opinion of the observation start time from regular sensors
        regular_sensors = ('air_temperature', 'air_relative_humidity', 'air_pressure',
                           'pos_actual_scan_azim', 'pos_actual_scan_elev', 'script_log')
        data_duration = self._timestamps[-1] + self.dump_period - self._timestamps[0]
        sensor_start_time = 0.0
        # Pick first regular sensor with longer data record than data (hopefully straddling it)
        for sensor_name, sensor_data in cache.iteritems():
            if sensor_name.endswith(regular_sensors) and len(sensor_data):
                proposed_sensor_start_time = sensor_data[0]['timestamp']
                sensor_duration = sensor_data[-1]['timestamp'] - proposed_sensor_start_time
                if sensor_duration > data_duration:
                    sensor_start_time = proposed_sensor_start_time
                    break
        # If CBF sync time was too long ago, move it forward in steps of wrap period
        while sensor_start_time - time_origin > adc_wrap_period:
            time_origin += adc_wrap_period
        if time_origin != old_origin:
            logger.warning("CBF sync time overridden or moved forward to avoid sample counter wrapping")
            logger.warning("Sync time changed from %s to %s (UTC)" %
                           (katpoint.Timestamp(old_origin), katpoint.Timestamp(time_origin)))
            logger.warning("THE DATA MAY BE CORRUPTED with e.g. delay tracking errors - proceed at own risk!")
        # Resynthesise the timestamps using the final scale and origin
        samples = old_scale * (self._timestamps - old_origin)
        self._timestamps = samples / time_scale + time_origin
        # Now remove any time wraps within the observation
        time_deltas = np.diff(self._timestamps)
        # Assume that a large decrease in timestamp is due to wrapping of ADC sample counter
        time_wraps = np.nonzero(time_deltas < -adc_wrap_period / 2.)[0]
        if time_wraps:
            time_deltas[time_wraps] += adc_wrap_period
            self._timestamps = np.cumsum(np.r_[self._timestamps[0], time_deltas])
            for wrap in time_wraps:
                logger.warning('Time wrap found and corrected at: %s UTC' %
                               (katpoint.Timestamp(self._timestamps[wrap])))
            logger.warning("THE DATA MAY BE CORRUPTED with e.g. delay tracking errors - proceed at own risk!")
        # Warn if there are any remaining decreases in timestamps not associated with wraps
        backward_jumps = np.nonzero(time_deltas < 0.0)[0]
        for jump in backward_jumps:
            logger.warning('CBF timestamps going backward at: %s UTC (%g seconds)' %
                           (katpoint.Timestamp(self._timestamps[jump]), time_deltas[jump]))

        # Check dimensions of timestamps vs those of visibility data
        num_dumps = len(self._timestamps)
        if num_dumps != self._vis.shape[0]:
            raise BrokenFile('Number of timestamps received from ingest '
                             '(%d) differs from number of dumps in data (%d)' % (num_dumps, self._vis.shape[0]))
        # Discard the last sample if the timestamp is a duplicate (caused by stop packet in k7_capture)
        num_dumps = (num_dumps - 1) if num_dumps > 1 and (self._timestamps[-1] == self._timestamps[-2]) else num_dumps
        self._timestamps = self._timestamps[:num_dumps]
        # The expected_dumps should always be an integer (like num_dumps), unless the timestamps and/or dump period
        # are messed up in the file, so the threshold of this test is a bit arbitrary (e.g. could use > 0.5)
        expected_dumps = (self._timestamps[-1] - self._timestamps[0]) / self.dump_period + 1
        if abs(expected_dumps - num_dumps) >= 0.01:
            # Warn the user, as this is anomalous
            logger.warning(("Irregular timestamps detected in file '%s': "
                           "expected %.3f dumps based on dump period and start/end times, got %d instead") %
                           (filename, expected_dumps, num_dumps))
        # Move timestamps from start of each dump to the middle of the dump
        self._timestamps += 0.5 * self.dump_period + self.time_offset
        if self._timestamps[0] < 1e9:
            logger.warning("File '%s' has invalid first correlator timestamp (%f)" % (filename, self._timestamps[0],))
        self._time_keep = np.ones(num_dumps, dtype=np.bool)
        self.start_time = katpoint.Timestamp(self._timestamps[0] - 0.5 * self.dump_period)
        self.end_time = katpoint.Timestamp(self._timestamps[-1] + 0.5 * self.dump_period)
        # Populate sensor cache with all HDF5 datasets below TelescopeModel group that fit the description of a sensor
        self.sensor = SensorCache(cache, self._timestamps, self.dump_period, keep=self._time_keep,
                                  props=SENSOR_PROPS, virtual=VIRTUAL_SENSORS, aliases=SENSOR_ALIASES)

        # ------ Extract flags ------

        # Check if flag group is present, else use dummy flag data
        self._flags = data_group['flags'] if 'flags' in data_group else \
                      dummy_dataset('dummy_flags', shape=self._vis.shape[:-1], dtype=np.uint8, value=0)
        # Obtain flag descriptions from file or recreate default flag description table
        self._flags_description = data_group['flags_description'] if 'flags_description' in data_group else \
                                  np.array(zip(FLAG_NAMES, FLAG_DESCRIPTIONS))

        # ------ Extract weights ------

        # check if weight group present, else use dummy weight data
        self._weights = data_group['weights'] if 'weights' in data_group else \
                        dummy_dataset('dummy_weights', shape=self._vis.shape[:-1] + (1,), dtype=np.float32, value=1.0)
        self._weights_description = np.array(zip(WEIGHT_NAMES, WEIGHT_DESCRIPTIONS))

        # ------ Extract observation parameters ------

        self.obs_params = {}
        # Replay obs_params sensor if available and update obs_params dict accordingly
        try:
            obs_params = self.sensor.get('Observation/params', extract=False)['value']
        except KeyError:
            obs_params = []
        for obs_param in obs_params:
            if obs_param:
                key, val = obs_param.split(' ', 1)
                self.obs_params[key] = np.lib.utils.safe_eval(val)
        # Get observation script parameters, with defaults
        self.observer = self.obs_params.get('observer', '')
        self.description = self.obs_params.get('description', '')
        self.experiment_id = self.obs_params.get('experiment_id', '')

        # ------ Extract subarrays ------

        # All antennas in configuration as katpoint Antenna objects
        ants = []
        for name in tm_group:
            if tm_group[name].attrs.get('class') != 'AntennaPositioner':
                continue
            try:
                ant_description = self.sensor['Antennas/%s/observer' % (name,)][0]
            except KeyError:
                try:
                    ant_description = tm_group[name].attrs['observer']
                except KeyError:
                    ant_description = tm_group[name].attrs['description']
            ants.append(katpoint.Antenna(ant_description))
        cam_ants = set(ant.name for ant in ants)
        # Original list of correlation products as pairs of input labels
        corrprods = cbf_group.attrs['bls_ordering']
        # Find names of all antennas with associated correlator data
        cbf_ants = set([cp[0][:-1] for cp in corrprods] + [cp[1][:-1] for cp in corrprods])
        # By default, only pick antennas that were in use by the script
        obs_ants = self.obs_params.get('ants')
        # Otherwise fall back to the list of antennas common to CAM and CBF
        obs_ants = obs_ants.split(',') if obs_ants else list(cam_ants & cbf_ants)
        self.ref_ant = obs_ants[0] if not ref_ant else ref_ant
        # Populate antenna -> receiver mapping
        for ant in cam_ants:
            band_sensor = 'Antennas/%s/ap_indexer_position' % (ant,)
            band = self.sensor[band_sensor][0] if band_sensor in self.sensor else ''
            rx_sensor = 'Antennas/%s/rsc_rx%s_serial_number' % (ant, band)
            rx_serial = self.sensor[rx_sensor][0] if rx_sensor in self.sensor else 0
            if band:
                self.receivers[ant] = '%s.%d' % (band, rx_serial)
        # Work around early RTS correlator bug by re-ordering labels
        if rotate_bls:
            corrprods = corrprods[range(1, len(corrprods)) + [0]]

        if len(corrprods) != self._vis.shape[2]:<commit_msg>Override receiver band in H5DataV3<commit_after>"""Data accessor class for HDF5 files produced by RTS correlator."""

import logging

import numpy as np
import h5py
import katpoint

from .dataset import DataSet, WrongVersion, BrokenFile, Subarray, SpectralWindow, \
                     DEFAULT_SENSOR_PROPS, DEFAULT_VIRTUAL_SENSORS, _robust_target
from .sensordata import SensorData, SensorCache
from .categorical import CategoricalData
from .lazy_indexer import LazyIndexer, LazyTransform

logger = logging.getLogger(__name__)

# Simplify the scan activities to derive the basic state of the antenna (slewing, scanning, tracking, stopped)
SIMPLIFY_STATE = {'scan_ready': 'slew', 'scan': 'scan', 'scan_complete': 'scan', 'track': 'track', 'slew': 'slew'}

SENSOR_PROPS = dict(DEFAULT_SENSOR_PROPS)
SENSOR_PROPS.update({
    '*activity': {'greedy_values': ('slew', 'stop'), 'initial_value': 'slew',
                  'transform': lambda act: SIMPLIFY_STATE.get(act, 'stop')},
    '*target': {'initial_value': '', 'transform': _robust_target},
    '*ap_indexer_position': {'initial_value': ''},
    '*_serial_number': {'initial_value': 0}
})

SENSOR_ALIASES = {
    'nd_coupler': 'dig_noise_diode',
}


def _calc_azel(cache, name, ant):
    """Calculate virtual (az, el) sensors from actual ones in sensor cache."""
    real_sensor = 'Antennas/%s/%s' % (ant, 'pos_actual_scan_azim' if name.endswith('az') else 'pos_actual_scan_elev')
    cache[name] = sensor_data = katpoint.deg2rad(cache.get(real_sensor))
    return sensor_data

VIRTUAL_SENSORS = dict(DEFAULT_VIRTUAL_SENSORS)
VIRTUAL_SENSORS.update({'Antennas/{ant}/az': _calc_azel, 'Antennas/{ant}/el': _calc_azel})

FLAG_NAMES = ('reserved0', 'static', 'cam', 'reserved3', 'detected_rfi', 'predicted_rfi', 'reserved6', 'reserved7')
FLAG_DESCRIPTIONS = ('reserved - bit 0', 'predefined static flag list', 'flag based on live CAM information',
                     'reserved - bit 3', 'RFI detected in the online system', 'RFI predicted from space based pollutants',
                     'reserved - bit 6', 'reserved - bit 7')
WEIGHT_NAMES = ('precision',)
WEIGHT_DESCRIPTIONS = ('visibility precision (inverse variance, i.e. 1 / sigma^2)',)

# Number of bits in ADC sample counter, used to timestamp correlator data in original SPEAD stream
ADC_COUNTER_BITS = 48

# -------------------------------------------------------------------------------------------------
# --- Utility functions
# -------------------------------------------------------------------------------------------------

def dummy_dataset(name, shape, dtype, value):
    """Dummy HDF5 dataset containing a single value.

    This creates a dummy HDF5 dataset in memory containing a single value. It
    can have virtually unlimited size as the dataset is highly compressed.

    Parameters
    ----------
    name : string
        Name of dataset
    shape : sequence of int
        Shape of dataset
    dtype : :class:`numpy.dtype` object or equivalent
        Type of data stored in dataset
    value : object
        All elements in the dataset will equal this value

    Returns
    -------
    dataset : :class:`h5py.Dataset` object
        Dummy HDF5 dataset

    """
    # It is important to randomise the filename as h5py does not allow two writable file objects with the same name
    # Without this randomness katdal can only open one file requiring a dummy dataset
    random_string = ''.join(['%02x' % (x,) for x in np.random.randint(256, size=8)])
    dummy_file = h5py.File('%s_%s.h5' % (name, random_string), driver='core', backing_store=False)
    return dummy_file.create_dataset(name, shape=shape, maxshape=shape,
                                     dtype=dtype, fillvalue=value, compression='gzip')

# -------------------------------------------------------------------------------------------------
# -- CLASS :  H5DataV3
# -------------------------------------------------------------------------------------------------

class H5DataV3(DataSet):
    """Load HDF5 format version 3 file produced by RTS correlator.

    For more information on attributes, see the :class:`DataSet` docstring.

    Parameters
    ----------
    filename : string
        Name of HDF5 file
    ref_ant : string, optional
        Name of reference antenna, used to partition data set into scans
        (default is first antenna in use)
    time_offset : float, optional
        Offset to add to all correlator timestamps, in seconds
    mode : string, optional
        HDF5 file opening mode (e.g. 'r+' to open file in write mode)
    time_scale : float or None, optional
        Resynthesise timestamps using this scale factor
    time_origin : float or None, optional
        Resynthesise timestamps using this sync time / epoch
    rotate_bls : {False, True}, optional
        Rotate baseline label list to work around early RTS correlator bug
    centre_freq : float or None, optional
        Override centre frequency if provided, in Hz
    band : string or None, optional
        Override receiver band if provided (e.g. 'l') - used to find ND models
    squeeze : {False, True}, optional
        Don't force vis / weights / flags to be 3-dimensional
    kwargs : dict, optional
        Extra keyword arguments, typically meant for other formats and ignored

    Attributes
    ----------
    file : :class:`h5py.File` object
        Underlying HDF5 file, exposed via :mod:`h5py` interface

    Notes
    -----
    The timestamps can be resynchronised from the original sample counter
    values by specifying *time_scale* and/or *time_origin*. The basic formula
    is given by::

      timestamp = sample_counter / time_scale + time_origin

    """
    def __init__(self, filename, ref_ant='', time_offset=0.0, mode='r',
                 time_scale=None, time_origin=None, rotate_bls=False,
                 centre_freq=None, band=None, squeeze=False, **kwargs):
        DataSet.__init__(self, filename, ref_ant, time_offset)

        # Load file
        self.file, self.version = H5DataV3._open(filename, mode)
        f = self.file

        # Load main HDF5 groups
        data_group, tm_group = f['Data'], f['TelescopeModel']
        # Pick first group with appropriate class as CBF
        cbfs = [comp for comp in tm_group
                if tm_group[comp].attrs.get('class') == 'CorrelatorBeamformer']
        cbf_group = tm_group[cbfs[0]]

        # ------ Extract sensors ------

        # Populate sensor cache with all HDF5 datasets below TelescopeModel group that fit the description of a sensor
        cache = {}
        def register_sensor(name, obj):
            """A sensor is defined as a non-empty dataset with expected dtype."""
            if isinstance(obj, h5py.Dataset) and obj.shape != () and obj.dtype.names == ('timestamp', 'value', 'status'):
                comp_name, sensor_name = name.split('/', 1)
                comp_type = tm_group[comp_name].attrs.get('class')
                # Mapping from specific components to generic sensor groups
                # Put antenna sensors in virtual Antenna group, the rest according to component type
                group_lookup = {'AntennaPositioner': 'Antennas/' + comp_name}
                group_name = group_lookup.get(comp_type, comp_type) if comp_type else comp_name
                name = '/'.join((group_name, sensor_name))
                cache[name] = SensorData(obj, name)
        tm_group.visititems(register_sensor)

        # ------ Extract vis and timestamps ------

        if cbf_group.attrs.keys() == ['class']:
            raise BrokenFile('File contains no correlator metadata')
        # Get SDP L0 dump period if available, else fall back to CBF dump period
        self.dump_period = cbf_group.attrs['int_time']
        if 'sdp' in tm_group:
            self.dump_period = tm_group['sdp'].attrs.get('l0_int_time', self.dump_period)
        # Obtain visibilities and timestamps (load the latter explicitly, but obviously not the former...)
        if 'correlator_data' in data_group:
            self._vis = data_group['correlator_data']
        else:
            raise BrokenFile('File contains no visibility data')
        self._timestamps = data_group['timestamps'][:]
        self._squeeze = squeeze

        # Resynthesise timestamps from sample counter based on a different scale factor or origin
        old_scale = cbf_group.attrs['scale_factor_timestamp']
        old_origin = cbf_group.attrs['sync_time']
        # If no new scale factor or origin is given, just use old ones - timestamps should be identical
        time_scale = old_scale if time_scale is None else time_scale
        time_origin = old_origin if time_origin is None else time_origin
        # Work around wraps in ADC sample counter
        adc_wrap_period = 2 ** ADC_COUNTER_BITS / time_scale
        # Get second opinion of the observation start time from regular sensors
        regular_sensors = ('air_temperature', 'air_relative_humidity', 'air_pressure',
                           'pos_actual_scan_azim', 'pos_actual_scan_elev', 'script_log')
        data_duration = self._timestamps[-1] + self.dump_period - self._timestamps[0]
        sensor_start_time = 0.0
        # Pick first regular sensor with longer data record than data (hopefully straddling it)
        for sensor_name, sensor_data in cache.iteritems():
            if sensor_name.endswith(regular_sensors) and len(sensor_data):
                proposed_sensor_start_time = sensor_data[0]['timestamp']
                sensor_duration = sensor_data[-1]['timestamp'] - proposed_sensor_start_time
                if sensor_duration > data_duration:
                    sensor_start_time = proposed_sensor_start_time
                    break
        # If CBF sync time was too long ago, move it forward in steps of wrap period
        while sensor_start_time - time_origin > adc_wrap_period:
            time_origin += adc_wrap_period
        if time_origin != old_origin:
            logger.warning("CBF sync time overridden or moved forward to avoid sample counter wrapping")
            logger.warning("Sync time changed from %s to %s (UTC)" %
                           (katpoint.Timestamp(old_origin), katpoint.Timestamp(time_origin)))
            logger.warning("THE DATA MAY BE CORRUPTED with e.g. delay tracking errors - proceed at own risk!")
        # Resynthesise the timestamps using the final scale and origin
        samples = old_scale * (self._timestamps - old_origin)
        self._timestamps = samples / time_scale + time_origin
        # Now remove any time wraps within the observation
        time_deltas = np.diff(self._timestamps)
        # Assume that a large decrease in timestamp is due to wrapping of ADC sample counter
        time_wraps = np.nonzero(time_deltas < -adc_wrap_period / 2.)[0]
        if time_wraps:
            time_deltas[time_wraps] += adc_wrap_period
            self._timestamps = np.cumsum(np.r_[self._timestamps[0], time_deltas])
            for wrap in time_wraps:
                logger.warning('Time wrap found and corrected at: %s UTC' %
                               (katpoint.Timestamp(self._timestamps[wrap])))
            logger.warning("THE DATA MAY BE CORRUPTED with e.g. delay tracking errors - proceed at own risk!")
        # Warn if there are any remaining decreases in timestamps not associated with wraps
        backward_jumps = np.nonzero(time_deltas < 0.0)[0]
        for jump in backward_jumps:
            logger.warning('CBF timestamps going backward at: %s UTC (%g seconds)' %
                           (katpoint.Timestamp(self._timestamps[jump]), time_deltas[jump]))

        # Check dimensions of timestamps vs those of visibility data
        num_dumps = len(self._timestamps)
        if num_dumps != self._vis.shape[0]:
            raise BrokenFile('Number of timestamps received from ingest '
                             '(%d) differs from number of dumps in data (%d)' % (num_dumps, self._vis.shape[0]))
        # Discard the last sample if the timestamp is a duplicate (caused by stop packet in k7_capture)
        num_dumps = (num_dumps - 1) if num_dumps > 1 and (self._timestamps[-1] == self._timestamps[-2]) else num_dumps
        self._timestamps = self._timestamps[:num_dumps]
        # The expected_dumps should always be an integer (like num_dumps), unless the timestamps and/or dump period
        # are messed up in the file, so the threshold of this test is a bit arbitrary (e.g. could use > 0.5)
        expected_dumps = (self._timestamps[-1] - self._timestamps[0]) / self.dump_period + 1
        if abs(expected_dumps - num_dumps) >= 0.01:
            # Warn the user, as this is anomalous
            logger.warning(("Irregular timestamps detected in file '%s': "
                           "expected %.3f dumps based on dump period and start/end times, got %d instead") %
                           (filename, expected_dumps, num_dumps))
        # Move timestamps from start of each dump to the middle of the dump
        self._timestamps += 0.5 * self.dump_period + self.time_offset
        if self._timestamps[0] < 1e9:
            logger.warning("File '%s' has invalid first correlator timestamp (%f)" % (filename, self._timestamps[0],))
        self._time_keep = np.ones(num_dumps, dtype=np.bool)
        self.start_time = katpoint.Timestamp(self._timestamps[0] - 0.5 * self.dump_period)
        self.end_time = katpoint.Timestamp(self._timestamps[-1] + 0.5 * self.dump_period)
        # Populate sensor cache with all HDF5 datasets below TelescopeModel group that fit the description of a sensor
        self.sensor = SensorCache(cache, self._timestamps, self.dump_period, keep=self._time_keep,
                                  props=SENSOR_PROPS, virtual=VIRTUAL_SENSORS, aliases=SENSOR_ALIASES)

        # ------ Extract flags ------

        # Check if flag group is present, else use dummy flag data
        self._flags = data_group['flags'] if 'flags' in data_group else \
                      dummy_dataset('dummy_flags', shape=self._vis.shape[:-1], dtype=np.uint8, value=0)
        # Obtain flag descriptions from file or recreate default flag description table
        self._flags_description = data_group['flags_description'] if 'flags_description' in data_group else \
                                  np.array(zip(FLAG_NAMES, FLAG_DESCRIPTIONS))

        # ------ Extract weights ------

        # check if weight group present, else use dummy weight data
        self._weights = data_group['weights'] if 'weights' in data_group else \
                        dummy_dataset('dummy_weights', shape=self._vis.shape[:-1] + (1,), dtype=np.float32, value=1.0)
        self._weights_description = np.array(zip(WEIGHT_NAMES, WEIGHT_DESCRIPTIONS))

        # ------ Extract observation parameters ------

        self.obs_params = {}
        # Replay obs_params sensor if available and update obs_params dict accordingly
        try:
            obs_params = self.sensor.get('Observation/params', extract=False)['value']
        except KeyError:
            obs_params = []
        for obs_param in obs_params:
            if obs_param:
                key, val = obs_param.split(' ', 1)
                self.obs_params[key] = np.lib.utils.safe_eval(val)
        # Get observation script parameters, with defaults
        self.observer = self.obs_params.get('observer', '')
        self.description = self.obs_params.get('description', '')
        self.experiment_id = self.obs_params.get('experiment_id', '')

        # ------ Extract subarrays ------

        # All antennas in configuration as katpoint Antenna objects
        ants = []
        for name in tm_group:
            if tm_group[name].attrs.get('class') != 'AntennaPositioner':
                continue
            try:
                ant_description = self.sensor['Antennas/%s/observer' % (name,)][0]
            except KeyError:
                try:
                    ant_description = tm_group[name].attrs['observer']
                except KeyError:
                    ant_description = tm_group[name].attrs['description']
            ants.append(katpoint.Antenna(ant_description))
        cam_ants = set(ant.name for ant in ants)
        # Original list of correlation products as pairs of input labels
        corrprods = cbf_group.attrs['bls_ordering']
        # Find names of all antennas with associated correlator data
        cbf_ants = set([cp[0][:-1] for cp in corrprods] + [cp[1][:-1] for cp in corrprods])
        # By default, only pick antennas that were in use by the script
        obs_ants = self.obs_params.get('ants')
        # Otherwise fall back to the list of antennas common to CAM and CBF
        obs_ants = obs_ants.split(',') if obs_ants else list(cam_ants & cbf_ants)
        self.ref_ant = obs_ants[0] if not ref_ant else ref_ant
        # Populate antenna -> receiver mapping
        band_override = band
        for ant in cam_ants:
            if band_override is None:
                band_sensor = 'Antennas/%s/ap_indexer_position' % (ant,)
                band = self.sensor[band_sensor][0] if band_sensor in self.sensor else ''
            rx_sensor = 'Antennas/%s/rsc_rx%s_serial_number' % (ant, band)
            rx_serial = self.sensor[rx_sensor][0] if rx_sensor in self.sensor else 0
            if band:
                self.receivers[ant] = '%s.%d' % (band, rx_serial)
        # Work around early RTS correlator bug by re-ordering labels
        if rotate_bls:
            corrprods = corrprods[range(1, len(corrprods)) + [0]]

        if len(corrprods) != self._vis.shape[2]:
{'fraction_non_alphanumeric': 0.1164795509222133, 'fraction_numerical': 0.006701798602359949, 'mean_word_length': 2.4269310040239476, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '305925', 'n_tokens_mistral': 10045, 'n_tokens_neox': 9646, 'n_words': 3043}