LastSensor.get_feature_value() - Code Metrics - Inspection of "CASAS: fuel: Functions to get activity and feature..." - TinghuiWang/pyActLearn - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d47136...8eb192 )

by Tinghui

created 2016-11-29 03:12 UTC

LastSensor.get_feature_value() B

↳ Parent: LastSensor

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
dl	0
loc	28
rs	8.0894
c	0
b	0
f	0

import abc
import logging
import numpy as np
from ..logging import logging_name

logger = logging.getLogger(__file__)


# region Abstract FeatureRoutineTemplate Class
class FeatureRoutineTemplate(metaclass=abc.ABCMeta):
    """Feature Routine Class

    A routine that calculate statistical features every time the window slides.

    Attributes:
        name (:obj:`str`): Feature routine name.
        description (:obj:`str`): Feature routine description.
        enabled (:obj:`str`): Feature routine enable flag.
    """
    def __init__(self, name, description, enabled=True):
        """
        Initialization of Template Class
        :return:
        """
        # Name
        self.name = name
        # Description
        self.description = description
        # enable
        self.enabled = enabled

    @abc.abstractmethod
    def update(self, data_list, cur_index, window_size, sensor_info):
        """Abstract update method

        For some features, we will update some statistical data every time
        we move forward a data record, instead of going back through the whole
        window and try to find the answer. This function will be called every time
        we advance in data record.

        Args:
            data_list (:obj:`list`): List of sensor data.
            cur_index (:obj:`int`): Index of current data record.
            window_size (:obj:`int`): Sliding window size.
            sensor_info (:obj:`dict`): Dictionary containing sensor index information.
        """
        return NotImplementedError()

    @abc.abstractmethod
    def clear(self):
        """Clear Internal Data Structures if recalculation is needed
        """
        return NotImplementedError()
# endregion


# region Abstract FeatureTemplate Class
class FeatureTemplate(metaclass=abc.ABCMeta):
    """Statistical Feature Template

    Args:
        name (:obj:`str`): Feature name.
        description (:obj:`str`): Feature description.
        per_sensor (:obj:`bool`): If the feature is calculated for each sensor.
        enabled (:obj:`bool`): If the feature is enabled.
        routine (:obj:`.FeatureRoutineTemplate`): Routine structure.
        normalized (:obj:`bool`): If the value of feature needs to be normalized.

    Attributes:
        name (:obj:`str`): Feature name.
        description (:obj:`str`): Feature description.
        index (:obj:`int`): Feature index.
        normalized (:obj:`bool`): If the value of feature needs to be normalized.
        per_sensor (:obj:`bool`): If the feature is calculated for each sensor.
        enabled (:obj:`bool`): If the feature is enabled.
        routine (:obj:`.FeatureRoutineTemplate`): Routine structure.
        _is_value_valid (:obj:`bool`): If the value calculated is valid
    """
    def __init__(self, name, description, enabled=True, normalized=True, per_sensor=False, routine=None):
        self.name = name
        self.description = description
        self.index = -1
        self.normalized = normalized
        self.per_sensor = per_sensor
        self.enabled = enabled
        self._is_value_valid = False
        # update Routine
        # For some feature, we will update statistical data every time we move forward
        # a data record. Instead of going back through previous window, the update function
        # in this routine structure will be called each time we advance to next data record
        self.routine = routine

    @abc.abstractmethod
    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """Abstract method to get feature value

        Args:
            data_list (:obj:`list`): List of sensor data.
            cur_index (:obj:`int`): Index of current data record.
            window_size (:obj:`int`): Sliding window size.
            sensor_info (:obj:`dict`): Dictionary containing sensor index information.
            sensor_name (:obj:`str`): Sensor Name.

        Returns:
            :obj:`double`: feature value
        """
        return NotImplementedError()

    @property
    def is_value_valid(self):
        """Statistical feature value valid check

        Due to errors and failures of sensors, the statistical feature calculated
        may go out of bound. This abstract method is used to check if the value
        calculated is valid. If not, it will not be inserted into feature vectors.

        Returns:
            :obj:`bool`: True if the result is valid.
        """
        return self._is_value_valid
# endregion


