<commit_before>#!/usr/bin/env python

"""
@package ion.agents.platform.platform_agent_stream_publisher
@file    ion/agents/platform/platform_agent_stream_publisher.py
@author  <NAME>
@brief   Stream publishing support for platform agents.
         NOTE immediate goal is to extract the functionality from the platform
         agent code itself for modularization purposes and
         facilitate alignment with AgentStreamPublisher. For the moment there
         is some replication here of part of the functions there.
"""

__author__ = '<NAME>'
__license__ = 'Apache 2.0'


from pyon.public import log
from pyon.ion.stream import StreamPublisher

from interface.objects import StreamRoute

from ion.agents.platform.platform_driver_event import AttributeValueDriverEvent

from ion.services.dm.utility.granule.record_dictionary import RecordDictionaryTool
import numpy

from coverage_model.parameter import ParameterDictionary

from pyon.core.bootstrap import get_obj_registry
from pyon.core.object import IonObjectDeserializer

import uuid

import logging


class PlatformAgentStreamPublisher(object):
    """
    Stream publishing support for platform agents.
    NOTE immediate goal of this module is to extract the functionality
    from the platform agent code itself. A next goal is to align
    (eventually extending) AgentStreamPublisher.
    """

    def __init__(self, agent):

        assert agent.resource_id, "agent's resource_id must have been set"

        self._agent = agent

        self._platform_id = agent._platform_id
        self.resource_id  = agent.resource_id
        self._pp          = agent._pp
        self.CFG          = agent.CFG

        # Dictionaries used for data publishing.
        self._data_streams = {}
        self._param_dicts = {}
        self._stream_defs = {}
        self._data_publishers = {}

        self._connection_ID = None
        self._connection_index = {}

        # Set of parameter names received in event notification but not
        # configured. Allows to log corresponding warning only once.
        self._unconfigured_params = set()

        stream_info = self.CFG.get('stream_config', None)
        if stream_info is None:
            # should not happen: PlatformAgent._validate_configuration validates this.
            log.error("%r: No stream_config given in CFG", self._platform_id)
            return

        for stream_name, stream_config in stream_info.iteritems():
            self._construct_stream_and_publisher(stream_name, stream_config)

        log.debug("%r: PlatformAgentStreamPublisher complete", self._platform_id)

    def _construct_stream_and_publisher(self, stream_name, stream_config):

        if log.isEnabledFor(logging.TRACE):  # pragma: no cover
            log.trace("%r: _construct_stream_and_publisher: "
                      "stream_name:%r, stream_config:\n%s",
                      self._platform_id, stream_name,
                      self._pp.pformat(stream_config))

        decoder = IonObjectDeserializer(obj_registry=get_obj_registry())

        if 'stream_def_dict' not in stream_config:
            # should not happen: PlatformAgent._validate_configuration validates this.
            log.error("'stream_def_dict' key not in configuration for stream %r" % stream_name)
            return

        stream_def_dict = stream_config['stream_def_dict']
        stream_def_dict['type_'] = 'StreamDefinition'
        stream_def_obj = decoder.deserialize(stream_def_dict)
        self._stream_defs[stream_name] = stream_def_obj

        routing_key           = stream_config['routing_key']
        stream_id             = stream_config['stream_id']
        exchange_point        = stream_config['exchange_point']
        parameter_dictionary  = stream_def_dict['parameter_dictionary']
        log.debug("%r: got parameter_dictionary from stream_def_dict", self._platform_id)

        self._data_streams[stream_name] = stream_id
        self._param_dicts[stream_name] = ParameterDictionary.load(parameter_dictionary)
        stream_route = StreamRoute(exchange_point=exchange_point, routing_key=routing_key)
        publisher = self._create_publisher(stream_id, stream_route)
        self._data_publishers[stream_name] = publisher

        log.debug("%r: created publisher for stream_name=%r", self._platform_id, stream_name)

    def _create_publisher(self, stream_id, stream_route):
        publisher = StreamPublisher(process=self._agent,
                                    stream_id=stream_id,
                                    stream_route=stream_route)
        return publisher

    def reset_connection(self):
        self._connection_ID = uuid.uuid4()
        self._connection_index = {stream_name : 0 for
                                  stream_name in self._data_streams.keys()}
        log.debug("%r: reset_connection: connection_id=%s, connection_index=%s",
                  self._platform_id, self._connection_ID.hex, self._connection_index)

    def handle_attribute_value_event(self, driver_event):

        assert isinstance(driver_event, AttributeValueDriverEvent)

        if log.isEnabledFor(logging.TRACE):  # pragma: no cover
            # show driver_event as retrieved (driver_event.vals_dict might be large)
            log.trace("%r: driver_event = %s", self._platform_id, driver_event)
            log.trace("%r: vals_dict:\n%s",
                      self._platform_id, self._pp.pformat(driver_event.vals_dict))

        elif log.isEnabledFor(logging.DEBUG):  # pragma: no cover
            log.debug("%r: driver_event = %s", self._platform_id, driver_event.brief())

        stream_name = driver_event.stream_name

        publisher = self._data_publishers.get(stream_name, None)
        if not publisher:
            log.warn('%r: no publisher configured for stream_name=%r. '
                     'Configured streams are: %s',
                     self._platform_id, stream_name, self._data_publishers.keys())
            return

        param_dict = self._param_dicts[stream_name]
        stream_def = self._stream_defs[stream_name]

        if isinstance(stream_def, str):
            rdt = RecordDictionaryTool(param_dictionary=param_dict.dump(),
                                       stream_definition_id=stream_def)
        else:
            rdt = RecordDictionaryTool(stream_definition=stream_def)

        self._publish_granule_with_multiple_params(publisher, driver_event,
                                                   param_dict, rdt)

    def _publish_granule_with_multiple_params(self, publisher, driver_event,
                                              param_dict, rdt):

        stream_name = driver_event.stream_name

        pub_params = {}
        selected_timestamps = None

        for param_name, param_value in driver_event.vals_dict.iteritems():

            param_name = param_name.lower()

            if param_name not in rdt:
                if param_name not in self._unconfigured_params:
                    # an unrecognized attribute for this platform:
                    self._unconfigured_params.add(param_name)
                    log.warn('%r: got attribute value event for unconfigured parameter %r in stream %r'
                             ' rdt.keys=%s',
                             self._platform_id, param_name, stream_name, list(rdt.iterkeys()))
                continue

            # Note that notification from the driver has the form
            # of a non-empty list of pairs (val, ts)
            assert isinstance(param_value, list)
            assert isinstance(param_value[0], tuple)

            # separate values and timestamps:
            vals, timestamps = zip(*param_value)

            self._agent._dispatch_value_alerts(stream_name, param_name, vals)
