openrvdas.logger.utils.simulate_serial

  1#!/usr/bin/env python3
  2
  3import logging
  4import os.path
  5import subprocess
  6import threading
  7import time
  8
  9from logger.readers.logfile_reader import LogfileReader  # noqa: E402
 10from logger.transforms.slice_transform import SliceTransform  # noqa: E402
 11from logger.writers.text_file_writer import TextFileWriter  # noqa: E402
 12
 13from logger.utils.read_config import read_config  # noqa: E402
 14from logger.utils.timestamp import TIME_FORMAT  # noqa: E402
 15
 16
 17class SimSerial:
 18    """Create a virtual serial port and feed stored logfile data to it."""
 19    ############################
 20
 21    def __init__(self, port, source_file, time_format=TIME_FORMAT,
 22                 use_timestamps=True,
 23                 baudrate=9600, bytesize=8, parity='N', stopbits=1,
 24                 timeout=None, xonxoff=False, rtscts=False, write_timeout=None,
 25                 dsrdtr=False, inter_byte_timeout=None, exclusive=None):
 26        """Takes source file, whether to deliver data at rate indicated by
 27        timestamps, and the standard parameters that a serial port takes."""
 28
 29        # flake8: noqa: E501
 30        logging.warning('simulate_serial.SimSerial is deprecated in favor of simulate_data.SimSerial')
 31
 32        self.source_file = source_file
 33        self.use_timestamps = use_timestamps
 34        self.time_format = time_format
 35
 36        # We'll create two virtual ports: 'port' and 'port_in'; we will write
 37        # to port_in and read the values back out from port
 38        self.read_port = port
 39        self.write_port = port + '_in'
 40
 41        self.serial_params = {'baudrate': baudrate,
 42                              'byteside': bytesize,
 43                              'parity': parity,
 44                              'stopbits': stopbits,
 45                              'timeout': timeout,
 46                              'xonxoff': xonxoff,
 47                              'rtscts': rtscts,
 48                              'write_timeout': write_timeout,
 49                              'dsrdtr': dsrdtr,
 50                              'inter_byte_timeout': inter_byte_timeout,
 51                              'exclusive': exclusive}
 52        self.quit = False
 53
 54        # Finally, find path to socat executable
 55        self.socat_path = None
 56        for socat_path in ['/usr/bin/socat', '/usr/local/bin/socat']:
 57            if os.path.exists(socat_path) and os.path.isfile(socat_path):
 58                self.socat_path = socat_path
 59        if not self.socat_path:
 60            raise NameError('Executable "socat" not found on path. Please refer '
 61                            'to installation guide to install socat.')
 62
 63    ############################
 64    def _run_socat(self):
 65        """Internal: run the actual command."""
 66        verbose = '-d'
 67        write_port_params = 'pty,link=%s,raw,echo=0' % self.write_port
 68        read_port_params = 'pty,link=%s,raw,echo=0' % self.read_port
 69
 70        cmd = [self.socat_path,
 71               verbose,
 72               # verbose,   # repeating makes it more verbose
 73               read_port_params,
 74               write_port_params,
 75               ]
 76        try:
 77            # Run socat process using Popen, checking every second or so whether
 78            # it's died (poll() != None) or we've gotten a quit signal.
 79            logging.info('Calling: %s', ' '.join(cmd))
 80            socat_process = subprocess.Popen(cmd)
 81            while not self.quit and not socat_process.poll():
 82                try:
 83                    socat_process.wait(1)
 84                except subprocess.TimeoutExpired:
 85                    pass
 86
 87        except Exception as e:
 88            logging.error('ERROR: socat command: %s', e)
 89
 90        # If here, process has terminated, or we've seen self.quit. We
 91        # want both to be true: if we've terminated, set self.quit so that
 92        # 'run' loop can exit. If self.quit, terminate process.
 93        if self.quit:
 94            socat_process.kill()
 95        else:
 96            self.quit = True
 97        logging.info('Finished: %s', ' '.join(cmd))
 98
 99    ############################
