<commit_before># -*- mode: python; coding: utf-8 -*-
# Copyright 2018 the HERA Collaboration
# Licensed under the 2-clause BSD license.

"""Correlator M&C interface

"""
from __future__ import absolute_import, division, print_function

import six
from math import floor
from astropy.time import Time
from sqlalchemy import Column, BigInteger, Integer, Float, Boolean, String, ForeignKeyConstraint

from . import MCDeclarativeBase

DefaultRedisAddress = "redishost"

# key is state type, value is method name in hera_corr_cm
state_dict = {'taking_data': 'is_recording', 'phase_switching': 'phase_switch_is_on',
              'noise_diode': 'noise_diode_is_on'}

tag_list = ['science', 'engineering']

# key is command, value is method name in hera_corr_cm
command_dict = {'take_data': 'take_data', 'stop_taking_data': 'stop_taking_data',
                'phase_switching_on': 'phase_switch_enable',
                'phase_switching_off': 'phase_switch_disable',
                'noise_diode_on': 'noise_diode_enable',
                'noise_diode_off': 'noise_diode_disable',
                'restart': 'restart'}

command_state_map = {'take_data': {'allowed_when_recording': False},
                     'stop_taking_data': {'state_type': 'taking_data', 'state': False,
                                          'allowed_when_recording': True},
                     'phase_switching_on': {'state_type': 'phase_switching',
                                            'state': True,
                                            'allowed_when_recording': False},
                     'phase_switching_off': {'state_type': 'phase_switching',
                                             'state': False,
                                             'allowed_when_recording': False},
                     'noise_diode_on': {'state_type': 'noise_diode', 'state': True,
                                        'allowed_when_recording': False},
                     'noise_diode_off': {'state_type': 'noise_diode', 'state': False,
                                         'allowed_when_recording': False},
                     'restart': {'allowed_when_recording': False}}


class CorrelatorControlState(MCDeclarativeBase):
    """
    Definition of correlator control state table.

    time: gps time of the control state, floored (BigInteger, part of primary_key).
    state_type: type of control state, one of the keys in state_dict (String, part of primary_key)
    state: boolean indicating whether the state_type is true or false
    """
    __tablename__ = 'correlator_control_state'
    time = Column(BigInteger, primary_key=True)
    state_type = Column(String, primary_key=True)
    state = Column(Boolean, nullable=False)

    @classmethod
    def create(cls, time, state_type, state):
        """
        Create a new correlator control state object.

        Parameters:
        ------------
        time: astropy time object
            astropy time object based on a timestamp reported by the correlator
        state_type: string
            must be a key in state_dict (e.g. 'taking_data', 'phase_switching', 'noise_diode')
        state: boolean
            is the state_type true or false
        """
        if not isinstance(time, Time):
            raise ValueError('time must be an astropy Time object')
        corr_time = floor(time.gps)

        if state_type not in list(state_dict.keys()):
            raise ValueError('state_type must be one of: ' + ', '.join(list(state_dict.keys()))
                             + '. state_type is actually {}'.format(state_type))

        return cls(time=corr_time, state_type=state_type, state=state)


def _get_control_state(correlator_redis_address=DefaultRedisAddress):
    """
    Loops through the state types in state_dict and gets the latest state and associated timestamp for each one.

    Returns a 2-level dict, top level key is a key from the state_dict,
    second level keys are 'timestamp' and 'state' (a boolean)
    """
    import hera_corr_cm

    corr_cm = hera_corr_cm.HeraCorrCM(redishost=correlator_redis_address)

    corr_state_dict = {}
    for key, value in enumerate(state_dict):
        # call each state query method and add to corr_state_dict
        timestamp, state = getattr(corr_cm, value)
        corr_state_dict[key]: {'timestamp': timestamp, 'state': state}

    return corr_state_dict


def create_control_state(correlator_redis_address=DefaultRedisAddress, corr_state_dict=None):
    """
    Return a list of correlator control state objects with data from the correlator.

    Parameters:
    ------------
    correlator_redis_address: Address where the correlator redis database can be accessed.
        Only used if corr_state_dict is None.
    corr_state_dict: A dict containing info as in the return dict from _get_control_state()
        for testing purposes. If None, _get_control_state() is called. Default: None

    Returns:
    -----------
    A list of CorrelatorControlState objects
    """

    if corr_state_dict is None:
        corr_state_dict = _get_control_state(correlator_redis_address=correlator_redis_address)

    corr_state_list = []
    for state_type, dict in six.iteritems(corr_state_dict):
        time = Time(dict['timestamp'], format='datetime', scale='utc')
        state = dict['state']

        corr_state_list.append(CorrelatorControlState.create(time, state_type, state))<commit_msg>bug fix<commit_after># -*- mode: python; coding: utf-8 -*-
# Copyright 2018 the HERA Collaboration
# Licensed under the 2-clause BSD license.

