openrvdas.logger.writers.logger_manager_writer

  1#!/usr/bin/env python3
  2"""
  3Writer that sends records it receives to the LoggerManager as commands. Can
  4be used in conjunction with GeofenceTransform for automatically switching
  5modes when entering/exiting an EEZ, or with a QCFilterTransform to turn off
  6loggers when conditions are violated.
  7
  8Multiple commands may be sent in a single record by separating them with
  9semicolons.
 10
 11In addition to the normally-accepted LoggerManager commands, an additional
 12one: "sleep N" is recognized, which will pause the writer N seconds before
 13writing the subsequent command. This allows time, if needed, for the effects
 14of prior commands to settle.
 15
 16Sample logger that switches modes when entering/exiting EEZ:
 17```
 18# Read the latest lat/lon from the Cached Data Server
 19readers:
 20  class: CachedDataReader
 21  kwargs:
 22    data_server: localhost:8766
 23    subscription:
 24      fields:
 25        s330Latitude:
 26          seconds: 0
 27        s330Longitude:
 28          seconds: 0
 29# Look for lat/lon values in the DASRecords and emit appropriate commands
 30# when entering/leaving EEZ. Note that EEZ files in GML format can be
 31# downloaded from https://marineregions.org/eezsearch.php.
 32transforms:
 33  - class: GeofenceTransform
 34    module: logger.transforms.geofence_transform
 35    kwargs:
 36      latitude_field_name: s330Latitude,
 37      longitude_field_name: s330Longitude
 38      boundary_file_name: /tmp/eez.gml
 39      leaving_boundary_message: set_active_mode write+influx
 40      entering_boundary_message: set_active_mode no_write+influx
 41# Send the messages that we get from geofence to the LoggerManager
 42writers:
 43  - class: LoggerManagerWriter
 44    module: logger.writers.logger_manager_writer
 45    kwargs:
 46      database: django
 47      allowed_prefixes:
 48        - 'set_active_mode '
 49        - 'sleep '
 50```
 51"""
 52import logging
 53import time
 54
 55
 56from logger.writers.writer import Writer  # noqa: E402
 57from server.server_api_command_line import ServerAPICommandLine  # noqa: E402
 58
 59
 60class LoggerManagerWriter(Writer):
 61    """Write received text records to the LoggerManager."""
 62
 63    def __init__(self, database=None, api=None, allowed_prefixes=[], **kwargs):
 64        """Write received text records as commands to the LoggerManager.
 65        ```
 66        database
 67                String indicating which database the LoggerManager is using: django,
 68                sqlite, memory. Either this or 'api', but not both, must be specified.
 69
 70        api
 71                An instance of server_api.ServerAPI to use to communicate with the
 72                LoggerManager. Either this or 'database', but not both, must be specified.
 73
 74        allowed_prefixes
 75                Optional list of strings. If specified, only records whose prefixes match
 76                something in this list will be passed on as commands.
 77
 78        See server/server_api_command_line.py for recognized commands.
 79
 80        Multiple commands may be sent in a single record by separating them with
 81        semicolons.
 82
 83        In addition to the normally-accepted LoggerManager commands, an additional
 84        one: "sleep N" is recognized, which will pause the writer N seconds before
 85        writing the subsequent command. This allows time, if needed, for the effects
 86        of prior commands to settle.
 87        ```
 88        """
 89        super().__init__(**kwargs)  # processes 'quiet' and type hints
 90
 91        if database and api:
 92            raise ValueError('Must specify either "database" or "api" but not both.')
 93
 94        # If database specified, create appropriate api instance
 95        if database:
 96            if database == 'django':
 97                from django_gui.django_server_api import DjangoServerAPI
 98                api = DjangoServerAPI()
 99            elif database == 'memory':
100                from server.in_memory_server_api import InMemoryServerAPI
101                api = InMemoryServerAPI()
102            elif database == 'sqlite':
103                from server.sqlite_server_api import SQLiteServerAPI
104                api = SQLiteServerAPI()
105            else:
106                raise ValueError('Parameter "database" must be one of [django, memory, sqlite], '
107                                 f'found "{database}"')
108
109        # If not database, we'd better have an api specified
110        elif not api:
111            raise ValueError('Must specify one of "database" or "api".')
112
113        self.command_parser = ServerAPICommandLine(api=api)
114
115        if not type(allowed_prefixes) is list:
116            raise ValueError(f'Parameter "allowed_prefixes" must be a list of strings, '
117                             f'found "{allowed_prefixes}"')
118        self.allowed_prefixes = allowed_prefixes
119
120    ############################
121    def write(self, record: str):
122        """ Write out record, appending a newline at end."""
123
124        # See if it's something we can process, and if not, try digesting
125        if not self.can_process_record(record):  # inherited from BaseModule()
126            self.digest_record(record)  # inherited from BaseModule()
127            return
128
129        # If there are semicolons, split into list and process sequentially
130        if record.find(';') > -1:
131            for single_record in record.split(';'):
132                self.write(single_record)
133            return
134
135        # Can we find our command in any of the allowed prefixes?
136        approved = True in [record.find(prefix) == 0 for prefix in self.allowed_prefixes]
137        if self.allowed_prefixes and not approved:
138            if not self.quiet:
139                logging.error(f'Command does not match any allowed prefixes: "{record}"')
140
141        # If it's our special "sleep" command
142        elif record.find('sleep') == 0:
143            try:
144                cmd, interval_str = record.split(' ')
145                interval = float(interval_str)
146            except ValueError:
147                if not self.quiet:
148                    logging.error(f'Could not parse command into "sleep [seconds]": {record}')
149                return
150            if not self.quiet:
151                logging.info(f'Sleeping {interval} seconds')
152            time.sleep(interval)
153
154        else:
155            if not self.quiet:
156                logging.info(f'Writing command: {record}')
157            self.command_parser.process_command(record)