openrvdas.logger.utils.simulate_network

  1#!/usr/bin/env python3
  2"""Read stored logger data and serve at realistic intervals over network ports.
  3
  4Run with, e.g.
  5```
  6    logger/utils/simulate_network.py \
  7         --config test/SKQ201822S/network_sim_SKQ201822S.yaml \
  8         --loop
  9```
 10and you should be able to listen on the various ports that Sikuliaq
 11instruments post data to:
 12```
 13    logger/listener/listen.py --network :53131 --write_file -
 14```
 15for wind_mast_stbd (see test/sikuliaq/skq_ports.txt for a table of the
 16instrument to port mappings active during cruise SKQ201822.
 17"""
 18import logging
 19import threading
 20
 21from logger.readers.logfile_reader import LogfileReader  # noqa: E402
 22from logger.transforms.slice_transform import SliceTransform  # noqa: E402
 23from logger.transforms.timestamp_transform import TimestampTransform  # noqa: E402
 24from logger.transforms.prefix_transform import PrefixTransform  # noqa: E402
 25from logger.writers.udp_writer import UDPWriter  # noqa: E402
 26
 27from logger.utils.read_config import read_config  # noqa: E402
 28
 29
 30class SimNetwork:
 31    """Open a network port and feed stored logfile data to it."""
 32    ############################
 33
 34    def __init__(self, port, filebase, instrument):
 35        """
 36        ```
 37        port -  UDP port on which to write records.
 38
 39        filebase - Prefix string to be matched (with a following "*") to fine
 40                   files to be used. e.g. /tmp/log/NBP1406/knud/raw/NBP1406_knud
 41
 42        instrument - Instrument name prefix to add before sendind out on wire
 43        ```
 44        """
 45        self.filebase = filebase
 46        self.reader = LogfileReader(filebase=filebase, use_timestamps=True)
 47        self.slice_n = SliceTransform(fields='1:')  # grab 2nd and subsequent fields
 48        self.timestamp = TimestampTransform()
 49        self.prefix = PrefixTransform(instrument)
 50        self.writer = UDPWriter(port=port)
 51        self.instrument = instrument
 52        self.first_time = True
 53        self.quit_flag = False
 54
 55    ############################
 56    def run(self, loop=False):
 57        """Start reading and writing data. If loop==True, loop when reaching
 58        end of input.
 59        """
 60        logging.info('Starting %s', self.instrument)
 61        try:
 62            while not self.quit_flag:
 63                record = self.reader.read()
 64
 65                # If we don't have a record, we're (probably) at the end of
 66                # the file. If it's the first time we've tried reading, it
 67                # means we probably didn't get a usable file. Either break out
 68                # (if we're not looping, or if we don't have a usable file),
 69                # or start reading from the beginning (if we are looping and
 70                # have a usable file).
 71                if not record:
 72                    if not loop or self.first_time:
 73                        break
 74                    logging.info('Looping instrument %s', self.instrument)
 75                    self.reader = LogfileReader(filebase=self.filebase,
 76                                                use_timestamps=True)
 77                    continue
 78
 79                # Strip off timestamp and tack on a new one
 80                record = self.slice_n.transform(record)
 81                record = self.timestamp.transform(record)
 82
 83                # Add instrument name back on, and write to specified network
 84                record = self.prefix.transform(record)
 85                self.writer.write(record)
 86                self.first_time = False
 87
 88        except (OSError, KeyboardInterrupt):
 89            self.quit_flag = True
 90
 91        logging.info('Finished %s', self.instrument)
 92
 93
 94################################################################################
 95if __name__ == '__main__':
 96    import argparse
 97    parser = argparse.ArgumentParser()
 98
 99    parser.add_argument('--config', dest='config', default=None,
100                        help='Config file of JSON specs for port-file mappings.')
101
102    parser.add_argument('--port', dest='port', default=None, type=int,
103                        help='UDP port to write to. E.g. 6224.')
104
105    parser.add_argument('--filebase', dest='filebase', default=None,
106                        help='Base string of log file to be read from (with '
107                        'a following "*" for match). E.g. '
108                        '/tmp/log/NBP1406/knud/raw/NBP1406_knud')
109
110    parser.add_argument('--instrument', dest='instrument', help='Prefix to add, '
111                        'if file *doesn\'t* have instrument prefix')
112
113    parser.add_argument('--loop', dest='loop', action='store_true',
114                        help='If True, loop when reaching end of sample data')
115
116    parser.add_argument('-v', '--verbosity', dest='verbosity',
117                        default=0, action='count',
118                        help='Increase output verbosity')
119    args = parser.parse_args()
120
121    LOGGING_FORMAT = '%(asctime)-15s %(message)s'
122    logging.basicConfig(format=LOGGING_FORMAT)
123
124    LOG_LEVELS = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}
125    args.verbosity = min(args.verbosity, max(LOG_LEVELS))
126    logging.getLogger().setLevel(LOG_LEVELS[args.verbosity])
127
128    # Okay - get to work here
129    if args.config:
130        configs = read_config(args.config)
131        logging.info('Read configs: %s', configs)
132        thread_list = []
133        writer_list = []
134        for instrument, config in configs.items():
135            port = config['port']
136            filebase = config['filebase']
137            writer = SimNetwork(port=port, filebase=filebase, instrument=instrument)
138            writer_thread = threading.Thread(target=writer.run,
139                                             kwargs={'loop': args.loop},
140                                             daemon=True)
141            writer_thread.start()
142            thread_list.append(writer_thread)
143            writer_list.append('%s %d, %s' % (instrument, port, filebase))
144
145        logging.warning('Running simulated ports for:\n%s', '\n'.join(writer_list))
146
147        # Wait for all the threads to end
148        for thread in thread_list:
149            thread.join()
150
151    # If no config file, just a simple, single network writer
152    elif args.port and args.filebase:
153        sim_network = SimNetwork(network=args.port, filebase=args.filebase,
154                                 instrument=args.instrument)
155        sim_network.run(args.loop)
156
157    # Otherwise, we don't have enough information to run
158    else:
159        parser.error('Either --config or --port, --filebase and --instrument '
160                     'must be specified')