Source code for otestpoint.labtools.transform

#
# Copyright (c) 2016-2017 - Adjacent Link LLC, Bridgewater, New Jersey
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions
# are met:
#
#  * Redistributions of source code must retain the above copyright
#    notice, this list of conditions and the following disclaimer.
#  * Redistributions in binary form must reproduce the above copyright
#    notice, this list of conditions and the following disclaimer in
#    the documentation and/or other materials provided with the
#    distribution.
#  * Neither the name of Adjacent Link LLC nor the names of its
#    contributors may be used to endorse or promote products derived
#    from this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
# FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
# COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
# BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#

"""Produce a transformed labtools variable instance using a callable.

"""

from subject import Subject
from observer import Observer
import numpy as np
import pandas as pd
import traceback

class Transform(Subject,Observer):
    """
    Transform subject data using a callable.
    """
    def __init__(self,variable,transform,**kwargs):
        """Creates a Transform instance.

        Transform instances work on subjects with single value state
        or list of values state:

          val
          [val_0,val_1,...val_N]


        Args:

          variable (str) - Vairable instance to transform.

          transform (callable) - Callable python object.

        Kwargs:

        name (str): Subject name. Default:
        Transform(transform(variable)).

        Raises:

          KeyError

        """
        Subject.__init__(self,variable.stream())
        Observer.__init__(self)
        self._transform = transform
        self._cache = {}
        self._tags = set()
        self._updated = set()

        self._name = kwargs.pop('name',None)

        if self._name == None:
            name = ""

            try:
                name = self._transform.__name__
            except:
                name =  self._transform.__class__.__name__

            self._name = "Transform(%s(%s))" % (name,variable.name())

        if kwargs:
            raise KeyError("Unknown key(s): %s" % ", ".join(kwargs.keys()))

        variable.attach(self)

    def name(self):
        """Gets the name of this subject.

        Returns:

          Subject name (str).
        """
        return self._name

    def notify(self,subject):
        """Processes subject update to produce new state."""
        data,updated = subject.state()

        if not data and not updated:
            self._cache.clear()
            self._tags.clear()
            self._updated.clear()
            self.notify_observers()
            return

        # remove values in cache and previous data no longer available
        # in subject
        for purge in set(self._cache.keys()).difference(data.keys()):
            del self._cache[purge]

        # update tags and existing cache with nan for any new tag entries
        for tags in data.values():
            for tag in tags:
                if tag not in self._tags:

                    # update existing cache/previous with np.nan subject values
                    for timestamp in self._cache:
                        self._cache[timestamp][tag] = np.nan

            # update tags
            self._tags.update(set(tags.keys()))

        # update cache to include np.nan data for new timestamps
        for timestamp in updated:
            if timestamp not in self._cache:
                self._cache[timestamp] = {}

                for tag in self._tags:
                    if tag not in self._cache[timestamp]:
                        self._cache[timestamp][tag] = np.nan

        for timestamp in updated:
            for tag in data[timestamp]:
                try:
                    self._cache[timestamp][tag] = self._transform(data[timestamp][tag])
                except:
                    print traceback.format_exc()
                    self._cache[timestamp][tag] = np.nan

        self._updated = updated
        self.notify_observers()
        self._updated = set()

    def state(self):
        """Gets the state.

        Returns:

          A dict of with timestamp (int) keys and dict values. Where
          each dict value has tag name (str) keys and the transformed
          value. Depending on the subject, that can be a single value,
          a list of values.

        """
        return self._cache,self._updated