class EventHour(FeatureTemplate):
    """Show the hour of the time of current event
    """
    def __init__(self, normalized=False):
        super().__init__(name='lastEventHour',
                         description='Time of the last sensor event in window (hour)',
                         normalized=normalized,
                         per_sensor=False,
                         enabled=True,
                         routine=None)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """Get the hour when the last sensor event in the window occurred

        Note:
            Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
            parameters.
        """
        self._is_value_valid = True
        if self.normalized:
            return np.float(data_list[cur_index]['datetime'].hour)/24
        else:
            return np.float(data_list[cur_index]['datetime'].hour)


class EventSeconds(FeatureTemplate):
    """Feature that shows the time (min, sec) of current event in seconds
    """
    def __init__(self, normalized=False):
        super().__init__(
            name='lastEventSeconds',
            description='Time of the last sensor event in window in seconds',
            normalized=normalized,
            per_sensor=False,
            enabled=True,
            routine=None)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """Get the time within an hour when the last sensor event in the window occurred (in seconds)

        Note:
            Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
            parameters.
        """
        self._is_value_valid = True
        time = data_list[cur_index]['datetime']
        if self.normalized:
            return np.float((time.minute * 60) + time.second)/3600
        else:
            return np.float((time.minute * 60) + time.second)


class WindowDuration(FeatureTemplate):
    """Length of the window in seconds
    """
    def __init__(self, normalized=False):
        super().__init__(name='windowDuration',
                         description='Duration of current window in seconds',
                         normalized=normalized,
                         per_sensor=False,
                         enabled=True,
                         routine=None)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """Get the duration of the window in seconds. Invalid if the duration is greater than half a day.

        Note:
            Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
            parameters.
        """
        self._is_value_valid = True
        timedelta = data_list[cur_index]['datetime'] - data_list[cur_index - window_size + 1]['datetime']
        window_duration = timedelta.total_seconds()
        if window_duration > 3600 * 12:
            self._is_value_valid = False
            # Window Duration is greater than a day - not possible
            # print('Warning: curIndex: %d; windowSize: %d; windowDuration: %f' %
            # (curIndex, windowSize, window_duration))
            window_duration -= 3600 * 12 * (int(window_duration) / (3600 * 12))
            # print('Fixed window duration %f' % window_duration)
            if data_list[cur_index]['datetime'].month != data_list[cur_index - 1]['datetime'].month or \
                    data_list[cur_index]['datetime'].day != data_list[cur_index - 1]['datetime'].day:
                date_advanced = (data_list[cur_index]['datetime'] - data_list[cur_index - 1]['datetime']).days
                hour_advanced = data_list[cur_index]['datetime'].hour - data_list[cur_index - 1]['datetime'].hour
                logger.warn(logging_name(self) + ': line %d - %d: %s' %
                            (cur_index, cur_index + 1, data_list[cur_index - 1]['datetime'].isoformat()))
                logger.warn(logging_name(self) + ': Date Advanced: %d; hour gap: %d' % (date_advanced, hour_advanced))
        if self.normalized:
            # Normalized to 12 hours
            return np.float(window_duration) / (3600 * 12)
        else:
            return np.float(window_duration)


class LastSensor(FeatureTemplate):
    """Get the last sensor in the window
    """
    def __init__(self, per_sensor=False):
        super().__init__(name='lastSensorInWindow',
                         description='Sensor ID in the current window',
                         per_sensor=per_sensor,
                         enabled=True,
                         routine=None)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """Get the sensor which fired the last event in the sliding window.

        If it is configured as per-sensor feature, it returns 1 if the sensor specified
        triggers the last event in the window. Otherwise returns 0.
        If it is configured as a non-per-sensor feature, it returns the index of the
        index corresponding to the dominant sensor name that triggered the last event.

        Note:
            Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
            parameters.
        """
        self._is_value_valid = True
        sensor_label = data_list[cur_index]['sensor_id']
        if self.per_sensor:
            if sensor_name is not None:
                if sensor_name == sensor_label:
                    return 1
                else:
                    return 0
        else:
            if sensor_info.get(sensor_label, None) is None:
                self._is_value_valid = False
                logger.warn(logging_name(self) + ': Cannot find sensor %s in sensor_info' % sensor_label)
                logger.debug(logging_name(self) + ': Available sensors are: ' + str(sensor_info.keys()))
                return 0
            else:
                return sensor_info[sensor_label]['index']


