openrvdas.logger.utils.validate_config

  1#!/usr/bin/env python3
  2"""Validation tool for OpenRVDAS configuration files.
  3
  4Validates three types of YAML configuration files:
  51. Device/device_type definitions (contrib/devices/*.yaml, logger/devices/*.yaml)
  62. Logger configurations (individual logger configs)
  73. Cruise definitions (cruise config files with loggers, modes, etc.)
  8
  9Usage:
 10    # Validate a single file
 11    python logger/utils/validate_config.py path/to/config.yaml
 12
 13    # Validate multiple files
 14    python logger/utils/validate_config.py file1.yaml file2.yaml
 15
 16    # Validate with verbose output
 17    python logger/utils/validate_config.py -v path/to/config.yaml
 18
 19    # Validate all device definitions
 20    python logger/utils/validate_config.py contrib/devices/*.yaml
 21"""
 22import argparse
 23import glob
 24import os
 25import sys
 26from typing import Dict, List, Any, Tuple
 27
 28try:
 29    import yaml
 30except ImportError:
 31    print("Error: PyYAML is required. Install with: pip install pyyaml")
 32    sys.exit(1)
 33
 34
 35class ValidationError:
 36    """Represents a single validation error."""
 37
 38    def __init__(self, message: str, path: str = "", severity: str = "error"):
 39        self.message = message
 40        self.path = path  # e.g., "loggers.PCOD.configs"
 41        self.severity = severity  # "error" or "warning"
 42
 43    def __str__(self):
 44        if self.path:
 45            return f"[{self.severity.upper()}] {self.path}: {self.message}"
 46        return f"[{self.severity.upper()}] {self.message}"
 47
 48
 49class ConfigValidator:
 50    """Validates OpenRVDAS configuration files."""
 51
 52    # Known reader, transform, and writer classes
 53    KNOWN_READERS = {
 54        'CachedDataReader', 'ComposedReader', 'DatabaseReader', 'HttpReader',
 55        'LogfileReader', 'ModbusReader', 'ModbusSerialReader', 'MQTTReader',
 56        'NetworkReader', 'PolledSerialReader', 'RedisReader', 'SealogReader',
 57        'SerialReader', 'SocketReader', 'TCPReader', 'TextFileReader',
 58        'TimeoutReader', 'UDPReader'
 59    }
 60
 61    KNOWN_TRANSFORMS = {
 62        'ConvertFieldsTransform', 'CountTransform', 'DeltaTransform',
 63        'DerivedDataTransform', 'ExtractFieldTransform', 'FormatTransform',
 64        'FromJsonTransform', 'GeofenceTransform', 'InterpolationTransform',
 65        'MaxMinTransform', 'ModifyValueTransform', 'NMEAChecksumTransform',
 66        'NMEATransform', 'ParseNMEATransform', 'ParseTransform',
 67        'PrefixTransform', 'QCFilterTransform', 'RegexFilterTransform',
 68        'RegexReplaceTransform', 'RegexParseTransform', 'SelectFieldsTransform',
 69        'SliceTransform', 'SplitTransform', 'SubsampleTransform',
 70        'TimestampTransform', 'ToJsonTransform', 'ToDASRecordTransform',
 71        'TrueWindsTransform', 'XMLAggregatorTransform'
 72    }
 73
 74    KNOWN_WRITERS = {
 75        'CachedDataWriter', 'ComposedWriter', 'DatabaseWriter', 'EmailWriter',
 76        'FileWriter', 'GoogleSheetsWriter', 'GrafanaLiveWriter',
 77        'InfluxDBWriter', 'LogfileWriter', 'LoggerManagerWriter', 'MQTTWriter',
 78        'NetworkWriter', 'RecordScreenWriter', 'RedisWriter',
 79        'RegexLogfileWriter', 'SealogWriter', 'SerialWriter', 'SocketWriter',
 80        'TCPWriter', 'TextFileWriter', 'UDPWriter'
 81    }
 82
 83    def __init__(self, verbose: bool = False):
 84        self.verbose = verbose
 85        self.errors: List[ValidationError] = []
 86
 87    def validate_file(self, file_path: str) -> Tuple[bool, List[ValidationError]]:
 88        """
 89        Validate a configuration file.
 90
 91        Returns:
 92            Tuple of (is_valid, list of errors/warnings)
 93        """
 94        self.errors = []
 95
 96        # Check file exists
 97        if not os.path.isfile(file_path):
 98            self.errors.append(ValidationError(f"File not found: {file_path}"))
 99            return False, self.errors
