openrvdas.logger.utils.simulate_data

  1#!/usr/bin/env python3
  2"""Simulate a live data feed by sending stored logger data to
  3specified UDP ports and/or simulated (temporary) serial ports.
  4
  5May either be invoked for a single data feed with command line
  6options, or by specifying a YAML-format configuration file that sets
  7up multiple feeds at once.
  8
  9To invoke a single data feed:
 10
 11   simulate_data.py --udp 5501 --filebase /data/2019-05-11/raw/GYRO
 12
 13will read timestamped lines from files matching /data/2019-05-11/raw/GYRO*
 14and broadcast them via UDP port 5501:
 15
 16$HEHDT,087.1,T*21
 17$HEHDT,087.1,T*21
 18$HEHDT,087.1,T*21
 19$HEHDT,087.1,T*21
 20
 21By default, the reader assumes that the log file record format is
 22"{timestamp:ti} {record}" but if, for example, the timestamp has a different
 23format, or a different separator is used between the timestamp and record,
 24the default may be overridden with the --record_format argument. E.g. if a
 25comma is used as the delimiter between timestamp and record:
 26
 272019-11-28T01:01:38.762221Z,$HEHDT,087.1,T*21
 282019-11-28T01:01:38.953182Z,$HEHDT,087.1,T*21
 29
 30you may specify
 31
 32   simulate_data.py --udp 6224 \
 33     --filebase test/NBP1406/gyr1/raw/NBP1406_gyr1-2014-08-01 \
 34     --record_format '{timestamp:ti},{record}'
 35
 36Unless --no-loop is specified on the command line, the system will
 37rewind to the beginning of all log files when it reaches the end of
 38its input.
 39
 40Instead of --udp, you may also specify --serial (and optionally
 41--baudrate) to simulate a serial port:
 42
 43   simulate_data.py --serial /tmp/ttyr05 --filebase /data/2019-05-11/raw/GYRO
 44
 45If --config is specified
 46
 47   simulate_data.py --config data/2019-05-11/simulate_config.yaml
 48
 49the script will expect a YAML file keyed by instrument names, where
 50each instrument name references a dict including keys 'class' (Serial
 51or UDP), 'port' (e.g. 5501 or /tmp/ttyr05) and 'filebase'. It may
 52optionally include 'eol', 'timestamp' and 'time_format'
 53keys:
 54
 55############# Gyro ###############
 56  gyro:
 57    class: UDP
 58    timestamp: true
 59    eol: \r
 60    port: 56332
 61    filebase: /data/2019-05-11/raw/GYRO
 62
 63############# Fluorometer ###############
 64fluorometer:
 65  class: Serial
 66  port: /tmp/ttyr04
 67  baudrate: 9600
 68  filebase: /data/2019-05-11/raw/FLUOROMETER
 69
 70Note that if class is 'Serial', it may also include the full range of
 71serial port options:
 72
 73    baudrate: 9600
 74    bytesize: 8
 75    parity: N
 76    stopbits: 1
 77    timeout: false
 78    xonxoff: false
 79    rtscts: false,
 80    write_timeout: false
 81    dsrdtr: false
 82    inter_byte_timeout: false
 83    exclusive: false
 84
 85"""
 86import glob
 87import logging
 88import os.path
 89import parse
 90import pty
 91import threading
 92import time
 93
 94from logger.readers.logfile_reader import LogfileReader  # noqa: E402
 95from logger.writers.udp_writer import UDPWriter  # noqa: E402
 96
 97from logger.utils.read_config import read_config  # noqa: E402
 98from logger.utils.timestamp import TIME_FORMAT  # noqa: E402
 99
