openrvdas.logger.listener.listen
Instantiates and runs the Listener class. Try
listen.py --help
for details.
Examples:
logger/listener/listen.py --logfile test/NBP1700/s330/raw/NBP1700_s330 --interval 0.25 --transform_slice 1: --transform_timestamp --transform_prefix s330 --write_file -
(Reads lines from the Seapath300 sample logfiles every 0.25 seconds, strips the old timestamps off, prepends a new one, then the prefix 's330', then writes the result to stdout.)
logger/listener/listen.py --config_file test/configs/simple_logger.yaml
(Instantiates logger from config file that says to read from the project's LICENSE file, prepend a timestamp and the string "license:" and writ to stdout every 0.2 seconds.)
The listen.py script is essentially a form of 'cat' on steroids, reading records from files, serial or network ports, modifying what it receives, then writing it back out to somewhere else.
For fun, you can even run listen.py as an Ouroboros script, feeding it on its own output:
echo x > tmp
listen.py --file tmp --prefix p --write_file tmp --tail --interval 1 -v -v
1#!/usr/bin/env python3 2"""Instantiates and runs the Listener class. Try 3``` 4 listen.py --help 5``` 6for details. 7 8Examples: 9 ``` 10 logger/listener/listen.py \ 11 --logfile test/NBP1700/s330/raw/NBP1700_s330 \ 12 --interval 0.25 \ 13 --transform_slice 1: \ 14 --transform_timestamp \ 15 --transform_prefix s330 \ 16 --write_file - 17 ``` 18(Reads lines from the Seapath300 sample logfiles every 0.25 seconds, 19strips the old timestamps off, prepends a new one, then the prefix 20's330', then writes the result to stdout.) 21 ``` 22 logger/listener/listen.py \ 23 --config_file test/configs/simple_logger.yaml 24 ``` 25(Instantiates logger from config file that says to read from the 26project's LICENSE file, prepend a timestamp and the string "license:" 27and writ to stdout every 0.2 seconds.) 28 29The listen.py script is essentially a form of 'cat' on steroids, 30reading records from files, serial or network ports, modifying what it 31receives, then writing it back out to somewhere else. 32 33For fun, you can even run listen.py as an Ouroboros script, feeding it on its 34own output: 35``` 36 echo x > tmp 37 listen.py --file tmp --prefix p --write_file tmp --tail --interval 1 -v -v 38``` 39""" 40import argparse 41import importlib 42import logging 43import pprint 44import re 45import sys 46 47 48# flake8: noqa E402, F406 49from logger.readers import * 50from logger.transforms import * 51from logger.writers import * 52from logger.utils import read_config, timestamp, nmea_parser, record_parser 53from logger.utils.stderr_logging import StdErrLoggingHandler, STDERR_FORMATTER 54from logger.listener.listener import Listener 55 56 57################################################################################ 58class ListenerFromLoggerConfig(Listener): 59 """Helper class for instantiating a Listener object from a Python dict.""" 60 ############################ 61 62 def __init__(self, config, log_level=None): 63 """Create a Listener from a Python config dict.""" 64 65 if not type(config) is dict: 66 raise ValueError('ListenerFromLoggerConfig expects config of type ' 67 '"dict" but received one of type "%s": %s' 68 % (type(config), str(config))) 69 70 # Extract keyword args from config and instantiate. 71 logging.debug('ListenerFromLoggerConfig instantiating logger ' 72 'config: %s', pprint.pformat(config)) 73 try: 74 kwargs = self._kwargs_from_config(config) 75 except ValueError as e: 76 config_name = config.get('name', 'unknown logger') 77 raise ValueError('Config for %s: %s' % (config_name, e)) 78 79 super().__init__(**kwargs) 80 81 ############################ 82 def _kwargs_from_config(self, config_dict): 83 """Parse a kwargs from a JSON string, making exceptions for keywords 84 'readers', 'transforms', and 'writers' as internal class references.""" 85 if not config_dict: 86 return {} 87 88 if not type(config_dict) is dict: 89 raise ValueError('Received config dict of type "%s" (instead of dict)' 90 % type(config_dict)) 91 92 # First we pull out the 'stderr_writers' spec as a special case so 93 # that we can catch and properly route stderr output from 94 # parsing/creation of the other keyword args. 95 kwargs = {} 96 stderr_writers_spec = config_dict.get('stderr_writers') 97 if stderr_writers_spec: 98 stderr_writers = self._class_kwargs_from_config(stderr_writers_spec) 99 logging.getLogger().addHandler(StdErrLoggingHandler(stderr_writers)) 100 101 # We've already initialized the logger for stderr_writers, so 102 # *don't* pass that arg on, or things will get logged twice. 103 del config_dict['stderr_writers'] 104 105 for key, value in config_dict.items(): 106 # Declaration of readers, transforms and writers. Note that the 107 # singular "reader" is a special case for TimeoutReader that 108 # takes a single reader. 109 if key in ['readers', 'reader', 'transforms', 'writers', 'writer', 110 'mirror_to']: 111 if not value: 112 raise ValueError('declaration of "%s" in class has no kwargs?!?' % key) 113 kwargs[key] = self._class_kwargs_from_config(value) 114 115 # If value is a simple float/int/string/etc, just add to keywords 116 elif value is None or type(value) in [float, bool, int, str, list, dict]: 117 kwargs[key] = value 118 119 # Else what do we have? 120 else: 121 raise ValueError('unexpected key:value in configuration: ' 122 '{}: {}'.format(key, str(value))) 123 return kwargs 124 125 ############################ 126 def _class_kwargs_from_config(self, class_json): 127 """Parse a class's kwargs from a JSON string.""" 128 if not type(class_json) in [list, dict]: 129 raise ValueError('class_kwargs_from_config expected dict or list; ' 130 'got: "{}"'.format(class_json)) 131 132 # If we've got a list, recurse on each element 133 if type(class_json) is list: 134 return [self._class_kwargs_from_config(c) for c in class_json] 135 136 # Get name and constructor for component we're going to instantiate 137 class_name = class_json.get('class') 138 if class_name is None: 139 raise ValueError('missing "class" definition in "{}"'.format(class_json)) 140 141 # Are they telling us where the class definition is? If so import it 142 class_module_name = class_json.get('module') 143 if class_module_name is not None: 144 module = importlib.import_module(class_module_name) 145 class_const = getattr(module, class_name, None) 146 if not class_const: 147 raise ValueError('No component class "{}" found in module "{}"'.format( 148 class_name, class_module_name)) 149 else: 150 # If they haven't given us a 'module' declaration, assume class 151 # is something that's already defined. 152 class_const = globals().get(class_name) 153 if not class_const: 154 raise ValueError('No component class "{}" found: "{}"'.format( 155 class_name, class_json)) 156 157 # Get the keyword args for the component 158 kwarg_dict = class_json.get('kwargs', {}) 159 try: 160 kwargs = self._kwargs_from_config(kwarg_dict) 161 except (ValueError, RuntimeError) as e: 162 raise ValueError('Class "%s": %s' % (class_name, e)) 163 164 if not kwargs: 165 logging.debug('No kwargs found for component {}'.format(class_name)) 166 167 # Instantiate! 168 logging.debug('Instantiating {}({})'.format(class_name, kwargs)) 169 try: 170 component = class_const(**kwargs) 171 except (TypeError, ValueError, RuntimeError) as e: 172 raise ValueError('Class {}: {}\nClass definition: {}'.format( 173 class_name, e, pprint.pformat(class_json))) 174 return component 175 176 177################################################################################ 178class ListenerFromLoggerConfigString(ListenerFromLoggerConfig): 179 """Helper class for instantiating a Listener object from a JSON/YAML string""" 180 ############################ 181 182 def __init__(self, config_str, log_level=None): 183 """Create a Listener from a JSON config string.""" 184 config = read_config.parse(config_str) 185 config = read_config.expand_cruise_definition(config) 186 logging.info('Received config string: %s', pprint.pformat(config)) 187 super().__init__(config=config) 188 189 190################################################################################ 191class ListenerFromLoggerConfigFile(ListenerFromLoggerConfig): 192 """Helper class for instantiating a Listener object from a JSON config.""" 193 ############################ 194 195 def __init__(self, config_file, config_name=None, log_level=None): 196 """Create a Listener from a Python config file. If the file name 197 format is file_name:config, then assume the file_name is that of a 198 cruise definition, and look for the config itself under the 199 'configs:' key of the file's YAML. 200 """ 201 # If they've got a ':' in the config file name, then we're 202 # expecting them to also give us a config name to look for. 203 if config_file.find(':') > 0: 204 (config_file, config_name) = config_file.split(':', maxsplit=1) 205 config = read_config.read_config(config_file) 206 207 # If we're loading a single config from a cruise definition file, 208 # expand the file. 209 if config_name: 210 config = read_config.expand_cruise_definition(config) 211 212 config_dict = config.get('configs') 213 if not config_dict: 214 raise ValueError('Configuration name "%s" specified, but no ' 215 '"configs" section found in file "%s"' 216 % (config_name, config_file)) 217 config = config_dict.get(config_name) 218 if not config: 219 raise ValueError('Configuration name "%s" not found in file "%s"' 220 % (config_name, config_file)) 221 222 logging.info('Loaded config file: %s', pprint.pformat(config)) 223 super().__init__(config=config) 224 225 226################################################################################ 227if __name__ == '__main__': 228 parser = argparse.ArgumentParser( 229 epilog='Note that arguments are parsed and applied IN ORDER, so if you ' 230 'want a flag like --tail to be applied to a reader, or --slice_separator ' 231 'to be applied to --transform_slice, it must appear before that reader on ' 232 'the command line. Similarly, transforms will be added to the queue and ' 233 'applied in the order they appear on the command line; multiple ' 234 'specifications of a reader, writer or transform will result in multiple ' 235 'instances of it being created. Trust us, that\'s a feature.' 236 ) 237 238 ############################ 239 # Set up from config file 240 parser.add_argument('--config_file', dest='config_file', default=None, 241 help='Read Listener configuration from YAML/JSON file. ' 242 'If argument contains a colon, it will be interpreted ' 243 'as cruise_def_file_name:logger_config, and the script ' 244 'will look for a logger config name under the file\'s ' 245 '"configs" section. ' 246 'If specified, no other command line arguments (except ' 247 '-v) are allowed.') 248 249 parser.add_argument('--config_string', dest='config_string', default=None, 250 help='Read Listener configuration from YAML/JSON string. ' 251 'If specified, no other command line arguments (except ' 252 '-v) are allowed.') 253 254 # New positional argument for config file 255 parser.add_argument('config_file_positional', nargs='?', default=None, 256 help='Read Listener configuration from YAML/JSON file ' 257 '(Alternative to --config_file).') 258 259 ############################ 260 # Readers 261 parser.add_argument('--network', dest='network', default=None, 262 help='Comma-separated network addresses to read from. ' 263 'NOTE: This has been REPLACED by --udp and --tcp.') 264 265 parser.add_argument('--tcp', dest='tcp', default=None, 266 help='Comma-separated tcp address to read from, ' 267 'where an address is of format [source:]port[,...] and ' 268 'source, when provided, is the address of the ' 269 'interface you want to listen on. NOTE: This replaces ' 270 'the old --network argument.') 271 272 parser.add_argument('--udp', dest='udp', default=None, 273 help='Comma-separated udp addresses to read from, ' 274 'where an address is of format [source:]port[,...] and ' 275 'source, when provided, is either the address of the ' 276 'interface you want to listen on, or a multicast ' 277 'group. NOTE: This replaces the old --network argument.') 278 279 parser.add_argument('--database', dest='database', default=None, 280 help='Format: user@host:database:field1,field2,... ' 281 'Read specified fields from database. If no fields are ' 282 'specified, read all fields in database. Should ' 283 'be accompanied by the --database_password flag.') 284 285 parser.add_argument('--file', dest='file', default=None, 286 help='Comma-separated files to read from in parallel. ' 287 'Note that wildcards in a filename will be expanded, ' 288 'and the resulting files read sequentially. A single ' 289 'dash (\'-\') will be interpreted as stdout.') 290 291 parser.add_argument('--logfile', dest='logfile', default=None, 292 help='Comma-separated logfile base filenames to read ' 293 'from in parallel. Logfile dates will be added ' 294 'automatically.') 295 296 parser.add_argument('--logfile_use_timestamps', dest='logfile_use_timestamps', 297 action='store_true', default=False, 298 help='Make LogfileReaders deliver records at intervals ' 299 'corresponding to the intervals indicated by the stored ' 300 'record timestamps.') 301 302 parser.add_argument('--cached_data', dest='cached_data_server', default=None, 303 help='Read from cached data server with argument ' 304 'field_1,field2,...[@host:port]. Defaults to ' 305 'localhost:8766.') 306 307 parser.add_argument('--redis', dest='redis', default=None, 308 help='Redis pubsub channel[@host[:port]] to read from. ' 309 'Defaults to localhost:6379.') 310 311 parser.add_argument('--serial', dest='serial', default=None, 312 help='Comma-separated serial port spec containing at ' 313 'least port=[port], but also optionally baudrate, ' 314 'timeout, max_bytes and/or other SerialReader ' 315 'parameters.') 316 317 parser.add_argument('--interval', dest='interval', type=float, default=0, 318 help='Number of seconds between reads') 319 320 parser.add_argument('--tail', dest='tail', 321 action='store_true', default=False, help='Do not ' 322 'exit after reading file EOF; continue to check for ' 323 'additional input.') 324 325 parser.add_argument('--refresh_file_spec', dest='refresh_file_spec', 326 action='store_true', default=False, help='When at EOF ' 327 'and --tail is specified, check for new matching files ' 328 'that may have appeared since our last attempt to read.') 329 330 ############################ 331 # Transforms 332 parser.add_argument('--transform_prefix', dest='prefix', default='', 333 help='Prefix each record with this string') 334 335 parser.add_argument('--transform_nmea', dest='nmea', action='store_true', 336 default=False, help='Build NMEA-formatted sentence from ' 337 'data fields') 338 339 parser.add_argument('--transform_timestamp', dest='timestamp', 340 action='store_true', default=False, 341 help='Timestamp each record as it is read') 342 343 parser.add_argument('--transform_slice', dest='slice', default='', 344 help='Return only the specified (space-separated) ' 345 'fields of a text record. Can be comma-separated ' 346 'integer values and/or ranges, e.g. "1,3,5:7,-1". ' 347 'Note: zero-base indexing, so "1:" means "start at ' 348 'second element.') 349 350 parser.add_argument('--slice_separator', dest='slice_separator', default=None, 351 help='Field separator for --slice.') 352 353 parser.add_argument('--transform_regex_filter', dest='regex_filter', 354 default='', 355 help='Only pass records containing this regex.') 356 357 parser.add_argument('--transform_extract', dest='extract', 358 default='', help='Extract the named field from ' 359 'passed DASRecord or data dict.') 360 361 parser.add_argument('--transform_qc_filter', dest='qc_filter', 362 default='', help='Pass nothing unless the fields in the ' 363 'received DASRecord exceed comma-separated ' 364 '<field_name>:<lower>:<upper> bounds.') 365 366 parser.add_argument('--transform_parse_nmea', dest='parse_nmea', 367 action='store_true', default=False, 368 help='Convert tagged, timestamped NMEA records into ' 369 'Python DASRecords.') 370 parser.add_argument('--parse_nmea_message_path', 371 dest='parse_nmea_message_path', 372 default=nmea_parser.DEFAULT_MESSAGE_PATH, 373 help='Comma-separated globs of NMEA message definition ' 374 'file names, e.g. ' 375 'local/usap/message/*.yaml') 376 parser.add_argument('--parse_nmea_sensor_path', 377 dest='parse_nmea_sensor_path', 378 default=nmea_parser.DEFAULT_SENSOR_PATH, 379 help='Comma-separated globs of NMEA sensor definition ' 380 'file names, e.g. ' 381 'local/usap/sensor/*.yaml') 382 parser.add_argument('--parse_nmea_sensor_model_path', 383 dest='parse_nmea_sensor_model_path', 384 default=nmea_parser.DEFAULT_SENSOR_MODEL_PATH, 385 help='Comma-separated globs of NMEA sensor model ' 386 'definition file names, e.g. ' 387 'local/usap/sensor_model/*.yaml') 388 389 parser.add_argument('--transform_parse', dest='parse', 390 action='store_true', default=False, 391 help='Convert tagged, records into dict of values (or' 392 'JSON or DASRecords if --parse_to_json or ' 393 '--parse_to_das_record are specified).') 394 parser.add_argument('--parse_definition_path', 395 dest='parse_definition_path', 396 default=record_parser.DEFAULT_DEFINITION_PATH, 397 help='Comma-separated globs of device definition ' 398 'file names, e.g. ' 399 'local/usap/devices/*.yaml') 400 parser.add_argument('--parse_to_json', 401 dest='parse_to_json', action='store_true', 402 help='If specified, parser outputs JSON.') 403 parser.add_argument('--parse_to_das_record', 404 dest='parse_to_das_record', action='store_true', 405 help='If specified, parser outputs DASRecords.') 406 407 parser.add_argument('--time_format', dest='time_format', 408 default=timestamp.TIME_FORMAT, 409 help='Format in which to expect time strings.') 410 411 parser.add_argument('--transform_aggregate_xml', dest='aggregate_xml', 412 default='', help='Aggregate records of XML until a ' 413 'completed XML record whose outer element matches ' 414 'the specified tag has been seen, then pass it along ' 415 'as a single record.') 416 417 parser.add_argument('--transform_max_min', dest='max_min', 418 action='store_true', default=False, 419 help='Return only values that exceed the ' 420 'previously-seen max or min for a field, annotated by ' 421 'the name "field:max" or "field:min".') 422 423 parser.add_argument('--transform_count', dest='count', 424 action='store_true', default=False, 425 help='Return the count of number of times fields in ' 426 'the passed record have been seen, annotated by ' 427 'the name "field:count".') 428 429 parser.add_argument('--transform_to_json', dest='to_json', 430 action='store_true', default=False, 431 help='Convert the passed value to a JSON string') 432 433 parser.add_argument('--transform_to_json_pretty', dest='to_json_pretty', 434 action='store_true', default=False, 435 help='Convert the passed value to a pretty-printed ' 436 'JSON string') 437 438 parser.add_argument('--transform_from_json', dest='from_json', 439 action='store_true', default=False, 440 help='Convert the passed string, assumed to be JSON ' 441 'to a dict.') 442 443 parser.add_argument('--transform_from_json_to_das_record', 444 dest='from_json_to_das_record', 445 action='store_true', default=False, 446 help='Convert the passed string, assumed to be JSON ' 447 'to a DASRecord.') 448 449 parser.add_argument('--transform_to_das_record', dest='to_das_record', 450 default=None, help='Convert the passed value to a ' 451 'DASRecord with single field whose name is the string ' 452 'specified here.') 453 454 ############################ 455 # Writers 456 parser.add_argument('--write_file', dest='write_file', default=None, 457 help='File(s) to write to (\'-\' for stdout)') 458 459 parser.add_argument('--write_logfile', dest='write_logfile', default=None, 460 help='Filename base to write to. A date string that ' 461 'corresponds to the timestamped date of each record ' 462 'Will be appended to filename, with one file per date.') 463 464 parser.add_argument('--write_network', dest='write_network', default=None, 465 help='Network address(es) to write to. NOTE: This has ' 466 'been REPLACED by --write_udp and --write_tcp.') 467 468 parser.add_argument('--write_tcp', dest='write_tcp', default=None, 469 help='TCP destination host/IP(s) and port(s) to write ' 470 'to. Format destination:port[,...]. NOTE: This replaces ' 471 'the old --write_network argument.') 472 473 parser.add_argument('--write_udp', dest='write_udp', default=None, 474 help='UDP interface(s) and port(s) to write to. Format ' 475 '[destination:]port[,...]. NOTE: This replaces the old ' 476 '--write_network argument.') 477 478 parser.add_argument('--write_serial', dest='write_serial', default=None, 479 help='Comma-separated serial port spec containing at ' 480 'least port=[port], but also optionally baudrate, ' 481 'timeout, max_bytes and/or other SerialReader ' 482 'parameters.') 483 484 parser.add_argument('--network_eol', dest='network_eol', default=None, 485 help='Optional EOL string to add to writen records.') 486 487 parser.add_argument('--encoding', dest='encoding', default='utf-8', 488 help="Optional encoding of records. Default is utf-8, " 489 "specify '' for raw/binary. NOTE: This applies to ALL " 490 "readers/writers/transforms, as you need to have one " 491 "consistent encoding from start to finish.") 492 493 parser.add_argument('--write_redis', dest='write_redis', default=None, 494 help='Redis pubsub channel[@host[:port]] to write to. ' 495 'Defaults to localhost:6379.') 496 497 parser.add_argument('--write_record_screen', dest='write_record_screen', 498 action='store_true', default=False, 499 help='Display the most current DASRecord field values ' 500 'on the terminal.') 501 502 parser.add_argument('--write_database', dest='write_database', default=None, 503 help='user@host:database to write to. Should be ' 504 'accompanied by the --database_password flag.') 505 506 parser.add_argument('--database_password', dest='database_password', 507 default=None, help='Password for database specified by ' 508 '--write_database and/or --read_database.') 509 510 parser.add_argument('--write_cached_data_server', 511 dest='write_cached_data_server', default=None, 512 help='Write to a CachedDataServer at the specified ' 513 'host:port') 514 515 parser.add_argument('--check_format', dest='check_format', 516 action='store_true', default=False, help='Deprecated ') 517 518 parser.add_argument('-v', '--verbosity', dest='verbosity', 519 default=0, action='count', 520 help='Increase output verbosity') 521 522 parsed_args = parser.parse_args() 523 524 ############################ 525 # Set up logging before we do any other argument parsing (so that we 526 # can log problems with argument parsing). 527 528 LOG_LEVELS = {0: logging.WARNING, 1: logging.INFO, 2: logging.DEBUG} 529 log_level = LOG_LEVELS[min(parsed_args.verbosity, max(LOG_LEVELS))] 530 logging.getLogger().setLevel(log_level) 531 532 console_handler = logging.StreamHandler() 533 console_handler.setFormatter(STDERR_FORMATTER) 534 logging.root.handlers = [console_handler] 535 536 ############################ 537 # If --config_file/--config_string present, create Listener from 538 # config file/string. If not, manually parse and create from all 539 # other arguments on command line. 540 541 # 1. Resolve positional argument if present 542 if parsed_args.config_file_positional: 543 if parsed_args.config_file: 544 parser.error('You may not specify both a positional config file and --config_file') 545 parsed_args.config_file = parsed_args.config_file_positional 546 547 if parsed_args.config_file and parsed_args.config_string: 548 parser.error('You may not specify both --config_file and --config_string') 549 550 if parsed_args.config_file or parsed_args.config_string: 551 # Ensure that no other flags have been specified. 552 i = 1 553 while i < len(sys.argv): 554 if sys.argv[i] in ['-v', '--verbosity']: 555 i += 1 556 elif '--config_file'.find(sys.argv[i]) == 0: 557 i += 2 558 elif '--config_string'.find(sys.argv[i]) == 0: 559 i += 2 560 elif sys.argv[i] == parsed_args.config_file: 561 # If the positional argument matches the config file, skip it 562 i += 1 563 else: 564 parser.error('When --config_file or --config_string are ' 565 'specified, no other command line args except -v, ' 566 'may be used: ' 567 '{}'.format(sys.argv[i])) 568 569 # Read config file or JSON string and instantiate. 570 if parsed_args.config_file: 571 listener = ListenerFromLoggerConfigFile(parsed_args.config_file) 572 else: 573 listener = ListenerFromLoggerConfigString(parsed_args.config_string) 574 575 # If not --config, go parse all those crazy command line arguments manually 576 else: 577 ############################ 578 # Where we'll store our components 579 readers = [] 580 transforms = [] 581 writers = [] 582 stderr_writers = [] 583 584 ############################ 585 # Parse args out. We do this in a rather non-standard way to use the 586 # order of args on the command line to determine the order of our 587 # transforms. Specifically: break command line up into sections that 588 # end with the next '-'-prefixed argument (excluding the empty 589 # argument '-' and arguments starting with a negative number), 590 # and process those sections sequentially, adding 591 # them to the 'args' namespace as we go. 592 # 593 # So 594 # 595 # listen.py -v 1 2 3 -w -x - -y -4 -1,1 -z 596 # 597 # will be processed in five chunks: 598 # 599 # ['-v', '1', '2', '3'] 600 # ['-w'] 601 # ['-x', '-'] 602 # ['-y', '-4', '-1,1'] 603 # ['-z'] 604 # 605 # 606 # Functionally, it means that 607 # 608 # --transform_a <params_a> --transform_b <params_b> 609 # 610 # will push transform_a into the transform list before transform_b, 611 # (meaning it will be applied to records first), while 612 # 613 # --transform_b <params_b> --transform_a <params_a> 614 # 615 # will do the opposite. It also means that repeating a transform on 616 # the command line will apply it twice. Repetitions of readers or 617 # writers will create multiple instances but, since readers and 618 # writers are applied in parallel, ordering is irrelevant. 619 620 arg_start = arg_end = 1 # start at beginning of args, minus script name; 621 all_args = None # initial namespace is empty 622 623 # Loop while we have args left 624 while arg_end <= len(sys.argv): 625 626 arg_start = arg_end 627 arg_end += 1 628 629 # Get everything up to, but not including, the next arg beginning with '-' 630 # that isn't a plain '-' or something numeric. 631 while arg_end < len(sys.argv): 632 next_arg = sys.argv[arg_end] 633 if next_arg.find('-') == 0: 634 if next_arg != '-' and not re.match(r'^-\d', next_arg): # noqa: W605 635 break 636 arg_end += 1 637 638 # We have our next set of arguments - parse them 639 arg_list = sys.argv[arg_start:arg_end] 640 logging.debug('next set of command line arguments: %s', arg_list) 641 642 # These are just the new values 643 new_args = parser.parse_args(arg_list) 644 645 # We also want to accumulate old arguments so that we have access 646 # to flags that have been previously set. 647 all_args = parser.parse_args(arg_list, all_args) 648 649 logging.debug('namespace of all command-line args so far: %s', all_args) 650 651 ########################## 652 # Now go through new_args and see what they want us to do. Draw 653 # on all_args for the previously-set options that a reader, 654 # transform or writer might need. 655 656 ########################## 657 # Readers 658 if new_args.file: 659 for filename in new_args.file.split(','): 660 readers.append(TextFileReader( 661 file_spec=filename, tail=all_args.tail, 662 refresh_file_spec=all_args.refresh_file_spec)) 663 664 if new_args.network: 665 encoding = parsed_args.encoding 666 for addr in new_args.network.split(','): 667 readers.append(NetworkReader(network=addr, encoding=encoding)) 668 669 if new_args.tcp: 670 eol = all_args.network_eol 671 encoding = parsed_args.encoding 672 for addr_str in new_args.tcp.split(','): 673 addr = addr_str.split(':') 674 if len(addr) > 2: 675 parser.error('Format error for --tcp argument. Format ' 676 'should be [source:]port,[,...]') 677 if len(addr) < 2: 678 addr.insert(0, '') 679 source = addr[0] 680 port = int(addr[1]) 681 readers.append(TCPReader(source, port, eol=eol, encoding=encoding)) 682 683 if new_args.udp: 684 encoding = parsed_args.encoding 685 for addr_str in new_args.udp.split(','): 686 addr = addr_str.split(':') 687 if len(addr) > 2: 688 parser.error('Format error for --udp argument. Format ' 689 'should be [source:]port[,...]') 690 if len(addr) < 2: 691 addr.insert(0, '') 692 source = addr[0] 693 port = int(addr[1]) 694 readers.append(UDPReader(source, port, encoding=encoding)) 695 696 if new_args.redis: 697 for channel in new_args.redis.split(','): 698 readers.append(RedisReader(channel=channel)) 699 700 if new_args.logfile: 701 for filebase in new_args.logfile.split(','): 702 readers.append(LogfileReader( 703 filebase=filebase, use_timestamps=all_args.logfile_use_timestamps, 704 time_format=all_args.time_format, 705 refresh_file_spec=all_args.refresh_file_spec)) 706 707 if new_args.cached_data_server: 708 fields = new_args.cached_data_server 709 server = None 710 if fields.find('@') > 0: 711 fields, server = fields.split('@') 712 subscription = {'fields': {f: {'seconds': 0} for f in fields.split(',')}} 713 if server: 714 readers.append(CachedDataReader(subscription=subscription, 715 data_server=server)) 716 else: 717 readers.append(CachedDataReader(subscription=subscription)) 718 719 # For each comma-separated spec, parse out values for 720 # user@host:database:data_id[:message_type]. We count on 721 # --database_password having been specified somewhere. 722 if new_args.database: 723 password = all_args.database_password 724 (user, host_db) = new_args.database.split('@') 725 (host, database) = host_db.split(':', maxsplit=1) 726 if ':' in database: 727 (database, fields) = database.split(':') 728 else: 729 fields = None 730 readers.append(DatabaseReader(fields=fields, 731 database=database, host=host, 732 user=user, password=password)) 733 734 # SerialReader is a little more complicated than other readers 735 # because it can take so many parameters. Use the kwargs trick to 736 # pass them all in. 737 if new_args.serial: 738 kwargs = {} 739 for pair in new_args.serial.split(','): 740 (key, value) = pair.split('=') 741 kwargs[key] = value 742 readers.append(SerialReader(**kwargs)) 743 744 ########################## 745 # Transforms 746 if new_args.slice: 747 transforms.append(SliceTransform(new_args.slice, 748 all_args.slice_separator)) 749 if new_args.nmea: 750 transforms.append(NMEATransform(new_args.nmea)) 751 if new_args.timestamp: 752 transforms.append(TimestampTransform(time_format=all_args.time_format)) 753 if new_args.prefix: 754 transforms.append(PrefixTransform(new_args.prefix)) 755 if new_args.extract: 756 transforms.append(ExtractFieldTransform(new_args.extract)) 757 if new_args.regex_filter: 758 transforms.append(RegexFilterTransform(new_args.regex_filter)) 759 if new_args.qc_filter: 760 transforms.append(QCFilterTransform(new_args.qc_filter)) 761 if new_args.parse_nmea: 762 transforms.append( 763 ParseNMEATransform( 764 message_path=all_args.parse_nmea_message_path, 765 sensor_path=all_args.parse_nmea_sensor_path, 766 sensor_model_path=all_args.parse_nmea_sensor_model_path, 767 time_format=all_args.time_format) 768 ) 769 if new_args.parse: 770 transforms.append( 771 ParseTransform( 772 definition_path=all_args.parse_definition_path, 773 return_json=all_args.parse_to_json, 774 return_das_record=all_args.parse_to_das_record) 775 ) 776 if new_args.aggregate_xml: 777 transforms.append(XMLAggregatorTransform(new_args.aggregate_xml)) 778 779 if new_args.max_min: 780 transforms.append(MaxMinTransform()) 781 782 if new_args.count: 783 transforms.append(CountTransform()) 784 785 if new_args.to_json: 786 transforms.append(ToJSONTransform()) 787 788 if new_args.to_json_pretty: 789 transforms.append(ToJSONTransform(pretty=True)) 790 791 if new_args.from_json: 792 transforms.append(FromJSONTransform()) 793 794 if new_args.from_json_to_das_record: 795 transforms.append(FromJSONTransform(das_record=True)) 796 797 if new_args.to_das_record: 798 transforms.append( 799 ToDASRecordTransform(field_name=new_args.to_das_record)) 800 801 ########################## 802 # Writers 803 if new_args.write_file: 804 encoding = parsed_args.encoding 805 for filename in new_args.write_file.split(','): 806 if filename == '-': 807 filename = None 808 writers.append(FileWriter(filename=filename, encoding=encoding)) 809 810 if new_args.write_logfile: 811 writers.append(LogfileWriter(filebase=new_args.write_logfile)) 812 813 if new_args.write_network: 814 eol = all_args.network_eol 815 encoding = parsed_args.encoding 816 for addr in new_args.write_network.split(','): 817 writers.append(NetworkWriter(network=addr, eol=eol, encoding=encoding)) 818 819 if new_args.write_tcp: 820 eol = all_args.network_eol 821 encoding = parsed_args.encoding 822 for addr_str in new_args.write_tcp.split(','): 823 addr = addr_str.split(':') 824 if len(addr) > 2: 825 parser.error('Format err for --write_tcp argument. Format ' 826 'should be [destination:]port[,...]') 827 if len(addr) < 2: 828 addr.insert(0, '') 829 dest = addr[0] 830 port = int(addr[1]) 831 writers.append(TCPWriter(dest, port, eol=eol, encoding=encoding)) 832 833 if new_args.write_udp: 834 eol = all_args.network_eol 835 encoding = parsed_args.encoding 836 for addr_str in new_args.write_udp.split(','): 837 addr = addr_str.split(':') 838 if len(addr) > 2: 839 parser.error('Format error for --write_udp argument. Format ' 840 'should be [destination:]port[,...]') 841 if len(addr) < 2: 842 addr.insert(0, '') 843 dest = addr[0] 844 port = int(addr[1]) 845 writers.append(UDPWriter(dest, port, eol=eol, encoding=encoding)) 846 847 # SerialWriter is a little more complicated than other readers 848 # because it can take so many parameters. Use the kwargs trick to 849 # pass them all in. 850 if new_args.write_serial: 851 kwargs = {} 852 for pair in new_args.write_serial.split(','): 853 (key, value) = pair.split('=') 854 kwargs[key] = value 855 writers.append(SerialWriter(**kwargs)) 856 857 if new_args.write_redis: 858 for channel in new_args.write_redis.split(','): 859 writers.append(RedisWriter(channel=channel)) 860 861 if new_args.write_record_screen: 862 writers.append(RecordScreenWriter()) 863 864 if new_args.write_database: 865 password = all_args.database_password 866 # Parse out values for user@host:database. We count on 867 # --database_password having been specified somewhere. 868 (user, host_db) = new_args.write_database.split('@') 869 (host, database) = host_db.split(':') 870 writers.append(DatabaseWriter(database=database, host=host, 871 user=user, password=password)) 872 873 if new_args.write_cached_data_server: 874 data_server = new_args.write_cached_data_server 875 writers.append(CachedDataWriter(data_server=data_server)) 876 877 if all_args.check_format: 878 logging.warning('Argument --check_format is deprecated and no longer ' 879 'serves any function.') 880 ########################## 881 # If we don't have any readers, read from stdin, if we don't have 882 # any writers, write to stdout. 883 if not readers: 884 readers.append(TextFileReader()) 885 if not writers: 886 writers.append(FileWriter()) 887 888 ########################## 889 # Now that we've got our readers, transforms and writers defined, 890 # create the Listener. 891 listener = Listener(readers=readers, transforms=transforms, writers=writers, 892 stderr_writers=stderr_writers, 893 interval=all_args.interval) 894 895 ############################ 896 # Whichever way we created the listener, run it. 897 listener.run()
class
ListenerFromLoggerConfig(logger.listener.listener.Listener):
59class ListenerFromLoggerConfig(Listener): 60 """Helper class for instantiating a Listener object from a Python dict.""" 61 ############################ 62 63 def __init__(self, config, log_level=None): 64 """Create a Listener from a Python config dict.""" 65 66 if not type(config) is dict: 67 raise ValueError('ListenerFromLoggerConfig expects config of type ' 68 '"dict" but received one of type "%s": %s' 69 % (type(config), str(config))) 70 71 # Extract keyword args from config and instantiate. 72 logging.debug('ListenerFromLoggerConfig instantiating logger ' 73 'config: %s', pprint.pformat(config)) 74 try: 75 kwargs = self._kwargs_from_config(config) 76 except ValueError as e: 77 config_name = config.get('name', 'unknown logger') 78 raise ValueError('Config for %s: %s' % (config_name, e)) 79 80 super().__init__(**kwargs) 81 82 ############################ 83 def _kwargs_from_config(self, config_dict): 84 """Parse a kwargs from a JSON string, making exceptions for keywords 85 'readers', 'transforms', and 'writers' as internal class references.""" 86 if not config_dict: 87 return {} 88 89 if not type(config_dict) is dict: 90 raise ValueError('Received config dict of type "%s" (instead of dict)' 91 % type(config_dict)) 92 93 # First we pull out the 'stderr_writers' spec as a special case so 94 # that we can catch and properly route stderr output from 95 # parsing/creation of the other keyword args. 96 kwargs = {} 97 stderr_writers_spec = config_dict.get('stderr_writers') 98 if stderr_writers_spec: 99 stderr_writers = self._class_kwargs_from_config(stderr_writers_spec) 100 logging.getLogger().addHandler(StdErrLoggingHandler(stderr_writers)) 101 102 # We've already initialized the logger for stderr_writers, so 103 # *don't* pass that arg on, or things will get logged twice. 104 del config_dict['stderr_writers'] 105 106 for key, value in config_dict.items(): 107 # Declaration of readers, transforms and writers. Note that the 108 # singular "reader" is a special case for TimeoutReader that 109 # takes a single reader. 110 if key in ['readers', 'reader', 'transforms', 'writers', 'writer', 111 'mirror_to']: 112 if not value: 113 raise ValueError('declaration of "%s" in class has no kwargs?!?' % key) 114 kwargs[key] = self._class_kwargs_from_config(value) 115 116 # If value is a simple float/int/string/etc, just add to keywords 117 elif value is None or type(value) in [float, bool, int, str, list, dict]: 118 kwargs[key] = value 119 120 # Else what do we have? 121 else: 122 raise ValueError('unexpected key:value in configuration: ' 123 '{}: {}'.format(key, str(value))) 124 return kwargs 125 126 ############################ 127 def _class_kwargs_from_config(self, class_json): 128 """Parse a class's kwargs from a JSON string.""" 129 if not type(class_json) in [list, dict]: 130 raise ValueError('class_kwargs_from_config expected dict or list; ' 131 'got: "{}"'.format(class_json)) 132 133 # If we've got a list, recurse on each element 134 if type(class_json) is list: 135 return [self._class_kwargs_from_config(c) for c in class_json] 136 137 # Get name and constructor for component we're going to instantiate 138 class_name = class_json.get('class') 139 if class_name is None: 140 raise ValueError('missing "class" definition in "{}"'.format(class_json)) 141 142 # Are they telling us where the class definition is? If so import it 143 class_module_name = class_json.get('module') 144 if class_module_name is not None: 145 module = importlib.import_module(class_module_name) 146 class_const = getattr(module, class_name, None) 147 if not class_const: 148 raise ValueError('No component class "{}" found in module "{}"'.format( 149 class_name, class_module_name)) 150 else: 151 # If they haven't given us a 'module' declaration, assume class 152 # is something that's already defined. 153 class_const = globals().get(class_name) 154 if not class_const: 155 raise ValueError('No component class "{}" found: "{}"'.format( 156 class_name, class_json)) 157 158 # Get the keyword args for the component 159 kwarg_dict = class_json.get('kwargs', {}) 160 try: 161 kwargs = self._kwargs_from_config(kwarg_dict) 162 except (ValueError, RuntimeError) as e: 163 raise ValueError('Class "%s": %s' % (class_name, e)) 164 165 if not kwargs: 166 logging.debug('No kwargs found for component {}'.format(class_name)) 167 168 # Instantiate! 169 logging.debug('Instantiating {}({})'.format(class_name, kwargs)) 170 try: 171 component = class_const(**kwargs) 172 except (TypeError, ValueError, RuntimeError) as e: 173 raise ValueError('Class {}: {}\nClass definition: {}'.format( 174 class_name, e, pprint.pformat(class_json))) 175 return component
Helper class for instantiating a Listener object from a Python dict.
ListenerFromLoggerConfig(config, log_level=None)
63 def __init__(self, config, log_level=None): 64 """Create a Listener from a Python config dict.""" 65 66 if not type(config) is dict: 67 raise ValueError('ListenerFromLoggerConfig expects config of type ' 68 '"dict" but received one of type "%s": %s' 69 % (type(config), str(config))) 70 71 # Extract keyword args from config and instantiate. 72 logging.debug('ListenerFromLoggerConfig instantiating logger ' 73 'config: %s', pprint.pformat(config)) 74 try: 75 kwargs = self._kwargs_from_config(config) 76 except ValueError as e: 77 config_name = config.get('name', 'unknown logger') 78 raise ValueError('Config for %s: %s' % (config_name, e)) 79 80 super().__init__(**kwargs)
Create a Listener from a Python config dict.
179class ListenerFromLoggerConfigString(ListenerFromLoggerConfig): 180 """Helper class for instantiating a Listener object from a JSON/YAML string""" 181 ############################ 182 183 def __init__(self, config_str, log_level=None): 184 """Create a Listener from a JSON config string.""" 185 config = read_config.parse(config_str) 186 config = read_config.expand_cruise_definition(config) 187 logging.info('Received config string: %s', pprint.pformat(config)) 188 super().__init__(config=config)
Helper class for instantiating a Listener object from a JSON/YAML string
ListenerFromLoggerConfigString(config_str, log_level=None)
183 def __init__(self, config_str, log_level=None): 184 """Create a Listener from a JSON config string.""" 185 config = read_config.parse(config_str) 186 config = read_config.expand_cruise_definition(config) 187 logging.info('Received config string: %s', pprint.pformat(config)) 188 super().__init__(config=config)
Create a Listener from a JSON config string.
192class ListenerFromLoggerConfigFile(ListenerFromLoggerConfig): 193 """Helper class for instantiating a Listener object from a JSON config.""" 194 ############################ 195 196 def __init__(self, config_file, config_name=None, log_level=None): 197 """Create a Listener from a Python config file. If the file name 198 format is file_name:config, then assume the file_name is that of a 199 cruise definition, and look for the config itself under the 200 'configs:' key of the file's YAML. 201 """ 202 # If they've got a ':' in the config file name, then we're 203 # expecting them to also give us a config name to look for. 204 if config_file.find(':') > 0: 205 (config_file, config_name) = config_file.split(':', maxsplit=1) 206 config = read_config.read_config(config_file) 207 208 # If we're loading a single config from a cruise definition file, 209 # expand the file. 210 if config_name: 211 config = read_config.expand_cruise_definition(config) 212 213 config_dict = config.get('configs') 214 if not config_dict: 215 raise ValueError('Configuration name "%s" specified, but no ' 216 '"configs" section found in file "%s"' 217 % (config_name, config_file)) 218 config = config_dict.get(config_name) 219 if not config: 220 raise ValueError('Configuration name "%s" not found in file "%s"' 221 % (config_name, config_file)) 222 223 logging.info('Loaded config file: %s', pprint.pformat(config)) 224 super().__init__(config=config)
Helper class for instantiating a Listener object from a JSON config.
ListenerFromLoggerConfigFile(config_file, config_name=None, log_level=None)
196 def __init__(self, config_file, config_name=None, log_level=None): 197 """Create a Listener from a Python config file. If the file name 198 format is file_name:config, then assume the file_name is that of a 199 cruise definition, and look for the config itself under the 200 'configs:' key of the file's YAML. 201 """ 202 # If they've got a ':' in the config file name, then we're 203 # expecting them to also give us a config name to look for. 204 if config_file.find(':') > 0: 205 (config_file, config_name) = config_file.split(':', maxsplit=1) 206 config = read_config.read_config(config_file) 207 208 # If we're loading a single config from a cruise definition file, 209 # expand the file. 210 if config_name: 211 config = read_config.expand_cruise_definition(config) 212 213 config_dict = config.get('configs') 214 if not config_dict: 215 raise ValueError('Configuration name "%s" specified, but no ' 216 '"configs" section found in file "%s"' 217 % (config_name, config_file)) 218 config = config_dict.get(config_name) 219 if not config: 220 raise ValueError('Configuration name "%s" not found in file "%s"' 221 % (config_name, config_file)) 222 223 logging.info('Loaded config file: %s', pprint.pformat(config)) 224 super().__init__(config=config)
Create a Listener from a Python config file. If the file name format is file_name:config, then assume the file_name is that of a cruise definition, and look for the config itself under the 'configs:' key of the file's YAML.