"""Correlator M&C interface

"""
from __future__ import absolute_import, division, print_function

import six
from math import floor
from astropy.time import Time
from sqlalchemy import Column, BigInteger, Integer, Float, Boolean, String, ForeignKeyConstraint

from . import MCDeclarativeBase

DefaultRedisAddress = "redishost"

# key is state type, value is method name in hera_corr_cm
state_dict = {'taking_data': 'is_recording', 'phase_switching': 'phase_switch_is_on',
              'noise_diode': 'noise_diode_is_on'}

tag_list = ['science', 'engineering']

# key is command, value is method name in hera_corr_cm
command_dict = {'take_data': 'take_data', 'stop_taking_data': 'stop_taking_data',
                'phase_switching_on': 'phase_switch_enable',
                'phase_switching_off': 'phase_switch_disable',
                'noise_diode_on': 'noise_diode_enable',
                'noise_diode_off': 'noise_diode_disable',
                'restart': 'restart'}

command_state_map = {'take_data': {'allowed_when_recording': False},
                     'stop_taking_data': {'state_type': 'taking_data', 'state': False,
                                          'allowed_when_recording': True},
                     'phase_switching_on': {'state_type': 'phase_switching',
                                            'state': True,
                                            'allowed_when_recording': False},
                     'phase_switching_off': {'state_type': 'phase_switching',
                                             'state': False,
                                             'allowed_when_recording': False},
                     'noise_diode_on': {'state_type': 'noise_diode', 'state': True,
                                        'allowed_when_recording': False},
                     'noise_diode_off': {'state_type': 'noise_diode', 'state': False,
                                         'allowed_when_recording': False},
                     'restart': {'allowed_when_recording': False}}


class CorrelatorControlState(MCDeclarativeBase):
    """
    Definition of correlator control state table.

    time: gps time of the control state, floored (BigInteger, part of primary_key).
    state_type: type of control state, one of the keys in state_dict (String, part of primary_key)
    state: boolean indicating whether the state_type is true or false
    """
    __tablename__ = 'correlator_control_state'
    time = Column(BigInteger, primary_key=True)
    state_type = Column(String, primary_key=True)
    state = Column(Boolean, nullable=False)

    @classmethod
    def create(cls, time, state_type, state):
        """
        Create a new correlator control state object.

        Parameters:
        ------------
        time: astropy time object
            astropy time object based on a timestamp reported by the correlator
        state_type: string
            must be a key in state_dict (e.g. 'taking_data', 'phase_switching', 'noise_diode')
        state: boolean
            is the state_type true or false
        """
        if not isinstance(time, Time):
            raise ValueError('time must be an astropy Time object')
        corr_time = floor(time.gps)

        if state_type not in list(state_dict.keys()):
            raise ValueError('state_type must be one of: ' + ', '.join(list(state_dict.keys()))
                             + '. state_type is actually {}'.format(state_type))

        return cls(time=corr_time, state_type=state_type, state=state)


def _get_control_state(correlator_redis_address=DefaultRedisAddress):
    """
    Loops through the state types in state_dict and gets the latest state and associated timestamp for each one.

    Returns a 2-level dict, top level key is a key from the state_dict,
    second level keys are 'timestamp' and 'state' (a boolean)
    """
    import hera_corr_cm

    corr_cm = hera_corr_cm.HeraCorrCM(redishost=correlator_redis_address)

    corr_state_dict = {}
    for key, value in enumerate(state_dict):
        # call each state query method and add to corr_state_dict
        timestamp, state = getattr(corr_cm, value)
        corr_state_dict[key] = {'timestamp': timestamp, 'state': state}

    return corr_state_dict


def create_control_state(correlator_redis_address=DefaultRedisAddress, corr_state_dict=None):
    """
    Return a list of correlator control state objects with data from the correlator.

    Parameters:
    ------------
    correlator_redis_address: Address where the correlator redis database can be accessed.
        Only used if corr_state_dict is None.
    corr_state_dict: A dict containing info as in the return dict from _get_control_state()
        for testing purposes. If None, _get_control_state() is called. Default: None

    Returns:
    -----------
    A list of CorrelatorControlState objects
    """

    if corr_state_dict is None:
        corr_state_dict = _get_control_state(correlator_redis_address=correlator_redis_address)

    corr_state_list = []
    for state_type, dict in six.iteritems(corr_state_dict):
        time = Time(dict['timestamp'], format='datetime', scale='utc')
        state = dict['state']

        corr_state_list.append(CorrelatorControlState.create(time, state_type, state))
{'fraction_non_alphanumeric': 0.09295019865102097, 'fraction_numerical': 0.0012935415319227572, 'mean_word_length': 2.6407669021190716, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '196600', 'n_tokens_mistral': 3085, 'n_tokens_neox': 2948, 'n_words': 858}