100
101        # Try to parse YAML
102        try:
103            with open(file_path, 'r') as f:
104                content = f.read()
105        except Exception as e:
106            self.errors.append(ValidationError(f"Cannot read file: {e}"))
107            return False, self.errors
108
109        try:
110            data = yaml.safe_load(content)
111        except yaml.YAMLError as e:
112            # Extract useful info from YAML error
113            error_msg = self._format_yaml_error(e)
114            self.errors.append(ValidationError(f"Invalid YAML: {error_msg}"))
115            return False, self.errors
116
117        if data is None:
118            self.errors.append(ValidationError("File is empty"))
119            return False, self.errors
120
121        if not isinstance(data, dict):
122            self.errors.append(ValidationError(
123                f"Expected YAML dictionary, got {type(data).__name__}"))
124            return False, self.errors
125
126        # Detect file type and validate accordingly
127        file_type = self._detect_file_type(data)
128
129        if self.verbose:
130            print(f"  Detected type: {file_type}")
131
132        if file_type == "device_definitions":
133            self._validate_device_definitions(data)
134        elif file_type == "logger_config":
135            self._validate_logger_config(data)
136        elif file_type == "cruise_definition":
137            self._validate_cruise_definition(data)
138        elif file_type == "logger_template":
139            self._validate_logger_template(data)
140        elif file_type == "config_template":
141            self._validate_config_template(data)
142        else:
143            self.errors.append(ValidationError(
144                f"Unknown file type. Expected device definitions, logger config, "
145                f"or cruise definition.", severity="warning"))
146
147        is_valid = not any(e.severity == "error" for e in self.errors)
148        return is_valid, self.errors
149
150    def _format_yaml_error(self, e: yaml.YAMLError) -> str:
151        """Format a YAML error into a readable message."""
152        if hasattr(e, 'problem_mark'):
153            mark = e.problem_mark
154            return f"line {mark.line + 1}, column {mark.column + 1}: {e.problem}"
155        return str(e)
156
157    def _detect_file_type(self, data: Dict) -> str:
158        """Detect the type of configuration file."""
159        # Check for cruise definition markers
160        if 'cruise' in data or 'loggers' in data and 'modes' in data:
161            return "cruise_definition"
162
163        # Check for logger template
164        if 'logger_templates' in data:
165            return "logger_template"
166
167        # Check for config template
168        if 'config_templates' in data:
169            return "config_template"
170
171        # Check for device/device_type definitions
172        has_device_defs = False
173        for key, value in data.items():
174            if isinstance(value, dict):
175                category = value.get('category')
176                if category in ('device', 'device_type'):
177                    has_device_defs = True
178                    break
179
180        if has_device_defs:
181            return "device_definitions"
182
183        # Check for logger config (has readers/writers at top level or nested)
184        if 'readers' in data or 'writers' in data:
185            return "logger_config"
186
187        # Check if it's a dict of logger configs
188        for key, value in data.items():
189            if isinstance(value, dict):
190                if 'readers' in value or 'writers' in value:
191                    return "logger_config"
192
193        return "unknown"
194
195    def _validate_device_definitions(self, data: Dict):
196        """Validate device and device_type definitions."""
197        for name, definition in data.items():
198            if not isinstance(definition, dict):
199                self.errors.append(ValidationError(
200                    f"Expected dictionary for definition",
201                    path=name))
202                continue
203
204            category = definition.get('category')
205            if category not in ('device', 'device_type'):
206                self.errors.append(ValidationError(
207                    f"Missing or invalid 'category' (expected 'device' or 'device_type')",
208                    path=name))
209                continue
210
211            if category == 'device_type':
212                self._validate_device_type(name, definition)
213            else:
214                self._validate_device(name, definition)
215
216    def _validate_device_type(self, name: str, definition: Dict):
217        """Validate a device_type definition."""
218        # Must have format
219        if 'format' not in definition:
220            self.errors.append(ValidationError(
221                "Missing required 'format' key",
222                path=name))
223            return
224
225        fmt = definition['format']
226
227        # Format can be string, list, or dict
228        if isinstance(fmt, str):
229            self._validate_format_string(fmt, f"{name}.format")
230        elif isinstance(fmt, list):
231            for i, f in enumerate(fmt):
232                if not isinstance(f, str):
233                    self.errors.append(ValidationError(
234                        f"Format list item {i} must be a string",
235                        path=f"{name}.format"))
236                else:
237                    self._validate_format_string(f, f"{name}.format[{i}]")
238        elif isinstance(fmt, dict):
239            for msg_type, f in fmt.items():
240                # Value can be a string or list of strings (alternatives)
241                if isinstance(f, str):
242                    self._validate_format_string(f, f"{name}.format.{msg_type}")
243                elif isinstance(f, list):
244                    for i, alt in enumerate(f):
245                        if isinstance(alt, str):
246                            self._validate_format_string(
247                                alt, f"{name}.format.{msg_type}[{i}]")
248                        else:
249                            self.errors.append(ValidationError(
250                                f"Format alternative must be a string",
251                                path=f"{name}.format.{msg_type}[{i}]"))
252                else:
253                    self.errors.append(ValidationError(
254                        f"Format for '{msg_type}' must be string or list of strings",
255                        path=f"{name}.format"))
256        else:
257            self.errors.append(ValidationError(
258                f"'format' must be string, list, or dict, got {type(fmt).__name__}",
259                path=name))
260
261        # Optional: validate fields if present
262        fields = definition.get('fields')
263        if fields is not None and not isinstance(fields, dict):
264            self.errors.append(ValidationError(
265                f"'fields' must be a dictionary",
266                path=name))
267
268    def _validate_format_string(self, fmt: str, path: str):
269        """Validate a parse format string."""
270        # Check for balanced braces
271        depth = 0
272        for char in fmt:
273            if char == '{':
274                depth += 1
275            elif char == '}':
276                depth -= 1
277            if depth < 0:
278                self.errors.append(ValidationError(
279                    "Unbalanced braces in format string",
280                    path=path))
281                return
282        if depth != 0:
283            self.errors.append(ValidationError(
284                "Unbalanced braces in format string",
285                path=path))
286
287    def _validate_device(self, name: str, definition: Dict):
288        """Validate a device definition."""
289        # Must have device_type
290        if 'device_type' not in definition:
291            self.errors.append(ValidationError(
292                "Missing required 'device_type' key",
293                path=name))
294
295        # Should have fields mapping
296        if 'fields' not in definition:
297            self.errors.append(ValidationError(
298                "Missing 'fields' mapping (device to device_type field names)",
299                path=name,
300                severity="warning"))
301
302    def _validate_logger_config(self, data: Dict):
303        """Validate a logger configuration."""
304        # Could be a single config or dict of configs
305        if 'readers' in data or 'writers' in data:
306            self._validate_single_logger_config(data, "")
307        else:
308            # Dict of named configs
309            for config_name, config in data.items():
310                if isinstance(config, dict):
311                    self._validate_single_logger_config(config, config_name)
312
313    def _validate_single_logger_config(self, config: Dict, path: str,
314                                        allow_empty: bool = True):
315        """Validate a single logger configuration."""
316        # Empty config is valid (represents "off" state)
317        if not config:
318            return
319
320        # Should have readers and writers
321        has_readers = 'readers' in config
322        has_writers = 'writers' in config
323
324        if not has_readers and not has_writers:
325            self.errors.append(ValidationError(
326                "Config should have 'readers' and/or 'writers'",
327                path=path,
328                severity="warning"))
329            return
330
331        if not has_readers:
332            self.errors.append(ValidationError(
333                "Config is missing 'readers'",
334                path=path,
335                severity="warning"))
336
337        if not has_writers:
338            self.errors.append(ValidationError(
339                "Config is missing 'writers'",
340                path=path,
341                severity="warning"))
342
343        # Validate readers
344        if has_readers:
345            self._validate_components(config['readers'], 'reader',
346                                      f"{path}.readers" if path else "readers")
347
348        # Validate transforms if present
349        if 'transforms' in config:
350            self._validate_components(config['transforms'], 'transform',
351                                      f"{path}.transforms" if path else "transforms")
352
353        # Validate writers
354        if has_writers:
355            self._validate_components(config['writers'], 'writer',
356                                      f"{path}.writers" if path else "writers")
357
358    def _validate_components(self, components: Any, component_type: str, path: str):
359        """Validate reader/transform/writer components."""
360        if components is None:
361            return
362
363        # Can be a single component or list
364        if isinstance(components, dict):
365            components = [components]
366        elif not isinstance(components, list):
367            self.errors.append(ValidationError(
368                f"Expected dict or list for {component_type}s",
369                path=path))
370            return
371
372        known_classes = {
373            'reader': self.KNOWN_READERS,
374            'transform': self.KNOWN_TRANSFORMS,
375            'writer': self.KNOWN_WRITERS
376        }[component_type]
377
378        for i, comp in enumerate(components):
379            comp_path = f"{path}[{i}]" if len(components) > 1 else path
380
381            if not isinstance(comp, dict):
382                self.errors.append(ValidationError(
383                    f"Component must be a dictionary",
384                    path=comp_path))
385                continue
386
387            # Must have 'class' key
388            if 'class' not in comp:
389                self.errors.append(ValidationError(
390                    f"Missing 'class' key",
391                    path=comp_path))
392                continue
393
394            class_name = comp['class']
395
396            # Check if it's a known class (skip if it contains variables)
397            if '<<' not in str(class_name) and class_name not in known_classes:
398                self.errors.append(ValidationError(
399                    f"Unknown {component_type} class: '{class_name}'",
400                    path=comp_path,
401                    severity="warning"))
402
403            # Validate kwargs if present
404            kwargs = comp.get('kwargs')
405            if kwargs is not None and not isinstance(kwargs, dict):
406                self.errors.append(ValidationError(
407                    f"'kwargs' must be a dictionary",
408                    path=comp_path))
409
410    def _validate_cruise_definition(self, data: Dict):
411        """Validate a cruise definition file."""
412        # Check for cruise info
413        if 'cruise' in data:
414            cruise = data['cruise']
415            if isinstance(cruise, dict):
416                if 'id' not in cruise:
417                    self.errors.append(ValidationError(
418                        "Missing 'id' in cruise section",
419                        path="cruise",
420                        severity="warning"))
421            else:
422                self.errors.append(ValidationError(
423                    "'cruise' should be a dictionary with 'id', 'start', 'end'",
424                    path="cruise"))
425
426        # Must have loggers
427        if 'loggers' not in data:
428            self.errors.append(ValidationError(
429                "Missing required 'loggers' section"))
430            return
431
432        loggers = data['loggers']
433        if not isinstance(loggers, dict):
434            self.errors.append(ValidationError(
435                "'loggers' must be a dictionary",
436                path="loggers"))
437            return
438
439        # Collect all config names declared by loggers and defined in configs
440        declared_configs = set()  # configs referenced by loggers
441        defined_configs = set()   # configs defined in top-level 'configs'
442        template_configs = set()  # configs that will be generated from templates
443        logger_names = set(loggers.keys())
444        uses_templates = False
445        uses_includes = 'includes' in data
446
447        # Get top-level configs if present
448        top_level_configs = data.get('configs', {})
449        if isinstance(top_level_configs, dict):
450            defined_configs = set(top_level_configs.keys())
451
452        # Get logger templates if present (for inferring generated config names)
453        logger_templates = data.get('logger_templates', {})
454
455        # Validate each logger
456        for logger_name, logger_def in loggers.items():
457            if not isinstance(logger_def, dict):
458                self.errors.append(ValidationError(
459                    "Logger definition must be a dictionary",
460                    path=f"loggers.{logger_name}"))
461                continue
462
463            # Logger must have configs or logger_template
464            has_configs = 'configs' in logger_def
465            has_template = 'logger_template' in logger_def
466
467            if not has_configs and not has_template:
468                self.errors.append(ValidationError(
469                    "Logger must have 'configs' or 'logger_template'",
470                    path=f"loggers.{logger_name}"))
471                continue
472
473            # If using template, infer config names from template
474            if has_template:
475                uses_templates = True
476                template_name = logger_def['logger_template']
477                template = logger_templates.get(template_name, {})
478                template_config_keys = template.get('configs', {})
479                if isinstance(template_config_keys, dict):
480                    for config_key in template_config_keys.keys():
481                        # Template configs get named as logger-configkey
482                        template_configs.add(f"{logger_name}-{config_key}")
483
484            # Track declared configs
485            if has_configs:
486                logger_configs = logger_def['configs']
487                if isinstance(logger_configs, list):
488                    # List of config names (references)
489                    for config_name in logger_configs:
490                        declared_configs.add(config_name)
491                elif isinstance(logger_configs, dict):
492                    # Dict of inline config definitions
493                    for config_key, config_def in logger_configs.items():
494                        full_name = f"{logger_name}-{config_key}"
495                        declared_configs.add(full_name)
496                        defined_configs.add(full_name)
497                        # Validate inline config
498                        if isinstance(config_def, dict):
499                            self._validate_single_logger_config(
500                                config_def, f"loggers.{logger_name}.configs.{config_key}")
501
502        # Check for configs declared but not defined (skip if using templates)
503        if not uses_templates:
504            undefined_configs = declared_configs - defined_configs
505            for config_name in sorted(undefined_configs):
506                self.errors.append(ValidationError(
507                    f"Config '{config_name}' is referenced but not defined",
508                    path="configs",
509                    severity="warning"))
510
511            # Check for extraneous configs not used by any logger
512            unused_configs = defined_configs - declared_configs
513            for config_name in sorted(unused_configs):
514                self.errors.append(ValidationError(
515                    f"Config '{config_name}' is defined but not used by any logger",
516                    path="configs",
517                    severity="warning"))
518
519        # Validate top-level config definitions
520        for config_name, config_def in top_level_configs.items():
521            if isinstance(config_def, dict):
522                self._validate_single_logger_config(
523                    config_def, f"configs.{config_name}")
524
525        # Check modes if present
526        if 'modes' in data:
527            modes = data['modes']
528            if not isinstance(modes, dict):
529                self.errors.append(ValidationError(
530                    "'modes' must be a dictionary",
531                    path="modes"))
532            elif not modes:
533                self.errors.append(ValidationError(
534                    "'modes' is empty",
535                    path="modes",
536                    severity="warning"))
537            else:
538                # Validate each mode
539                # Include template-generated configs as valid
540                all_valid_configs = declared_configs | defined_configs | template_configs
541
542                # Skip detailed mode validation if using includes with templates
543                # (configs will be generated at runtime from included templates)
544                skip_config_check = uses_includes and uses_templates
545
546                for mode_name, mode_def in modes.items():
547                    if isinstance(mode_def, dict):
548                        # Dict mapping logger -> config
549                        for logger, config in mode_def.items():
550                            if logger not in logger_names:
551                                self.errors.append(ValidationError(
552                                    f"Unknown logger '{logger}'",
553                                    path=f"modes.{mode_name}",
554                                    severity="warning"))
555                            if not skip_config_check and config not in all_valid_configs:
556                                self.errors.append(ValidationError(
557                                    f"Unknown config '{config}' for logger '{logger}'",
558                                    path=f"modes.{mode_name}",
559                                    severity="warning"))
560                        # Check if all loggers are covered
561                        missing_loggers = logger_names - set(mode_def.keys())
562                        for logger in sorted(missing_loggers):
563                            self.errors.append(ValidationError(
564                                f"Logger '{logger}' has no config in this mode",
565                                path=f"modes.{mode_name}",
566                                severity="warning"))
567                    elif isinstance(mode_def, list):
568                        # List of config names
569                        if not skip_config_check:
570                            for config in mode_def:
571                                if config not in all_valid_configs:
572                                    self.errors.append(ValidationError(
573                                        f"Unknown config '{config}'",
574                                        path=f"modes.{mode_name}",
575                                        severity="warning"))
576
577        # Check for default_mode if modes exist
578        if 'modes' in data and 'default_mode' not in data:
579            self.errors.append(ValidationError(
580                "No 'default_mode' specified (first mode will be used)",
581                path="modes",
582                severity="warning"))
583
584    def _validate_logger_template(self, data: Dict):
585        """Validate a logger template file."""
586        templates = data.get('logger_templates', {})
587        if not isinstance(templates, dict):
588            self.errors.append(ValidationError(
589                "'logger_templates' must be a dictionary"))
590            return
591
592        for name, template in templates.items():
593            if not isinstance(template, dict):
594                self.errors.append(ValidationError(
595                    "Template must be a dictionary",
596                    path=f"logger_templates.{name}"))
597                continue
598
599            # Template should have configs
600            if 'configs' not in template:
601                self.errors.append(ValidationError(
602                    "Template should have 'configs'",
603                    path=f"logger_templates.{name}",
604                    severity="warning"))
605
606    def _validate_config_template(self, data: Dict):
607        """Validate a config template file."""
608        templates = data.get('config_templates', {})
609        if not isinstance(templates, dict):
610            self.errors.append(ValidationError(
611                "'config_templates' must be a dictionary"))
612            return
613
614        for name, template in templates.items():
615            if not isinstance(template, dict):
616                self.errors.append(ValidationError(
617                    "Template must be a dictionary",
618                    path=f"config_templates.{name}"))
619
620
621def validate(file_path: str) -> Tuple[bool, str]:
622    """
623    Validate a configuration file and return a simple result.
624
625    This is a convenience function for programmatic use (e.g., from listen.py
626    or the Django GUI).
627
628    Args:
629        file_path: Path to the configuration file to validate
630
631    Returns:
632        Tuple of (is_valid, error_message)
633        - is_valid: True if file is valid, False otherwise
634        - error_message: Empty string if valid, otherwise a formatted error message
635    """
636    validator = ConfigValidator()
637    is_valid, errors = validator.validate_file(file_path)
638
639    if is_valid and not errors:
640        return True, ""
641
642    # Format errors into a readable message
643    error_lines = [str(e) for e in errors if e.severity == "error"]
644    warning_lines = [str(e) for e in errors if e.severity == "warning"]
645
646    message_parts = []
647    if error_lines:
648        message_parts.extend(error_lines)
649    if warning_lines:
650        message_parts.extend(warning_lines)
651
652    return is_valid, "\n".join(message_parts)
653
654
655def validate_files(file_patterns: List[str], verbose: bool = False) -> int:
656    """
657    Validate multiple files and return exit code.
658
659    Returns:
660        0 if all files valid, 1 if any errors found
661    """
662    validator = ConfigValidator(verbose=verbose)
663    all_valid = True
664    files_checked = 0
665    files_with_errors = 0
666
667    # Expand file patterns
668    files = []
669    for pattern in file_patterns:
670        expanded = glob.glob(pattern)
671        if not expanded:
672            print(f"Warning: No files match pattern: {pattern}")
673        files.extend(expanded)
674
675    if not files:
676        print("No files to validate")
677        return 1
678
679    for file_path in sorted(files):
680        files_checked += 1
681
682        if verbose:
683            print(f"\nValidating: {file_path}")
684
685        is_valid, errors = validator.validate_file(file_path)
686
687        if not is_valid or errors:
688            files_with_errors += 1
689            all_valid = False
690
691            if not verbose:
692                print(f"\n{file_path}:")
693
694            for error in errors:
695                print(f"  {error}")
696        elif verbose:
697            print("  OK")
698
699    # Summary
700    print(f"\n{'='*60}")
701    print(f"Validated {files_checked} file(s): ", end="")
702    if all_valid:
703        print("All valid")
704    else:
705        print(f"{files_with_errors} with errors/warnings")
706
707    return 0 if all_valid else 1
708
709
710def main():
711    parser = argparse.ArgumentParser(
712        description="Validate OpenRVDAS configuration files",
713        formatter_class=argparse.RawDescriptionHelpFormatter,
714        epilog="""
715Examples:
716  %(prog)s config.yaml                    Validate a single file
717  %(prog)s -v config.yaml                 Verbose output
718  %(prog)s contrib/devices/*.yaml         Validate all device definitions
719  %(prog)s test/configs/*.yaml            Validate all test configs
720""")
721    parser.add_argument('files', nargs='+', help='File(s) or pattern(s) to validate')
722    parser.add_argument('-v', '--verbose', action='store_true',
723                        help='Verbose output')
724
725    args = parser.parse_args()
726    sys.exit(validate_files(args.files, verbose=args.verbose))
727
728
729if __name__ == '__main__':
730    main()