openrvdas.logger.utils.base_module

  1#!/usr/bin/env python3
  2"""
  3The biggest thing that this abstract parent class does is help with (optional)
  4type checking of the child class' inputs and outputs. In the past an explicit,
  5but very awkward, form of type checking was used, with child classes passing the
  6Transform class a list of input_format and output_format specifications.
  7
  8That is now deprecated in favor of using Python's type hints. Type hints should be
  9specified for the child class' read(), transform() or write() method. Then the method can
 10call self.can_process_record(record) to see whether it's one of the input types it
 11can handle, and/or check_result to see if the output is as expected. If not, it can
 12"return self.digest_record(record) to have the parent class try to deal with it:
 13
 14E.g.:
 15     def transform(self, record: Union[int, str, float]):
 16        if not self.can_process_record(record):  # inherited from BaseModule()
 17            return self.digest_record(record)  # inherited from BaseModule()
 18         return str(record) + '+'
 19
 20If no type hints are specified, can_process_record() will return True for all
 21records *except* those of type "None" or "list". The logic is that digest_record()
 22will return a None when given a None, and when given a list, will iteratively
 23apply the transform to every element of the list and return the resulting list.
 24
 25Note that the child class can explicitly call super().__init__(quiet=True) or such
 26to initialize the type checking and set its debugging level. If it is not explicitly
 27initialized, it will be done implicitly the first time can_process_record() or
 28digest_record() are called, but with the default of quiet=False.
 29"""
 30import inspect
 31import logging
 32import threading
 33import queue
 34from typing import get_args
 35
 36from logger.utils.das_record import DASRecord  # noqa E402
 37
 38
 39########################################
 40def get_method_type_hints(method):
 41    """
 42    When passed a method via something like
 43
 44        get_method_type_hints(self.__class__.transform)
 45
 46    return a dict of the method's type hints for the arguments
 47    and return value. E.g., if transform() is defined as:
 48
 49        def transform(self, record: int|float) -> str:
 50
 51    will return
 52
 53        {'return': (<class 'str'>),
 54         'record': (<class 'int'>, <class 'float'>)}
 55
 56    The point of this routine is to allow Transform and Writer to sanity
 57    check their inputs.
 58    """
 59    method_args = inspect.getfullargspec(method).annotations
 60    method_types = {k: tuple([value])
 61                    if isinstance(value, type)
 62                    else tuple(get_args(value))
 63                    for k, value in method_args.items()
 64                    }
 65    return method_types
 66
 67
 68################################################################################
 69class BaseModule:
 70    """
 71    Base class for OpenRVDAS Readers, Transforms and Writers.
 72
 73    Implements method for checking whether a received record is in a format
 74    that the derived class can process, and also a method for splitting a
 75    list of records into its elements and calling subclass transform() on them.
 76
 77    Starting with v0.6, BaseModule also implements "mirroring" functionality.
 78    If the optional 'mirror_to' argument is passed to the constructor (and
 79    the subclass is valid for mirroring, i.e. not a Writer), BaseModule
 80    will spin up a thread and a queue to asynchronously write a copy of
 81    every record it processes to the specified Writer.
 82    """
 83    ############################
 84    def __init__(self, quiet=False, encoding='utf-8', encoding_errors='ignore',
 85                 mirror_to=None, *args, **kwargs):
 86        """
 87        ```
 88        quiet - if type checking should log type errors or operate silently.
 89
 90        Two additional arguments govern how records will be encoded/decoded
 91        from bytes, if desired by the Writer subclass when it calls
 92        _encode_str() or _decode_bytes:
 93
 94        encoding - 'utf-8' by default. If empty or None, do not attempt any
 95                decoding and return raw bytes. Other possible encodings are
 96                listed in online documentation here:
 97                https://docs.python.org/3/library/codecs.html#standard-encodings
 98
 99        encoding_errors - 'ignore' by default. Other error strategies are
100                'strict', 'replace', and 'backslashreplace', described here:
101                https://docs.python.org/3/howto/unicode.html#encodings
102
103        mirror_to - Optional Writer to which all records read or transformed
104                by this module (if it is a Reader or Transform) will be
105                "mirrored" (copied). Mirroring happens asynchronously via
106                a queue and background thread to minimize impact on the
107                primary data flow. Writers cannot be mirrored.
108        ```
109        """
110        if kwargs.get('input_format'):
111            logging.warning(f'Code warning: {self.__class__.__name__} use of "input_format"'
112                            'is deprecated in favor of type hints. Please see documentation'
113                            'in logger/utils/base_module.py.')
114        if kwargs.get('output_format'):
115            logging.warning(f'Code warning: {self.__class__.__name__} use of "output_format"'
116                            'is deprecated in favor of type hints. Please see documentation'
117                            'in logger/utils/base_module.py.')
118        self.quiet = quiet
119
120        # Make sure '' behaves the same as None, which is what all the
121        # docstrings say, and would be logical... but then certain things treat
122        # them differently (e.g., file.open(mode='ab', encoding='') throws
123        # ValueError: binary mode doesn't take an encoding argument)
124        if encoding == '':
125            encoding = None
126        self.encoding = encoding
127        self.encoding_errors = encoding_errors
128
129        # Handle mirroring
130        self.mirror_to = mirror_to
131        if self.mirror_to:
132            from logger.writers.writer import Writer
133            if isinstance(self, Writer):
134                logging.warning(f'Writer {self.__class__.__name__} passed "mirror_to" argument. '
135                                'Writers cannot be mirrored.')
136                self.mirror_to = None
137            elif not isinstance(self.mirror_to, Writer):
138                raise TypeError(f'mirror_to must be a Writer, not {type(self.mirror_to)}')
139            else:
140                self.mirror_queue = queue.Queue()
141                self.mirror_thread = threading.Thread(target=self._mirror_output_thread,
142                                                      daemon=True)
143                self.mirror_thread.start()
144
145                # Dynamically wrap the read() or transform() method
146                if hasattr(self, 'read'):
147                    self._original_read = self.read
148                    self.read = self._wrapped_read
149                elif hasattr(self, 'transform'):
150                    self._original_transform = self.transform
151                    self.transform = self._wrapped_transform
152
153    def _mirror_output_thread(self):
154        """Thread to pull records from the queue and write them to the
155        mirror_to writer."""
156        while True:
157            record = self.mirror_queue.get()
158            try:
159                self.mirror_to.write(record)
160            except Exception as e:
161                logging.warning(f'Error writing to mirror_to: {e}')
162
163    def _wrapped_read(self):
164        """Wrapped version of read() that intercepts records and sends
165        them to the mirror_to writer."""
166        record = self._original_read()
167        if record is not None:
168            self.mirror_queue.put(record)
169        return record
170
171    def _wrapped_transform(self, record):
172        """Wrapped version of transform() that intercepts records and sends
173        them to the mirror_to writer."""
174        result = self._original_transform(record)
175        if result is not None:
176            self.mirror_queue.put(result)
177        return result
178
179    ############################
180    def _initialize_type_hints(self, module_type, module_method):
181        """We should only get called from the _initialize_type_hints method of
182        Reader/Transform/Writer subclasses, which should fill in all parameters.
183
184        Retrieve any type hints for child read()/transform()/write() method so we
185        can check whether the type of record we've received can be parsed
186        natively or not."""
187        self.module_type = module_type
188        self.module_method = module_method
189
190        # We make stupid assumption that the input variable is called 'record'
191        method_type_hints = get_method_type_hints(self.module_method)
192        self.input_types = method_type_hints.get('record')
193        self.return_types = method_type_hints.get('return')
194
195        # logging.warning(f'input_types: {self.input_types}')
196        # logging.warning(f'return_types: {self.return_types}')
197
198        # Other things we'd want to make sure are defined.
199        self.class_name = self.__class__.__name__
200        self.initialized = True
201
202    ############################
203    def can_process_record(self, record):
204        """ Is this record in a format that the transform or writer can handle?
205
206        - If there are type hints: True if type of record is in type hints.
207        - If there are no type hints: False if None or list, otherwise True.
208
209        The logic is that if there are no type hints and we see False or
210        a list, we expect digest_record() to be called to deal with it."""
211
212        try:  # if we've not been initialized with type hints, initialize now
213            self.initialized or True
214        except AttributeError:
215            # This will call the subclass initialization, e.g.
216            # Transform._initialize_type_hints(), which will in turn call
217            # OpenRVDASModule._initialize_type_hints()
218            self._initialize_type_hints()
219
220        # Special case: we want to turn empty str records into None. By saying
221        # no, record should get punted to digest_record(), which will do the
222        # right thing.
223        if isinstance(record, str) and not len(record):
224            return False
225
226        if self.input_types:
227            return isinstance(record, self.input_types)
228
229        # If not type hints, make some judgment calls. Say "no" to None
230        # and to lists, because we'll expect that answer to trigger a
231        # call to digest_record(), which will handle them.
232        if record is None or isinstance(record, list):
233            return False
234        return True
235
236    ############################
237    def digest_record(self, record):
238        """ Try to digest record down into a format that the method can
239        handle. Typically that will mean that we've been handed a list of
240        records that we need to break into individual records."""
241
242        try:  # if we've not been initialized with type hints, initialize now
243            self.initialized or True
244        except AttributeError:
245            self._initialize_type_hints()
246
247        # Go through our litany of things that reduce to None
248        if record is None:
249            return None
250
251        if isinstance(record, str) and not len(record):
252            return None
253
254        # If it's a type the method can handle directly (though, if so,
255        # why were we called?!?
256        if self.can_process_record(record) and not self.quiet:
257            logging.warning(f'{self.class_name}: digest_record() called unnecessarily.')
258            logging.warning(f'Can process {self.input_types}; received {type(record)}: {record}')
259            return self.module_method(record)
260
261        # We know how to deal with it if it's a list: Apply to components,
262        # stripping out any None's
263        if isinstance(record, list):
264            result = [self.module_method(self, r) for r in record if r is not None]
265            return [r for r in result if r is not None]  # remove Nones
266
267        # Is record a number we can convert to a string?
268        if str in self.input_types and isinstance(record, (int, float)):
269            return str(record)
270
271        # If it's a DASRecord, serialize it as JSON
272        if str in self.input_types and isinstance(record, DASRecord):
273            return record.as_json()
274
275        # If we don't know how to deal with it
276        if not self.quiet:
277            logging.warning(f'Unable to convert record to format "{self.class_name}" can process')
278            logging.warning(f'Must be instance or list of {self.input_types}')
279            logging.warning(f'Received {type(record)}: {record}')
280        return None
281
282    ############################
283    def _unescape_str(self, the_str):
284        """Unescape a string by encoding it to bytes, then unescaping when we
285        decode it. Ugly.
286        """
287        if not self.encoding:
288            return the_str
289
290        encoded = the_str.encode(encoding=self.encoding, errors=self.encoding_errors)
291        return encoded.decode('unicode_escape')
292
293    ############################
294    def _encode_str(self, the_str, unescape=False):
295        """Encode a string to bytes, optionally unescaping things like \n and \r.
296        Unescaping requires ugly convolutions of encoding, then decoding while we
297        escape things, then encoding a second time.
298        """
299        if not self.encoding:
300            return the_str
301        if unescape:
302            the_str = self._unescape_str(the_str)
303        return the_str.encode(encoding=self.encoding, errors=self.encoding_errors)
304
305    ############################
306    def _decode_bytes(self, record, allow_empty: bool = False):
307        """Decode a record from bytes to str, if we have an encoding specified."""
308        if record is None:
309            return None
310
311        if not record and not allow_empty:  # if it's an empty record but not None
312            return None
313
314        if not self.encoding:
315            return record
316
317        if self.encoding == 'hex':
318            try:
319                r = record.hex()
320                return r
321            except Exception as e:
322                logging.warning('Error decoding string "%s" from encoding "%s": %s',
323                                record, self.encoding, str(e))
324                return None
325
326        try:
327            return record.decode(encoding=self.encoding,
328                                 errors=self.encoding_errors)
329        except UnicodeDecodeError as e:
330            logging.warning('Error decoding string "%s" from encoding "%s": %s',
331                            record, self.encoding, str(e))
332            return None