100    def run(self, loop=False):
101        """Create the virtual port with socat and start feeding it records from
102        the designated logfile. If loop==True, loop when reaching end of input."""
103        self.socat_thread = threading.Thread(target=self._run_socat, daemon=True)
104        self.socat_thread.start()
105        time.sleep(0.2)
106
107        self.reader = LogfileReader(filebase=self.source_file,
108                                    use_timestamps=self.use_timestamps,
109                                    time_format=self.time_format)
110
111        self.strip = SliceTransform('1:')  # strip off the first field)
112        self.writer = TextFileWriter(self.write_port, truncate=True)
113
114        while not self.quit:
115            try:
116                record = self.reader.read()  # get the next record
117                logging.debug('SimSerial got: %s', record)
118
119                # End of input? If loop==True, re-open the logfile from the start
120                if record is None:
121                    if not loop:
122                        break
123                    self.reader = LogfileReader(filebase=self.source_file,
124                                                use_timestamps=self.use_timestamps)
125
126                record = self.strip.transform(record)  # strip the timestamp
127                if record:
128                    logging.debug('SimSerial writing: %s', record)
129                    self.writer.write(record)   # and write it to the virtual port
130            except (OSError, KeyboardInterrupt):
131                break
132
133        # If we're here, we got None from our input, and are done. Signal
134        # for run_socat to exit
135        self.quit = True
136
137
138################################################################################
139if __name__ == '__main__':
140    import argparse
141    parser = argparse.ArgumentParser()
142
143    parser.add_argument('--config', dest='config', default=None,
144                        help='Config file of JSON specs for port-file mappings.')
145
146    parser.add_argument('--logfile', dest='logfile',
147                        help='Log file to read from.')
148
149    parser.add_argument('--time_format', dest='time_format', default=TIME_FORMAT,
150                        help='Format string for parsing timestamp')
151
152    parser.add_argument('--loop', dest='loop', action='store_true',
153                        help='If True, loop when reaching end of sample data')
154
155    parser.add_argument('--port', dest='port',
156                        help='Virtual serial port to open')
157    parser.add_argument('--baud', dest='baud', type=int,
158                        help='Baud rate for port.')
159
160    parser.add_argument('-v', '--verbosity', dest='verbosity',
161                        default=0, action='count',
162                        help='Increase output verbosity')
163    args = parser.parse_args()
164
165    LOGGING_FORMAT = '%(asctime)-15s %(message)s'
166    logging.basicConfig(format=LOGGING_FORMAT)
167
168    LOG_LEVELS = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG}
169    args.verbosity = min(args.verbosity, max(LOG_LEVELS))
170    logging.getLogger().setLevel(LOG_LEVELS[args.verbosity])
171
172    # Okay - get to work here
173
174    if args.config:
175        configs = read_config(args.config)
176        logging.info('Read configs: %s', configs)
177        thread_list = []
178        for inst in configs:
179            config = configs[inst]
180            sim = SimSerial(port=config['port'], source_file=config['logfile'],
181                            time_format=config.get('time_format', args.time_format))
182            sim_thread = threading.Thread(target=sim.run, kwargs={'loop': args.loop},
183                                          daemon=True)
184            sim_thread.start()
185            thread_list.append(sim_thread)
186
187        logging.warning('Running simulated ports for %s', ', '.join(configs.keys()))
188        for thread in thread_list:
189            thread.join()
190
191    # If no config file, just a simple, single serial port
192    elif args.logfile and args.port:
193        sim_serial = SimSerial(port=args.port, baudrate=args.baud,
194                               source_file=args.logfile)
195        sim_serial.run(args.loop)
196
197    # Otherwise, we don't have enough information to run
198    else:
199        parser.error('Either --config or both --logfile and --port must '
200                     'be specified')