class SensorCountRoutine(FeatureRoutineTemplate):
    """Routine to count occurance of each sensor

    Attributes:
        sensor_count (:obj:`dict`): Dictionary that counts the occurrance of each sensor
    """
    def __init__(self):
        super().__init__(
            name='SensorCountRoutine',
            description='Count Occurrence of all sensors in current event window',
            enabled=True
        )
        # Dominant Sensor
        self.sensor_count = {}

    def update(self, data_list, cur_index, window_size, sensor_info):
        """Record the number of occurrence of each sensor in the sensor count dictionary.
        """
        self.sensor_count = {}
        for sensor_label in sensor_info.keys():
            if sensor_info[sensor_label]['enable']:
                self.sensor_count[sensor_label] = 0
        for index in range(0, window_size):
            if data_list[cur_index - index]['sensor_id'] in self.sensor_count.keys():
                self.sensor_count[data_list[cur_index - index]['sensor_id']] += 1

    def clear(self):
        self.sensor_count = {}

sensor_count_routine = SensorCountRoutine()


class SensorCount(FeatureTemplate):
    """Counts the occurrence of each sensor
    """
    def __init__(self, normalized=False):
        super().__init__(name='sensorCount',
                         description='Number of Events in the window related to the sensor',
                         normalized=normalized,
                         per_sensor=True,
                         enabled=True,
                         routine=sensor_count_routine)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """Counts the number of occurrence of the sensor specified in current window.
        """
        count = self.routine.sensor_count.get(sensor_name, None)
        if count is None:
            logger.error(logging_name(self) + ': Cannot find sensor %s in sensor list' % sensor_name)
            self._is_value_valid = False
        else:
            self._is_value_valid = True
            if self.normalized:
                return float(count)/(window_size * 2)
            else:
                return float(count)


class SensorElapseTimeRoutine(FeatureRoutineTemplate):
    """Routine to record last occurrence of each sensor

    Attributes:
        sensor_fire_log (:obj:`dict`): Dictionary that record the last firing state of each sensor
    """
    def __init__(self):
        super().__init__(name='SensorElapseTimeUpdateRoutine',
                         description='Update Sensor Elapse Time for all enabled sensors',
                         enabled=True)
        # Sensor Fire Log
        self.sensor_fire_log = {}

    def update(self, data_list, cur_index, window_size, sensor_info):
        """Record the number of occurrence of each sensor in the sensor count dictionary.
        """
        if not self.sensor_fire_log:
            for sensor_label in sensor_info.keys():
                self.sensor_fire_log[sensor_label] = data_list[cur_index - window_size + 1]['datetime']
            for i in range(0, window_size):
                self.sensor_fire_log[data_list[cur_index - i]['sensor_id']] = data_list[cur_index - i]['datetime']
        self.sensor_fire_log[data_list[cur_index]['sensor_id']] = data_list[cur_index]['datetime']

    def clear(self):
        self.sensor_fire_log = {}

sensor_elapse_time_routine = SensorElapseTimeRoutine()


class SensorElapseTime(FeatureTemplate):
    """The time elapsed since last firing (in seconds)
    """
    def __init__(self, normalized=False):
        super().__init__(name='sensorElapseTime',
                         description='Time since each sensor fired (in seconds)',
                         normalized=normalized,
                         per_sensor=True,
                         enabled=True,
                         routine=sensor_elapse_time_routine)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """Get elapse time of specified sensor in seconds
        """
        self._is_value_valid = True
        timedelta = data_list[cur_index]['datetime'] - self.routine.sensor_fire_log[sensor_name]
        sensor_duration = timedelta.total_seconds()
        if self.normalized:
            elapse_time = float(sensor_duration)/(12*3600)
            # If the sensor is not fired in past 12 hours, just round it up to 12 hours
            if elapse_time > 1:
                elapse_time = 1.
            return elapse_time
        else:
            return float(sensor_duration)