<commit_msg>removed asserts from stream publisher<commit_after>#!/usr/bin/env python

"""
@package ion.agents.platform.platform_agent_stream_publisher
@file    ion/agents/platform/platform_agent_stream_publisher.py
@author  <NAME>
@brief   Stream publishing support for platform agents.
         NOTE immediate goal is to extract the functionality from the platform
         agent code itself for modularization purposes and
         facilitate alignment with AgentStreamPublisher. For the moment there
         is some replication here of part of the functions there.
"""

__author__ = '<NAME>'
__license__ = 'Apache 2.0'


from pyon.public import log
from pyon.ion.stream import StreamPublisher

from interface.objects import StreamRoute

from ion.agents.platform.platform_driver_event import AttributeValueDriverEvent

from ion.services.dm.utility.granule.record_dictionary import RecordDictionaryTool
import numpy

from coverage_model.parameter import ParameterDictionary

from pyon.core.bootstrap import get_obj_registry
from pyon.core.object import IonObjectDeserializer

import uuid

import logging


class PlatformAgentStreamPublisher(object):
    """
    Stream publishing support for platform agents.
    NOTE immediate goal of this module is to extract the functionality
    from the platform agent code itself. A next goal is to align
    (eventually extending) AgentStreamPublisher.
    """

    def __init__(self, agent):
        self._agent = agent

        self._platform_id = agent._platform_id
        self.resource_id  = agent.resource_id
        self._pp          = agent._pp
        self.CFG          = agent.CFG

        # Dictionaries used for data publishing.
        self._data_streams = {}
        self._param_dicts = {}
        self._stream_defs = {}
        self._data_publishers = {}

        self._connection_ID = None
        self._connection_index = {}

        # Set of parameter names received in event notification but not
        # configured. Allows to log corresponding warning only once.
        self._unconfigured_params = set()

        stream_info = self.CFG.get('stream_config', None)
        if stream_info is None:
            # should not happen: PlatformAgent._validate_configuration validates this.
            log.error("%r: No stream_config given in CFG", self._platform_id)
            return

        for stream_name, stream_config in stream_info.iteritems():
            self._construct_stream_and_publisher(stream_name, stream_config)

        log.debug("%r: PlatformAgentStreamPublisher complete", self._platform_id)

    def _construct_stream_and_publisher(self, stream_name, stream_config):

        if log.isEnabledFor(logging.TRACE):  # pragma: no cover
            log.trace("%r: _construct_stream_and_publisher: "
                      "stream_name:%r, stream_config:\n%s",
                      self._platform_id, stream_name,
                      self._pp.pformat(stream_config))

        decoder = IonObjectDeserializer(obj_registry=get_obj_registry())

        if 'stream_def_dict' not in stream_config:
            # should not happen: PlatformAgent._validate_configuration validates this.
            log.error("'stream_def_dict' key not in configuration for stream %r" % stream_name)
            return

        stream_def_dict = stream_config['stream_def_dict']
        stream_def_dict['type_'] = 'StreamDefinition'
        stream_def_obj = decoder.deserialize(stream_def_dict)
        self._stream_defs[stream_name] = stream_def_obj

        routing_key           = stream_config['routing_key']
        stream_id             = stream_config['stream_id']
        exchange_point        = stream_config['exchange_point']
        parameter_dictionary  = stream_def_dict['parameter_dictionary']
        log.debug("%r: got parameter_dictionary from stream_def_dict", self._platform_id)

        self._data_streams[stream_name] = stream_id
        self._param_dicts[stream_name] = ParameterDictionary.load(parameter_dictionary)
        stream_route = StreamRoute(exchange_point=exchange_point, routing_key=routing_key)
        publisher = self._create_publisher(stream_id, stream_route)
        self._data_publishers[stream_name] = publisher

        log.debug("%r: created publisher for stream_name=%r", self._platform_id, stream_name)

    def _create_publisher(self, stream_id, stream_route):
        publisher = StreamPublisher(process=self._agent,
                                    stream_id=stream_id,
                                    stream_route=stream_route)
        return publisher

    def reset_connection(self):
        self._connection_ID = uuid.uuid4()
        self._connection_index = {stream_name : 0 for
                                  stream_name in self._data_streams.keys()}
        log.debug("%r: reset_connection: connection_id=%s, connection_index=%s",
                  self._platform_id, self._connection_ID.hex, self._connection_index)

    def handle_attribute_value_event(self, driver_event):
        if log.isEnabledFor(logging.TRACE):  # pragma: no cover
            # show driver_event as retrieved (driver_event.vals_dict might be large)
            log.trace("%r: driver_event = %s", self._platform_id, driver_event)
            log.trace("%r: vals_dict:\n%s",
                      self._platform_id, self._pp.pformat(driver_event.vals_dict))

        elif log.isEnabledFor(logging.DEBUG):  # pragma: no cover
            log.debug("%r: driver_event = %s", self._platform_id, driver_event.brief())

        stream_name = driver_event.stream_name

        publisher = self._data_publishers.get(stream_name, None)
        if not publisher:
            log.warn('%r: no publisher configured for stream_name=%r. '
                     'Configured streams are: %s',
                     self._platform_id, stream_name, self._data_publishers.keys())
            return

        param_dict = self._param_dicts[stream_name]
        stream_def = self._stream_defs[stream_name]

        if isinstance(stream_def, str):
            rdt = RecordDictionaryTool(param_dictionary=param_dict.dump(),
                                       stream_definition_id=stream_def)
        else:
            rdt = RecordDictionaryTool(stream_definition=stream_def)

        self._publish_granule_with_multiple_params(publisher, driver_event,
                                                   param_dict, rdt)

    def _publish_granule_with_multiple_params(self, publisher, driver_event,
                                              param_dict, rdt):

        stream_name = driver_event.stream_name

        pub_params = {}
        selected_timestamps = None

        for param_name, param_value in driver_event.vals_dict.iteritems():

            param_name = param_name.lower()

            if param_name not in rdt:
                if param_name not in self._unconfigured_params:
                    # an unrecognized attribute for this platform:
                    self._unconfigured_params.add(param_name)
                    log.warn('%r: got attribute value event for unconfigured parameter %r in stream %r'
                             ' rdt.keys=%s',
                             self._platform_id, param_name, stream_name, list(rdt.iterkeys()))
                continue

            # separate values and timestamps:
            vals, timestamps = zip(*param_value)

            self._agent._dispatch_value_alerts(stream_name, param_name, vals)

{'fraction_non_alphanumeric': 0.0744932877072914, 'fraction_numerical': 0.0005922611213477231, 'mean_word_length': 2.549042503503036, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3397586', 'n_tokens_mistral': 4106, 'n_tokens_neox': 3937, 'n_words': 938}