openrvdas.logger.transforms.to_das_record_transform
No module-level documentation available.
1#!/usr/bin/env python3 2 3import logging 4 5from typing import Union 6from json import JSONDecodeError 7 8from logger.utils.das_record import DASRecord # noqa: E402 9from logger.transforms.transform import Transform # noqa: E402 10 11 12################################################################################ 13# 14class ToDASRecordTransform(Transform): 15 """Convert passed record to DASRecord. If record is a str, assume a 16 JSON-encoded DASRecord. If record is a dict, use its fields as DASRecord 17 fields. If initialized with a field_name, expect to be passed strings, 18 and use those strings as the corresponding field values. 19 """ 20 21 def __init__(self, data_id=None, field_name=None, **kwargs): 22 super().__init__(**kwargs) # processes 'quiet' and type hints 23 24 self.data_id = data_id 25 self.field_name = field_name 26 27 ############################ 28 def transform(self, record: Union[str, dict]) -> DASRecord: 29 """Convert record to DASRecord.""" 30 31 # See if it's something we can process, and if not, try digesting 32 if not self.can_process_record(record): # inherited from BaseModule() 33 return self.digest_record(record) # inherited from BaseModule() 34 35 if isinstance(record, str): 36 # If str, assume it's JSON unless field_name is set 37 if self.field_name: 38 return DASRecord(data_id=self.data_id, fields={self.field_name: record}) 39 else: 40 try: 41 return DASRecord(json_str=record) 42 except JSONDecodeError: 43 logging.warning(f'String could not be parsed as JSON DASRecord: {record}') 44 return None 45 # Else, if it's a dict, figure out whether it's a simple dict, or has a timestamp, 46 # fields, etc. If not, use keys, values as fields 47 elif isinstance(record, dict): 48 data_id = self.data_id or record.get('data_id') 49 timestamp = record.get('timestamp') 50 fields = record.get('fields') 51 52 # Does it have keys that mark it as a proper DASRecord already? 53 if isinstance(fields, dict): 54 return DASRecord(data_id=data_id, timestamp=timestamp, fields=fields) 55 56 # Otherwise, assume the whole dict is a dict of fields 57 return DASRecord(data_id=self.data_id, fields=record) 58 else: 59 logging.warning('ToDASRecordTransform input should be of type ' 60 f'str or dict, but received {type(record)}: {record}') 61 return None
class
ToDASRecordTransform(logger.transforms.transform.Transform):
15class ToDASRecordTransform(Transform): 16 """Convert passed record to DASRecord. If record is a str, assume a 17 JSON-encoded DASRecord. If record is a dict, use its fields as DASRecord 18 fields. If initialized with a field_name, expect to be passed strings, 19 and use those strings as the corresponding field values. 20 """ 21 22 def __init__(self, data_id=None, field_name=None, **kwargs): 23 super().__init__(**kwargs) # processes 'quiet' and type hints 24 25 self.data_id = data_id 26 self.field_name = field_name 27 28 ############################ 29 def transform(self, record: Union[str, dict]) -> DASRecord: 30 """Convert record to DASRecord.""" 31 32 # See if it's something we can process, and if not, try digesting 33 if not self.can_process_record(record): # inherited from BaseModule() 34 return self.digest_record(record) # inherited from BaseModule() 35 36 if isinstance(record, str): 37 # If str, assume it's JSON unless field_name is set 38 if self.field_name: 39 return DASRecord(data_id=self.data_id, fields={self.field_name: record}) 40 else: 41 try: 42 return DASRecord(json_str=record) 43 except JSONDecodeError: 44 logging.warning(f'String could not be parsed as JSON DASRecord: {record}') 45 return None 46 # Else, if it's a dict, figure out whether it's a simple dict, or has a timestamp, 47 # fields, etc. If not, use keys, values as fields 48 elif isinstance(record, dict): 49 data_id = self.data_id or record.get('data_id') 50 timestamp = record.get('timestamp') 51 fields = record.get('fields') 52 53 # Does it have keys that mark it as a proper DASRecord already? 54 if isinstance(fields, dict): 55 return DASRecord(data_id=data_id, timestamp=timestamp, fields=fields) 56 57 # Otherwise, assume the whole dict is a dict of fields 58 return DASRecord(data_id=self.data_id, fields=record) 59 else: 60 logging.warning('ToDASRecordTransform input should be of type ' 61 f'str or dict, but received {type(record)}: {record}') 62 return None
Convert passed record to DASRecord. If record is a str, assume a JSON-encoded DASRecord. If record is a dict, use its fields as DASRecord fields. If initialized with a field_name, expect to be passed strings, and use those strings as the corresponding field values.
ToDASRecordTransform(data_id=None, field_name=None, **kwargs)
22 def __init__(self, data_id=None, field_name=None, **kwargs): 23 super().__init__(**kwargs) # processes 'quiet' and type hints 24 25 self.data_id = data_id 26 self.field_name = field_name
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[str, dict]) -> logger.utils.das_record.DASRecord:
29 def transform(self, record: Union[str, dict]) -> DASRecord: 30 """Convert record to DASRecord.""" 31 32 # See if it's something we can process, and if not, try digesting 33 if not self.can_process_record(record): # inherited from BaseModule() 34 return self.digest_record(record) # inherited from BaseModule() 35 36 if isinstance(record, str): 37 # If str, assume it's JSON unless field_name is set 38 if self.field_name: 39 return DASRecord(data_id=self.data_id, fields={self.field_name: record}) 40 else: 41 try: 42 return DASRecord(json_str=record) 43 except JSONDecodeError: 44 logging.warning(f'String could not be parsed as JSON DASRecord: {record}') 45 return None 46 # Else, if it's a dict, figure out whether it's a simple dict, or has a timestamp, 47 # fields, etc. If not, use keys, values as fields 48 elif isinstance(record, dict): 49 data_id = self.data_id or record.get('data_id') 50 timestamp = record.get('timestamp') 51 fields = record.get('fields') 52 53 # Does it have keys that mark it as a proper DASRecord already? 54 if isinstance(fields, dict): 55 return DASRecord(data_id=data_id, timestamp=timestamp, fields=fields) 56 57 # Otherwise, assume the whole dict is a dict of fields 58 return DASRecord(data_id=self.data_id, fields=record) 59 else: 60 logging.warning('ToDASRecordTransform input should be of type ' 61 f'str or dict, but received {type(record)}: {record}') 62 return None
Convert record to DASRecord.