class DominantSensorRoutine(FeatureRoutineTemplate):
    """Routine to record the occurance of each sensor within the sliding window

    Attributes:
        dominant_sensor_list (:obj:`dict`): Dictionary that record the last firing state of each sensor
    """
    def __init__(self):
        super().__init__(name='DominantSensorRoutine',
                         description='DominantSensorUpdateRoutine',
                         enabled=True)
        # Dominant Sensor
        self.dominant_sensor_list = {}

    def update(self, data_list, cur_index, window_size, sensor_info):
        """Calculate the dominant sensor of current window and store
        the name of the sensor in the dominant sensor array. The
        information is fetched by dominant sensor features.
        """
        if cur_index < window_size:
            logger.warn(logging_name(self) + ': current index %d is smaller than window size %d.' % (cur_index, window_size))
        sensor_count = {}
        for index in range(0, window_size):
            if data_list[cur_index - index]['sensor_id'] in sensor_count.keys():
                sensor_count[data_list[cur_index - index]['sensor_id']] += 1
            else:
                sensor_count[data_list[cur_index - index]['sensor_id']] = 1
        # Find the Dominant one
        max_count = 0
        for sensor_label in sensor_count.keys():
            if sensor_count[sensor_label] > max_count:
                max_count = sensor_count[sensor_label]
                self.dominant_sensor_list[cur_index] = sensor_label

    def clear(self):
        self.dominant_sensor_list = {}

dominant_sensor_routine = DominantSensorRoutine()


class DominantSensor(FeatureTemplate):

    """Dominant Sensor of current window
    """
    def __init__(self, per_sensor=False):
        super().__init__(name='DominantSensor',
                         description='Dominant Sensor in the window',
                         normalized=True,
                         per_sensor=per_sensor,
                         enabled=True,
                         routine=dominant_sensor_routine)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """If per_sensor is True, returns 1 with corresponding sensor Id.
        otherwise, return the index of last sensor in the window
        """
        self._is_value_valid = True
        dominant_sensor_label = self.routine.dominant_sensor_list.get(cur_index, None)
        if dominant_sensor_label is None:
            logger.warn(logging_name(self) + ': cannot find dominant sensor label for window index %d' % cur_index)
        if self.per_sensor:
            if sensor_name is not None:
                if sensor_name == dominant_sensor_label:
                    return 1
                else:
                    return 0
        else:
            return sensor_info[dominant_sensor_label]['index']


class DominantSensorPreviousWindow(FeatureTemplate):

    """Dominant Sensor of previous window
    """
    def __init__(self, per_sensor=False):
        super().__init__(name='DominantSensorPreviousWindow',
                         description='Dominant Sensor in the previous window',
                         normalized=True,
                         per_sensor=per_sensor,
                         enabled=True,
                         routine=dominant_sensor_routine)

    def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
        """If per_sensor is True, returns 1 with corresponding sensor Id.
        otherwise, return the index of last sensor in the window
        """
        dominant_sensor_label = self.routine.dominant_sensor_list.get([cur_index-1], None)
        if dominant_sensor_label is None:
            logger.warn(logging_name(self) + ': cannot find dominant sensor label for window index %d' % cur_index)
        if self.per_sensor:
            if sensor_name is not None:
                if sensor_name == dominant_sensor_label:
                    return 1
                else:
                    return 0
        else:
            return sensor_info[dominant_sensor_label]['index']


