openrvdas.logger.transforms.convert_fields_transform

  1#!/usr/bin/env python3
  2"""Transform to convert fields in a DASRecord to specified types.
  3
  4This is a thin wrapper around the convert_fields utility function,
  5providing the Transform interface for use in listener pipelines.
  6"""
  7
  8import copy
  9import logging
 10from typing import Union
 11
 12
 13from logger.utils.das_record import DASRecord  # noqa: E402
 14from logger.utils.convert_fields import convert_fields  # noqa: E402
 15from logger.transforms.transform import Transform  # noqa: E402
 16
 17
 18################################################################################
 19class ConvertFieldsTransform(Transform):
 20    """
 21    Converts fields in a DASRecord from strings (or other types) to specific types
 22    defined in a configuration dictionary. Also handles NMEA-style latitude/longitude
 23    conversions (e.g., combining a value and a cardinal direction field).
 24    """
 25
 26    def __init__(self, fields=None, delete_source_fields=False,
 27                 delete_unconverted_fields=False, **kwargs):
 28        """
 29        Args:
 30            fields (dict): A dictionary mapping field names to their target configuration.
 31                           This accepts two formats for the value:
 32                           1. A dictionary containing metadata (preferred).
 33                              keys:
 34                                'data_type': target type (float, int, str, bool, hex,
 35                                             nmea_lat, nmea_lon)
 36                                'direction_field': (for nmea_*) name of direction field
 37                              Example:
 38                                  {'Latitude': {'data_type': 'nmea_lat',
 39                                                'direction_field': 'NorS'}}
 40                           2. A simple string specifying the data type (backward compatibility).
 41                              Example:
 42                                  {'heave': 'float'}
 43
 44            delete_source_fields (bool): If True, source fields (e.g. 'raw_lat', 'lat_dir')
 45                                         are removed after successful conversion.
 46                                         Defaults to False.
 47
 48            delete_unconverted_fields (bool): If True, fields NOT involved in conversion
 49                                              are removed. Defaults to False.
 50        """
 51        super().__init__(**kwargs)  # processes 'quiet' and type hints
 52
 53        self.field_specs = {}
 54        self.lat_lon_specs = {}
 55        self.delete_source_fields = delete_source_fields
 56        self.delete_unconverted_fields = delete_unconverted_fields
 57
 58        if fields:
 59            # Process fields to separate standard conversions from special NMEA ones
 60            for f_name, f_def in fields.items():
 61                # Normalize definition
 62                if isinstance(f_def, str):
 63                    f_def = {'data_type': f_def}
 64
 65                dtype = f_def.get('data_type')
 66
 67                # Check for declarative NMEA configuration
 68                if dtype in ['nmea_lat', 'nmea_lon']:
 69                    dir_field = f_def.get('direction_field')
 70                    if dir_field:
 71                        # Format: target_field -> (value_field, direction_field)
 72                        self.lat_lon_specs[f_name] = (f_name, dir_field)
 73                    else:
 74                        logging.warning(f"Field '{f_name}' has type '{dtype}' but "
 75                                        "missing 'direction_field'. Ignoring.")
 76                else:
 77                    # Standard field
 78                    self.field_specs[f_name] = f_def
 79
 80    ############################
 81    def transform(self, record: Union[str, dict, DASRecord])\
 82            -> Union[str, dict, DASRecord]:
 83        """
 84        Return a copy of the passed record with fields converted.
 85        """
 86        # See if it's something we can process, and if not, try digesting
 87        if not self.can_process_record(record):  # BaseModule
 88            return self.digest_record(record)  # BaseModule
 89
 90        # We need to make a deep copy because we modify the record in place
 91        new_record = copy.deepcopy(record)
 92
 93        # Handle list of records
 94        if isinstance(new_record, list):
 95            new_record_list = []
 96            for single_record in new_record:
 97                result = self.transform(single_record)
 98                if result:
 99                    new_record_list.append(result)
100            return new_record_list
101
102        # Identify the fields dictionary
103        if isinstance(new_record, DASRecord):
104            fields = new_record.fields
105        elif isinstance(new_record, dict):
106            if 'fields' in new_record and isinstance(new_record['fields'], dict):
107                fields = new_record['fields']
108            else:
109                fields = new_record
110        else:
111            logging.warning('ConvertFieldsTransform received unknown record type: %s',
112                            type(new_record))
113            return None
114
115        # Delegate to the utility function
116        result = convert_fields(
117            fields,
118            self.field_specs,
119            self.lat_lon_specs,
120            delete_source_fields=self.delete_source_fields,
121            delete_unconverted_fields=self.delete_unconverted_fields,
122            quiet=self.quiet
123        )
124
125        # If no fields remain, return None
126        if result is None:
127            return None
128
129        return new_record