openrvdas.logger.transforms.to_json_transform

No module-level documentation available.
 1#!/usr/bin/env python3
 2
 3import json
 4import logging
 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 ToJSONTransform(Transform):
14    """Convert passed DASRecords, lists or dicts to JSON. If pretty == True,
15    format the JSON output for easy reading.
16    """
17
18    ############################
19    def __init__(self, pretty=False, **kwargs):
20        super().__init__(**kwargs)  # processes 'quiet' and type hints
21        self.pretty = pretty
22
23    ############################
24    def transform(self, record: Union[DASRecord, float, int, bool, str, dict, list, set]) -> str:
25        """Convert record to JSON."""
26
27        # See if it's something we can process, and if not, try digesting
28        if not self.can_process_record(record):  # inherited from BaseModule()
29            return self.digest_record(record)  # inherited from BaseModule()
30
31        if type(record) is DASRecord:
32            return record.as_json(self.pretty)
33
34        if type(record) in [float, int, bool, str, dict, list, set]:
35            if self.pretty:
36                return json.dumps(record, sort_keys=True, indent=4)
37            else:
38                return json.dumps(record)
39
40        logging.warning('ToJSON transform received record format it could not '
41                        'serialize: "%s"', type(record))
42        return None
class ToJSONTransform(logger.transforms.transform.Transform): 
14class ToJSONTransform(Transform):
15    """Convert passed DASRecords, lists or dicts to JSON. If pretty == True,
16    format the JSON output for easy reading.
17    """
18
19    ############################
20    def __init__(self, pretty=False, **kwargs):
21        super().__init__(**kwargs)  # processes 'quiet' and type hints
22        self.pretty = pretty
23
24    ############################
25    def transform(self, record: Union[DASRecord, float, int, bool, str, dict, list, set]) -> str:
26        """Convert record to JSON."""
27
28        # See if it's something we can process, and if not, try digesting
29        if not self.can_process_record(record):  # inherited from BaseModule()
30            return self.digest_record(record)  # inherited from BaseModule()
31
32        if type(record) is DASRecord:
33            return record.as_json(self.pretty)
34
35        if type(record) in [float, int, bool, str, dict, list, set]:
36            if self.pretty:
37                return json.dumps(record, sort_keys=True, indent=4)
38            else:
39                return json.dumps(record)
40
41        logging.warning('ToJSON transform received record format it could not '
42                        'serialize: "%s"', type(record))
43        return None

Convert passed DASRecords, lists or dicts to JSON. If pretty == True, format the JSON output for easy reading.

ToJSONTransform(pretty=False, **kwargs) 
20    def __init__(self, pretty=False, **kwargs):
21        super().__init__(**kwargs)  # processes 'quiet' and type hints
22        self.pretty = pretty

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.
pretty
def transform( self, record: Union[logger.utils.das_record.DASRecord, float, int, bool, str, dict, list, set]) -> str: 
25    def transform(self, record: Union[DASRecord, float, int, bool, str, dict, list, set]) -> str:
26        """Convert record to JSON."""
27
28        # See if it's something we can process, and if not, try digesting
29        if not self.can_process_record(record):  # inherited from BaseModule()
30            return self.digest_record(record)  # inherited from BaseModule()
31
32        if type(record) is DASRecord:
33            return record.as_json(self.pretty)
34
35        if type(record) in [float, int, bool, str, dict, list, set]:
36            if self.pretty:
37                return json.dumps(record, sort_keys=True, indent=4)
38            else:
39                return json.dumps(record)
40
41        logging.warning('ToJSON transform received record format it could not '
42                        'serialize: "%s"', type(record))
43        return None

Convert record to JSON.