1		import abc
2		import logging
3		import numpy as np
4		from ..logging import logging_name
5
6		logger = logging.getLogger(__file__)
7
8
9		# region Abstract FeatureRoutineTemplate Class
10		class FeatureRoutineTemplate(metaclass=abc.ABCMeta):
11		"""Feature Routine Class
12
13		A routine that calculate statistical features every time the window slides.
14
15		Attributes:
16		name (:obj:`str`): Feature routine name.
17		description (:obj:`str`): Feature routine description.
18		enabled (:obj:`str`): Feature routine enable flag.
19		"""
20		def __init__(self, name, description, enabled=True):
21		"""
22		Initialization of Template Class
23		:return:
24		"""
25		# Name
26		self.name = name
27		# Description
28		self.description = description
29		# enable
30		self.enabled = enabled
31
32		@abc.abstractmethod
33		def update(self, data_list, cur_index, window_size, sensor_info):
34		"""Abstract update method
35
36		For some features, we will update some statistical data every time
37		we move forward a data record, instead of going back through the whole
38		window and try to find the answer. This function will be called every time
39		we advance in data record.
40
41		Args:
42		data_list (:obj:`list`): List of sensor data.
43		cur_index (:obj:`int`): Index of current data record.
44		window_size (:obj:`int`): Sliding window size.
45		sensor_info (:obj:`dict`): Dictionary containing sensor index information.
46		"""
47		return NotImplementedError()
48
49		@abc.abstractmethod
50		def clear(self):
51		"""Clear Internal Data Structures if recalculation is needed
52		"""
53		return NotImplementedError()
54		# endregion
55
56
57		# region Abstract FeatureTemplate Class
58		class FeatureTemplate(metaclass=abc.ABCMeta):
59		"""Statistical Feature Template
60
61		Args:
62		name (:obj:`str`): Feature name.
63		description (:obj:`str`): Feature description.
64		per_sensor (:obj:`bool`): If the feature is calculated for each sensor.
65		enabled (:obj:`bool`): If the feature is enabled.
66		routine (:obj:`.FeatureRoutineTemplate`): Routine structure.
67		normalized (:obj:`bool`): If the value of feature needs to be normalized.
68
69		Attributes:
70		name (:obj:`str`): Feature name.
71		description (:obj:`str`): Feature description.
72		index (:obj:`int`): Feature index.
73		normalized (:obj:`bool`): If the value of feature needs to be normalized.
74		per_sensor (:obj:`bool`): If the feature is calculated for each sensor.
75		enabled (:obj:`bool`): If the feature is enabled.
76		routine (:obj:`.FeatureRoutineTemplate`): Routine structure.
77		_is_value_valid (:obj:`bool`): If the value calculated is valid
78		"""
79		def __init__(self, name, description, enabled=True, normalized=True, per_sensor=False, routine=None):
80		self.name = name
81		self.description = description
82		self.index = -1
83		self.normalized = normalized
84		self.per_sensor = per_sensor
85		self.enabled = enabled
86		self._is_value_valid = False
87		# update Routine
88		# For some feature, we will update statistical data every time we move forward
89		# a data record. Instead of going back through previous window, the update function
90		# in this routine structure will be called each time we advance to next data record
91		self.routine = routine
92
93		@abc.abstractmethod
94		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
95		"""Abstract method to get feature value
96
97		Args:
98		data_list (:obj:`list`): List of sensor data.
99		cur_index (:obj:`int`): Index of current data record.
100		window_size (:obj:`int`): Sliding window size.
101		sensor_info (:obj:`dict`): Dictionary containing sensor index information.
102		sensor_name (:obj:`str`): Sensor Name.
103
104		Returns:
105		:obj:`double`: feature value
106		"""
107		return NotImplementedError()
108
109		@property
110		def is_value_valid(self):
111		"""Statistical feature value valid check
112
113		Due to errors and failures of sensors, the statistical feature calculated
114		may go out of bound. This abstract method is used to check if the value
115		calculated is valid. If not, it will not be inserted into feature vectors.
116
117		Returns:
118		:obj:`bool`: True if the result is valid.
119		"""
120		return self._is_value_valid
121		# endregion
122
123
124		class EventHour(FeatureTemplate):
125		"""Show the hour of the time of current event
126		"""
127		def __init__(self, normalized=False):
128		super().__init__(name='lastEventHour',
129		description='Time of the last sensor event in window (hour)',
130		normalized=normalized,
131		per_sensor=False,
132		enabled=True,
133		routine=None)
134
135		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
136		"""Get the hour when the last sensor event in the window occurred
137
138		Note:
139		Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
140		parameters.
141		"""
142		self._is_value_valid = True
143		if self.normalized:
144		return np.float(data_list[cur_index]['datetime'].hour)/24
145		else:
146		return np.float(data_list[cur_index]['datetime'].hour)
147
148
149		class EventSeconds(FeatureTemplate):
150		"""Feature that shows the time (min, sec) of current event in seconds
151		"""
152		def __init__(self, normalized=False):
153		super().__init__(
154		name='lastEventSeconds',
155		description='Time of the last sensor event in window in seconds',
156		normalized=normalized,
157		per_sensor=False,
158		enabled=True,
159		routine=None)
160
161		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
162		"""Get the time within an hour when the last sensor event in the window occurred (in seconds)
163
164		Note:
165		Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
166		parameters.
167		"""
168		self._is_value_valid = True
169		time = data_list[cur_index]['datetime']
170		if self.normalized:
171		return np.float((time.minute * 60) + time.second)/3600
172		else:
173		return np.float((time.minute * 60) + time.second)
174
175
176		class WindowDuration(FeatureTemplate):
177		"""Length of the window in seconds
178		"""
179		def __init__(self, normalized=False):
180		super().__init__(name='windowDuration',
181		description='Duration of current window in seconds',
182		normalized=normalized,
183		per_sensor=False,
184		enabled=True,
185		routine=None)
186
187		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
188		"""Get the duration of the window in seconds. Invalid if the duration is greater than half a day.
189
190		Note:
191		Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
192		parameters.
193		"""
194		self._is_value_valid = True
195		timedelta = data_list[cur_index]['datetime'] - data_list[cur_index - window_size + 1]['datetime']
196		window_duration = timedelta.total_seconds()
197		if window_duration > 3600 * 12:
198		self._is_value_valid = False
199		# Window Duration is greater than a day - not possible
200		# print('Warning: curIndex: %d; windowSize: %d; windowDuration: %f' %
201		# (curIndex, windowSize, window_duration))
202		window_duration -= 3600 * 12 * (int(window_duration) / (3600 * 12))
203		# print('Fixed window duration %f' % window_duration)
204		if data_list[cur_index]['datetime'].month != data_list[cur_index - 1]['datetime'].month or \
205		data_list[cur_index]['datetime'].day != data_list[cur_index - 1]['datetime'].day:
206		date_advanced = (data_list[cur_index]['datetime'] - data_list[cur_index - 1]['datetime']).days
207		hour_advanced = data_list[cur_index]['datetime'].hour - data_list[cur_index - 1]['datetime'].hour
208		logger.warn(logging_name(self) + ': line %d - %d: %s' %
209		(cur_index, cur_index + 1, data_list[cur_index - 1]['datetime'].isoformat()))
210		logger.warn(logging_name(self) + ': Date Advanced: %d; hour gap: %d' % (date_advanced, hour_advanced))
211		if self.normalized:
212		# Normalized to 12 hours
213		return np.float(window_duration) / (3600 * 12)
214		else:
215		return np.float(window_duration)
216
217
218		class LastSensor(FeatureTemplate):
219		"""Get the last sensor in the window
220		"""
221		def __init__(self, per_sensor=False):
222		super().__init__(name='lastSensorInWindow',
223		description='Sensor ID in the current window',
224		per_sensor=per_sensor,
225		enabled=True,
226		routine=None)
227
228		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
229		"""Get the sensor which fired the last event in the sliding window.
230
231		If it is configured as per-sensor feature, it returns 1 if the sensor specified
232		triggers the last event in the window. Otherwise returns 0.
233		If it is configured as a non-per-sensor feature, it returns the index of the
234		index corresponding to the dominant sensor name that triggered the last event.
235
236		Note:
237		Please refer to :meth:`~.FeatureTemplate.get_feature_value` for information about
238		parameters.
239		"""
240		self._is_value_valid = True
241		sensor_label = data_list[cur_index]['sensor_id']
242		if self.per_sensor:
243		if sensor_name is not None:
244		if sensor_name == sensor_label:
245		return 1
246		else:
247		return 0
248		else:
249		if sensor_info.get(sensor_label, None) is None:
250		self._is_value_valid = False
251		logger.warn(logging_name(self) + ': Cannot find sensor %s in sensor_info' % sensor_label)
252		logger.debug(logging_name(self) + ': Available sensors are: ' + str(sensor_info.keys()))
253		return 0
254		else:
255		return sensor_info[sensor_label]['index']
256
257
258		class SensorCountRoutine(FeatureRoutineTemplate):
259		"""Routine to count occurance of each sensor
260
261		Attributes:
262		sensor_count (:obj:`dict`): Dictionary that counts the occurrance of each sensor
263		"""
264		def __init__(self):
265		super().__init__(
266		name='SensorCountRoutine',
267		description='Count Occurrence of all sensors in current event window',
268		enabled=True
269		)
270		# Dominant Sensor
271		self.sensor_count = {}
272
273		def update(self, data_list, cur_index, window_size, sensor_info):
274		"""Record the number of occurrence of each sensor in the sensor count dictionary.
275		"""
276		self.sensor_count = {}
277		for sensor_label in sensor_info.keys():
278		if sensor_info[sensor_label]['enable']:
279		self.sensor_count[sensor_label] = 0
280		for index in range(0, window_size):
281		if data_list[cur_index - index]['sensor_id'] in self.sensor_count.keys():
282		self.sensor_count[data_list[cur_index - index]['sensor_id']] += 1
283
284		def clear(self):
285		self.sensor_count = {}
286
287		sensor_count_routine = SensorCountRoutine()
288
289
290		class SensorCount(FeatureTemplate):
291		"""Counts the occurrence of each sensor
292		"""
293		def __init__(self, normalized=False):
294		super().__init__(name='sensorCount',
295		description='Number of Events in the window related to the sensor',
296		normalized=normalized,
297		per_sensor=True,
298		enabled=True,
299		routine=sensor_count_routine)
300
301		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
302		"""Counts the number of occurrence of the sensor specified in current window.
303		"""
304		count = self.routine.sensor_count.get(sensor_name, None)
305		if count is None:
306		logger.error(logging_name(self) + ': Cannot find sensor %s in sensor list' % sensor_name)
307		self._is_value_valid = False
308		else:
309		self._is_value_valid = True
310		if self.normalized:
311		return float(count)/(window_size * 2)
312		else:
313		return float(count)
314
315
316		class SensorElapseTimeRoutine(FeatureRoutineTemplate):
317		"""Routine to record last occurrence of each sensor
318
319		Attributes:
320		sensor_fire_log (:obj:`dict`): Dictionary that record the last firing state of each sensor
321		"""
322		def __init__(self):
323		super().__init__(name='SensorElapseTimeUpdateRoutine',
324		description='Update Sensor Elapse Time for all enabled sensors',
325		enabled=True)
326		# Sensor Fire Log
327		self.sensor_fire_log = {}
328
329		def update(self, data_list, cur_index, window_size, sensor_info):
330		"""Record the number of occurrence of each sensor in the sensor count dictionary.
331		"""
332		if not self.sensor_fire_log:
333		for sensor_label in sensor_info.keys():
334		self.sensor_fire_log[sensor_label] = data_list[cur_index - window_size + 1]['datetime']
335		for i in range(0, window_size):
336		self.sensor_fire_log[data_list[cur_index - i]['sensor_id']] = data_list[cur_index - i]['datetime']
337		self.sensor_fire_log[data_list[cur_index]['sensor_id']] = data_list[cur_index]['datetime']
338
339		def clear(self):
340		self.sensor_fire_log = {}
341
342		sensor_elapse_time_routine = SensorElapseTimeRoutine()
343
344
345		class SensorElapseTime(FeatureTemplate):
346		"""The time elapsed since last firing (in seconds)
347		"""
348		def __init__(self, normalized=False):
349		super().__init__(name='sensorElapseTime',
350		description='Time since each sensor fired (in seconds)',
351		normalized=normalized,
352		per_sensor=True,
353		enabled=True,
354		routine=sensor_elapse_time_routine)
355
356		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
357		"""Get elapse time of specified sensor in seconds
358		"""
359		self._is_value_valid = True
360		timedelta = data_list[cur_index]['datetime'] - self.routine.sensor_fire_log[sensor_name]
361		sensor_duration = timedelta.total_seconds()
362		if self.normalized:
363		elapse_time = float(sensor_duration)/(12*3600)
364		# If the sensor is not fired in past 12 hours, just round it up to 12 hours
365		if elapse_time > 1:
366		elapse_time = 1.
367		return elapse_time
368		else:
369		return float(sensor_duration)
370
371
372		class DominantSensorRoutine(FeatureRoutineTemplate):
373		"""Routine to record the occurance of each sensor within the sliding window
374
375		Attributes:
376		dominant_sensor_list (:obj:`dict`): Dictionary that record the last firing state of each sensor
377		"""
378		def __init__(self):
379		super().__init__(name='DominantSensorRoutine',
380		description='DominantSensorUpdateRoutine',
381		enabled=True)
382		# Dominant Sensor
383		self.dominant_sensor_list = {}
384
385		def update(self, data_list, cur_index, window_size, sensor_info):
386		"""Calculate the dominant sensor of current window and store
387		the name of the sensor in the dominant sensor array. The
388		information is fetched by dominant sensor features.
389		"""
390		if cur_index < window_size:
391		logger.warn(logging_name(self) + ': current index %d is smaller than window size %d.' % (cur_index, window_size))
392		sensor_count = {}
393		for index in range(0, window_size):
394		if data_list[cur_index - index]['sensor_id'] in sensor_count.keys():
395		sensor_count[data_list[cur_index - index]['sensor_id']] += 1
396		else:
397		sensor_count[data_list[cur_index - index]['sensor_id']] = 1
398		# Find the Dominant one
399		max_count = 0
400		for sensor_label in sensor_count.keys():
401		if sensor_count[sensor_label] > max_count:
402		max_count = sensor_count[sensor_label]
403		self.dominant_sensor_list[cur_index] = sensor_label
404
405		def clear(self):
406		self.dominant_sensor_list = {}
407
408		dominant_sensor_routine = DominantSensorRoutine()
409
410
411	View Code Duplication	class DominantSensor(FeatureTemplate):
		0 ignored issues – show Duplication introduced 2016-11-29 03:12 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
