openrvdas.logger.transforms.max_min_transform
No module-level documentation available.
1#!/usr/bin/env python3 2 3import logging 4 5 6from typing import Union 7from logger.utils.das_record import DASRecord # noqa: E402 8from logger.transforms.transform import Transform # noqa: E402 9 10 11################################################################################ 12# 13class MaxMinTransform(Transform): 14 """Transform that returns None unless values in passed DASRecord or 15 dict are greater than/less than the largest/smallest values seen for 16 their respective variables. Otherwise returns dict of colon-suffixed 17 field names that have new max or min values. E.g.: 18 ``` 19 max_min = MaxMinTransform() 20 max_min.transform({'f1': 1, 'f2': 1.5}) -> {'f1:max':1, 'f1:min':1, 21 'f2:max':1.5, 'f2:min':1.5} 22 max_min.transform({'f1': 1, 'f2': 1.5}) -> {} 23 max_min.transform({'f1': 1.1, 'f2': 1.4}) -> {'f1:max':1.1, 'f2:min':1.4,} 24 ``` 25 Note: ignores fields that are not bool, int or float. 26 """ 27 28 def __init__(self, **kwargs): 29 super().__init__(**kwargs) # processes 'quiet' and type hints 30 self.max = {} 31 self.min = {} 32 33 ############################ 34 def transform(self, record: Union[DASRecord, dict]): 35 """Does record exceed any previously-observed bounds?""" 36 # See if it's something we can process, and if not, try digesting 37 if not self.can_process_record(record): # inherited from BaseModule() 38 return self.digest_record(record) # inherited from BaseModule() 39 40 if type(record) is DASRecord: 41 fields = record.fields 42 elif type(record) is dict: 43 fields = record 44 else: 45 logging.warning('Input to MaxMinTransform must be either ' 46 'DASRecord or dict. Received type "%s"', type(record)) 47 return None 48 49 new_limits = {} 50 51 for field, value in fields.items(): 52 # Max and Min only make sense for int, float and bool 53 if not type(value) in [int, float, bool]: 54 continue 55 56 if field not in self.max or value > self.max[field]: 57 self.max[field] = value 58 new_limits[field + ':max'] = value 59 if field not in self.min or value < self.min[field]: 60 self.min[field] = value 61 new_limits[field + ':min'] = value 62 63 if not new_limits: 64 return None 65 66 if type(record) is DASRecord: 67 if record.data_id: 68 data_id = record.data_id + '_limits' if record.data_id else 'limits' 69 return DASRecord(data_id=data_id, 70 message_type=record.message_type, 71 timestamp=record.timestamp, 72 fields=new_limits) 73 74 return new_limits
class
MaxMinTransform(logger.transforms.transform.Transform):
14class MaxMinTransform(Transform): 15 """Transform that returns None unless values in passed DASRecord or 16 dict are greater than/less than the largest/smallest values seen for 17 their respective variables. Otherwise returns dict of colon-suffixed 18 field names that have new max or min values. E.g.: 19 ``` 20 max_min = MaxMinTransform() 21 max_min.transform({'f1': 1, 'f2': 1.5}) -> {'f1:max':1, 'f1:min':1, 22 'f2:max':1.5, 'f2:min':1.5} 23 max_min.transform({'f1': 1, 'f2': 1.5}) -> {} 24 max_min.transform({'f1': 1.1, 'f2': 1.4}) -> {'f1:max':1.1, 'f2:min':1.4,} 25 ``` 26 Note: ignores fields that are not bool, int or float. 27 """ 28 29 def __init__(self, **kwargs): 30 super().__init__(**kwargs) # processes 'quiet' and type hints 31 self.max = {} 32 self.min = {} 33 34 ############################ 35 def transform(self, record: Union[DASRecord, dict]): 36 """Does record exceed any previously-observed bounds?""" 37 # See if it's something we can process, and if not, try digesting 38 if not self.can_process_record(record): # inherited from BaseModule() 39 return self.digest_record(record) # inherited from BaseModule() 40 41 if type(record) is DASRecord: 42 fields = record.fields 43 elif type(record) is dict: 44 fields = record 45 else: 46 logging.warning('Input to MaxMinTransform must be either ' 47 'DASRecord or dict. Received type "%s"', type(record)) 48 return None 49 50 new_limits = {} 51 52 for field, value in fields.items(): 53 # Max and Min only make sense for int, float and bool 54 if not type(value) in [int, float, bool]: 55 continue 56 57 if field not in self.max or value > self.max[field]: 58 self.max[field] = value 59 new_limits[field + ':max'] = value 60 if field not in self.min or value < self.min[field]: 61 self.min[field] = value 62 new_limits[field + ':min'] = value 63 64 if not new_limits: 65 return None 66 67 if type(record) is DASRecord: 68 if record.data_id: 69 data_id = record.data_id + '_limits' if record.data_id else 'limits' 70 return DASRecord(data_id=data_id, 71 message_type=record.message_type, 72 timestamp=record.timestamp, 73 fields=new_limits) 74 75 return new_limits
Transform that returns None unless values in passed DASRecord or dict are greater than/less than the largest/smallest values seen for their respective variables. Otherwise returns dict of colon-suffixed field names that have new max or min values. E.g.:
max_min = MaxMinTransform()
max_min.transform({'f1': 1, 'f2': 1.5}) -> {'f1:max':1, 'f1:min':1,
'f2:max':1.5, 'f2:min':1.5}
max_min.transform({'f1': 1, 'f2': 1.5}) -> {}
max_min.transform({'f1': 1.1, 'f2': 1.4}) -> {'f1:max':1.1, 'f2:min':1.4,}
Note: ignores fields that are not bool, int or float.
MaxMinTransform(**kwargs)
29 def __init__(self, **kwargs): 30 super().__init__(**kwargs) # processes 'quiet' and type hints 31 self.max = {} 32 self.min = {}
quiet - if type checking should log type errors or operate silently.
Two additional arguments govern how records will be encoded/decoded
from bytes, if desired by the Writer subclass when it calls
_encode_str() or _decode_bytes:
encoding - 'utf-8' by default. If empty or None, do not attempt any
decoding and return raw bytes. Other possible encodings are
listed in online documentation here:
https://docs.python.org/3/library/codecs.html#standard-encodings
encoding_errors - 'ignore' by default. Other error strategies are
'strict', 'replace', and 'backslashreplace', described here:
https://docs.python.org/3/howto/unicode.html#encodings
mirror_to - Optional Writer to which all records read or transformed
by this module (if it is a Reader or Transform) will be
"mirrored" (copied). Mirroring happens asynchronously via
a queue and background thread to minimize impact on the
primary data flow. Writers cannot be mirrored.
def
transform(self, record: Union[logger.utils.das_record.DASRecord, dict]):
35 def transform(self, record: Union[DASRecord, dict]): 36 """Does record exceed any previously-observed bounds?""" 37 # See if it's something we can process, and if not, try digesting 38 if not self.can_process_record(record): # inherited from BaseModule() 39 return self.digest_record(record) # inherited from BaseModule() 40 41 if type(record) is DASRecord: 42 fields = record.fields 43 elif type(record) is dict: 44 fields = record 45 else: 46 logging.warning('Input to MaxMinTransform must be either ' 47 'DASRecord or dict. Received type "%s"', type(record)) 48 return None 49 50 new_limits = {} 51 52 for field, value in fields.items(): 53 # Max and Min only make sense for int, float and bool 54 if not type(value) in [int, float, bool]: 55 continue 56 57 if field not in self.max or value > self.max[field]: 58 self.max[field] = value 59 new_limits[field + ':max'] = value 60 if field not in self.min or value < self.min[field]: 61 self.min[field] = value 62 new_limits[field + ':min'] = value 63 64 if not new_limits: 65 return None 66 67 if type(record) is DASRecord: 68 if record.data_id: 69 data_id = record.data_id + '_limits' if record.data_id else 'limits' 70 return DASRecord(data_id=data_id, 71 message_type=record.message_type, 72 timestamp=record.timestamp, 73 fields=new_limits) 74 75 return new_limits
Does record exceed any previously-observed bounds?