100
101class SimUDP:
102    """Open a network port and feed stored logfile data to it."""
103    ############################
104
105    def __init__(self, port, name=None, filebase=None, record_format=None,
106                 time_format=TIME_FORMAT, eol='\n', input_eol=None, quiet=False,
107                 use_timestamps=True):
108        """
109        ```
110        port -  UDP port on which to write records.
111
112        name -  Optional user-friendly name to display on warning messages
113
114        filebase - Prefix string to be matched (with a following "*") to find
115                   files to be used. e.g. /tmp/log/NBP1406/knud/raw/NBP1406_knud
116
117        record_format
118                     If specified, a custom record format to use for extracting
119                     timestamp and record. The default is '{timestamp:ti} {record}'
120
121        time_format - What format to use for timestamp
122
123        eol - String to append to end of an output record
124
125        input_eol - Optional string by which to recognize the end of a record
126
127        use_timestamps - If true, emit records at delays corresponding to the
128                         differences in their timestamps
129        ```
130        """
131        self.port = port
132        self.name = name
133        self.time_format = time_format
134        self.filebase = filebase
135        self.record_format = record_format or '{timestamp:ti} {record}'
136        self.compiled_record_format = parse.compile(self.record_format)
137        self.eol = eol
138        self.input_eol = input_eol
139        self.quiet = quiet
140        self.use_timestamps = use_timestamps
141
142        # Do we have any files we can actually read from?
143        if not glob.glob(filebase + '*'):
144            logging.warning('No files matching "%s*"', filebase)
145            self.quit_flag = True
146            return
147
148        self.reader = LogfileReader(filebase=filebase,
149                                    use_timestamps=self.use_timestamps,
150                                    record_format=self.record_format,
151                                    time_format=self.time_format,
152                                    eol=self.input_eol, quiet=self.quiet)
153        self.writer = UDPWriter(port=port, eol=eol)
154
155        self.first_time = True
156        self.quit_flag = False
157
158    ############################
159    def run(self, loop=False):
160        """Start reading and writing data. If loop==True, loop when reaching
161        end of input.
162        """
163        logging.info('Starting %s: %s', self.port, self.filebase)
164        try:
165            while not self.quit_flag:
166                record = self.reader.read()
167
168                # If we don't have a record, we're (probably) at the end of
169                # the file. If it's the first time we've tried reading, it
170                # means we probably didn't get a usable file. Either break out
171                # (if we're not looping, or if we don't have a usable file),
172                # or start reading from the beginning (if we are looping and
173                # have a usable file).
174                if record is None:
175                    if not loop or self.first_time:
176                        break
177                    logging.info('Looping instrument %s', self.filebase)
178                    self.reader = LogfileReader(filebase=self.filebase,
179                                                record_format=self.record_format,
180                                                use_timestamps=True,
181                                                eol=self.input_eol,
182                                                quiet=self.quiet)
183                    continue
184
185                # We've now got a record. Try parsing timestamp off it
186                try:
187                    parsed_record = self.compiled_record_format.parse(record).named
188                    record = parsed_record['record']
189
190                # We had a problem parsing. Discard record and try reading next one.
191                except (KeyError, ValueError, AttributeError):
192                    logging.warning('%s: Unable to parse record into "%s"',
193                                    self.name, self.record_format)
194                    logging.warning('Record: "%s"', record)
195                    continue
196
197                if not record:
198                    continue
199
200                self.writer.write(record)
201                self.first_time = False
202
203        except (OSError, KeyboardInterrupt):
204            self.quit_flag = True
205
206        logging.info('Finished %s', self.filebase)
207
208
209################################################################################
210class SimSerial:
211    """Create a virtual serial port and feed stored logfile data to it."""
212    ############################
213
214    def __init__(self, port, name=None, time_format=TIME_FORMAT, filebase=None,
215                 record_format=None, eol='\n', input_eol=None,
216                 baudrate=9600, bytesize=8, parity='N', stopbits=1,
217                 timeout=None, xonxoff=False, rtscts=False, write_timeout=None,
218                 dsrdtr=False, inter_byte_timeout=None, exclusive=None,
219                 quiet=False, use_timestamps=True):
220        """
221        Simulate a serial port, feeding it data from the specified file.
222
223        ```
224        port - Temporary serial port to create and make available for reading
225               records.
226
227        name -  Optional user-friendly name to display on warning messages
228
229        time_format - What format to use for timestamp
230
231        filebase     Possibly wildcarded string specifying files to be opened.
232
233        record_format
234                     If specified, a custom record format to use for extracting
235                     timestamp and record. The default is '{timestamp:ti} {record}'.
236
237        eol - String by which to recognize the end of a record
238
239        input_eol - Optional string by which to recognize the end of a record
240
241        use_timestamps - If true, emit records at delays corresponding to the
242                         differences in their timestamps
243        ```
244        """
245        # We'll create two virtual ports: 'port' and 'port_in'; we will write
246        # to port_in and read the values back out from port
247        self.read_port = port
248        self.write_port = port + '_in'
249        self.name = name
250        self.time_format = time_format
251        self.filebase = filebase
252        self.record_format = record_format or '{timestamp:ti} {record}'
253        self.compiled_record_format = parse.compile(self.record_format)
254        self.eol = eol
255        self.input_eol = input_eol
256        self.serial_params = None
257        self.quiet = quiet
258        self.use_timestamps = use_timestamps
259
260        # Complain, but go ahead if read_port or write_port exist.
261        for path in [self.read_port, self.write_port]:
262            if os.path.exists(path):
263                logging.warning('Path %s exists; overwriting!', path)
264
265        # Do we have any files we can actually read from?
266        if not glob.glob(filebase + '*'):
267            logging.warning('No files matching "%s*"', filebase)
268            return
269
270        # Set up our parameters
271        self.serial_params = {'baudrate': baudrate,
272                              'byteside': bytesize,
273                              'parity': parity,
274                              'stopbits': stopbits,
275                              'timeout': timeout,
276                              'xonxoff': xonxoff,
277                              'rtscts': rtscts,
278                              'write_timeout': write_timeout,
279                              'dsrdtr': dsrdtr,
280                              'inter_byte_timeout': inter_byte_timeout,
281                              'exclusive': exclusive}
282        self.quit = False
283
284        # Create simulated serial port (something like /dev/ttys2), and link
285        # it to the port they want to connect to (like /tmp/tty_s330).
286        self.write_fd, self.read_fd = pty.openpty()  # open the pseudoterminal
287        true_read_port = os.ttyname(self.read_fd)  # this is the true filename of port
288
289        # Get rid of any previous symlink if it exists, and symlink the new pty
290        try:
291            os.unlink(self.read_port)
292        except FileNotFoundError:
293            pass
294        os.symlink(true_read_port, self.read_port)
295
296        self.reader = LogfileReader(filebase=self.filebase,
297                                    use_timestamps=self.use_timestamps,
298                                    record_format=self.record_format,
299                                    time_format=self.time_format,
300                                    eol=self.input_eol, quiet=self.quiet)
301
302    ############################
303    def __del__(self):
304        # Get rid of the symlink we've created
305        try:
306            os.unlink(self.read_port)
307        except FileNotFoundError:
308            pass
309
310    ############################
311    def run(self, loop=False):
312        # If self.serial_params is None, it means that either read or
313        # write device already exist, so we shouldn't actually run, or
314        # we'll destroy them.
315        if not self.serial_params:
316            return
317
318        time.sleep(0.05)
319
320        logging.info('Starting %s: %s', self.read_port, self.filebase)
321        while not self.quit:
322            try:
323                record = self.reader.read()  # get the next record
324                logging.debug('SimSerial got: %s', record)
325
326                # End of input? If loop==True, re-open the logfile from the start
327                if record is None:
328                    if not loop:
329                        break
330
331                    self.reader = LogfileReader(filebase=self.filebase,
332                                                use_timestamps=True,
333                                                record_format=self.record_format,
334                                                time_format=self.time_format,
335                                                eol=self.input_eol, quiet=self.quiet)
336
337                # We've now got a record. Try parsing timestamp off it
338                logging.debug(f'Read record: "{record}"')
339                try:
340                    parsed_record = self.compiled_record_format.parse(record).named
341                    record = parsed_record['record']
342
343                # We had a problem parsing. Discard record and try reading next one.
344                except (KeyError, ValueError, TypeError, AttributeError):
345                    logging.warning('%s: Unable to parse record into "%s"',
346                                    self.name, self.record_format)
347                    logging.warning('Record: "%s"', record)
348                    continue
349
350                if not record:
351                    continue
352
353                logging.debug('SimSerial writing: %s', record)
354                os.write(self.write_fd, (record + self.eol).encode('utf8'))
355
356            except (OSError, KeyboardInterrupt):
357                break
358
359        # If we're here, we got None from our input, and are done.
360        self.quit = True
361
362
363################################################################################
364if __name__ == '__main__':
365    import argparse
366    parser = argparse.ArgumentParser()
367
368    parser.add_argument('--config', dest='config', default=None,
369                        help='Config file of JSON specs for port-file mappings.')
370
371    parser.add_argument('--serial', dest='serial',
372                        help='Virtual serial port to open')
373    parser.add_argument('--baud', dest='baud', type=int,
374                        help='Optional baud rate for serial port.')
375
376    parser.add_argument('--udp', dest='udp', type=int,
377                        help='UDP port to broadcast on')
378
379    parser.add_argument('--time_format', dest='time_format', default=TIME_FORMAT,
380                        help='Format string for parsing timestamp')
381
382    parser.add_argument('--filebase', dest='filebase',
383                        help='Basename of logfiles to read from. A "*" will be '
384                        'appended to this string and all matching files will '
385                        'be read in order.')
386
387    parser.add_argument('--record_format', dest='record_format',
388                        default='{timestamp:ti} {record}',
389                        help='If specified, a custom record format to use for extracting '
390                        'timestamp and record. The default is {timestamp:ti} {record}')
391
392    parser.add_argument('--loop', dest='loop', action='store_true', default=True,
393                        help='If True, loop when reaching end of sample data')
394
395    parser.add_argument('--quiet', dest='quiet', action='store_true', default=True,
396                        help='If True, silently ignore unparseable records')
397
398    parser.add_argument('--no_loop', dest='no_loop', action='store_true',
399                        help='If True, don\'t loop when reaching end of '
400                        'sample data')
401
402    parser.add_argument('-v', '--verbosity', dest='verbosity',
403                        default=0, action='count',
404                        help='Increase output verbosity')
405    args = parser.parse_args()
406
407    LOGGING_FORMAT = '%(asctime)-15s %(lineno)d %(message)s'
408    logging.basicConfig(format=LOGGING_FORMAT)
409
410    LOG_LEVELS = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}
411    args.verbosity = min(args.verbosity, max(LOG_LEVELS))
412    logging.getLogger().setLevel(LOG_LEVELS[args.verbosity])
413
414    # Default is to loop unless told otherwise
415    loop = args.loop and not args.no_loop
416
417    # Have we been given a config file?
418    if args.config:
419        configs = read_config(args.config)
420        logging.info('Read configs: %s', configs)
421        thread_list = []
422        for inst, config in configs.items():
423            if 'class' not in config:
424                logging.warning('No class definition for config "%s"', inst)
425                continue
426
427            # Stash name of config
428            config['name'] = inst
429            # Save class of simulator, and remove it from the config dict
430            inst_class = config['class']
431            del config['class']
432
433            # Fold in some things from the command line, if they're
434            # not specified in the config itself.
435            if 'time_format' not in config:
436                config['time_format'] = args.time_format
437            if 'record_format' not in config:
438                config['record_format'] = args.record_format
439            if 'quiet' not in config:
440                config['quiet'] = args.quiet
441
442            # Create the appropriate simulator with the config
443            if inst_class == 'Serial':
444                writer = SimSerial(**config)
445            elif inst_class == 'UDP':
446                writer = SimUDP(**config)
447            else:
448                logging.error('Unknown class for config %s', inst_class)
449                logging.error('Acceptable classes are "Serial" and "UDP"')
450                continue
451
452            writer_thread = threading.Thread(target=writer.run, kwargs={'loop': loop},
453                                             name=inst, daemon=True)
454            writer_thread.start()
455            thread_list.append(writer_thread)
456
457        logging.info('Running simulated ports for %s', ', '.join(configs.keys()))
458
459        try:
460            for thread in thread_list:
461                thread.join()
462            logging.warning('All processes have completed - exiting')
463        except KeyboardInterrupt:
464            logging.warning('Keyboard interrupt - exiting')
465            pass
466
467    # If no config file, just a simple, single source, create and run a
468    # single simulator.
469    else:
470        if not args.filebase:
471            parser.error('Either --config or --filebase must be specified')
472
473        # Is it a serial port?
474        if args.serial:
475            simulator = SimSerial(port=args.serial,
476                                  time_format=args.time_format,
477                                  baudrate=args.baud,
478                                  filebase=args.filebase,
479                                  record_format=args.record_format,
480                                  quiet=args.quiet)
481        # Is it a UDP port?
482        elif args.udp:
483            simulator = SimUDP(port=args.udp,
484                               time_format=args.time_format,
485                               filebase=args.filebase,
486                               record_format=args.record_format,
487                               quiet=args.quiet)
488        else:
489            parser.error('If --filebase specified, must also specify either --serial '
490                         'or --udp.')
491
492        # Run it
493        simulator.run(loop)