412		"""Dominant Sensor of current window
413		"""
414		def __init__(self, per_sensor=False):
415		super().__init__(name='DominantSensor',
416		description='Dominant Sensor in the window',
417		normalized=True,
418		per_sensor=per_sensor,
419		enabled=True,
420		routine=dominant_sensor_routine)
421
422		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
423		"""If per_sensor is True, returns 1 with corresponding sensor Id.
424		otherwise, return the index of last sensor in the window
425		"""
426		self._is_value_valid = True
427		dominant_sensor_label = self.routine.dominant_sensor_list.get(cur_index, None)
428		if dominant_sensor_label is None:
429		logger.warn(logging_name(self) + ': cannot find dominant sensor label for window index %d' % cur_index)
430		if self.per_sensor:
431		if sensor_name is not None:
432		if sensor_name == dominant_sensor_label:
433		return 1
434		else:
435		return 0
436		else:
437		return sensor_info[dominant_sensor_label]['index']
438
439
440	View Code Duplication	class DominantSensorPreviousWindow(FeatureTemplate):
		0 ignored issues – show Duplication introduced 2016-11-29 03:12 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
441		"""Dominant Sensor of previous window
442		"""
443		def __init__(self, per_sensor=False):
444		super().__init__(name='DominantSensorPreviousWindow',
445		description='Dominant Sensor in the previous window',
446		normalized=True,
447		per_sensor=per_sensor,
448		enabled=True,
449		routine=dominant_sensor_routine)
450
451		def get_feature_value(self, data_list, cur_index, window_size, sensor_info, sensor_name=None):
452		"""If per_sensor is True, returns 1 with corresponding sensor Id.
453		otherwise, return the index of last sensor in the window
454		"""
455		dominant_sensor_label = self.routine.dominant_sensor_list.get([cur_index-1], None)
456		if dominant_sensor_label is None:
457		logger.warn(logging_name(self) + ': cannot find dominant sensor label for window index %d' % cur_index)
458		if self.per_sensor:
459		if sensor_name is not None:
460		if sensor_name == dominant_sensor_label:
461		return 1
462		else:
463		return 0
464		else:
465		return sensor_info[dominant_sensor_label]['index']
466

TinghuiWang / pyActLearn

Push — master ( d47136...8eb192 )

LastSensor.get_feature_value() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like