openrvdas.logger.utils.convert_fields

  1#!/usr/bin/env python3
  2"""Utilities for converting field values to specified types."""
  3
  4import logging
  5
  6# Map string type names to actual python types/conversion functions.
  7# Recognized types include:
  8#   float, double -> float
  9#   int, short, ushort, uint, long, ubyte, byte, hex_int -> int
 10#   str, char, string, text -> str
 11#   bool, boolean -> bool
 12TYPE_MAP = {
 13    'float': float,
 14    'double': float,
 15    'int': int,
 16    'short': int,
 17    'ushort': int,
 18    'uint': int,
 19    'long': int,
 20    'ubyte': int,
 21    'byte': int,
 22    'str': str,
 23    'char': str,
 24    'string': str,
 25    'text': str,
 26    'bool': bool,
 27    'boolean': bool,
 28    'hex_int': lambda x: int(str(x), 16),  # Handles "1A", "0x1A", etc.
 29}
 30
 31
 32def convert_lat_lon(value, direction, rounding_decimals=5):
 33    """
 34    Convert NMEA style lat/lon (DDMM.MMMM) and direction (N/S/E/W)
 35    to decimal degrees.
 36
 37    Args:
 38        value: NMEA format value (e.g., "4807.038" for 48°07.038')
 39        direction: Cardinal direction ('N', 'S', 'E', 'W')
 40        rounding_decimals: Number of decimal places to round to
 41
 42    Returns:
 43        Decimal degrees as float, or None if conversion fails
 44    """
 45    try:
 46        val = float(value)
 47        # NMEA format is roughly DDMM.MMMM
 48        # Degrees is the integer part of val / 100
 49        degrees = int(val / 100)
 50        minutes = val - (degrees * 100)
 51        decimal = degrees + (minutes / 60)
 52
 53        if direction.upper() in ['S', 'W']:
 54            decimal = -decimal
 55
 56        return round(decimal, rounding_decimals)
 57    except (ValueError, TypeError, AttributeError) as e:
 58        logging.warning(f'Failed to convert lat/lon: value=\'{value}\', '
 59                        f'direction=\'{direction}\' - {e}')
 60        return None
 61
 62
 63def convert_field_value(value, target_type_str, quiet=False):
 64    """
 65    Convert a single field value to the specified type.
 66
 67    Args:
 68        value: The value to convert
 69        target_type_str: String name of target type (e.g., 'float', 'int', 'str')
 70        quiet: If True, suppress warning messages
 71
 72    Returns:
 73        Converted value, or original value if conversion fails
 74    """
 75    converter = TYPE_MAP.get(target_type_str)
 76    if not converter:
 77        if not quiet:
 78            logging.warning(f'Unknown type \'{target_type_str}\' requested')
 79        return value
 80
 81    try:
 82        # Special case to head off ValueError of int("123.0")
 83        if isinstance(value, str) and converter is int:
 84            try:
 85                value = float(value)
 86            except ValueError:
 87                # Not a float string, let int() call below handle/fail it naturally
 88                pass
 89
 90        return converter(value)
 91    except ValueError:
 92        if not quiet:
 93            logging.warning(f'Failed to convert value \'{value}\' '
 94                            f'(type={type(value).__name__}) to \'{target_type_str}\'')
 95        return value
 96
 97
 98def convert_fields(fields, field_specs, lat_lon_specs=None,
 99                   delete_source_fields=False, delete_unconverted_fields=False,
100                   quiet=False):
101    """
102    Convert fields in a dictionary according to specifications.
103
104    This is a shared utility used by ConvertFieldsTransform and can be used
105    directly by parsers for field conversion.
106
107    Args:
108        fields: Dict of field_name -> value (modified in place)
109        field_specs: Dict of field_name -> target_type or {data_type: target_type}
110        lat_lon_specs: Dict of target_field -> (value_field, direction_field)
111                      for NMEA lat/lon conversion
112        delete_source_fields: If True, remove source fields after lat/lon conversion
113        delete_unconverted_fields: If True, remove fields not involved in conversion
114        quiet: If True, suppress warning messages
115
116    Returns:
117        The modified fields dict, or None if no fields remain after processing
118    """
119    if not fields:
120        return None
121
122    # Track which fields were successfully converted or used
123    processed_fields = set()
124
125    # 1. Handle simple Type Conversions
126    if field_specs:
127        for field_name, field_def in field_specs.items():
128            if field_name not in fields:
129                continue
130
131            # Extract target type from dict, or use string directly
132            if isinstance(field_def, dict):
133                target_type_str = field_def.get('data_type')
134            elif isinstance(field_def, str):
135                target_type_str = field_def
136            else:
137                continue
138
139            if not target_type_str:
140                continue
141
142            val = fields[field_name]
143            converter = TYPE_MAP.get(target_type_str)
144            if converter:
145                try:
146                    # Special case to head off ValueError of int("123.0")
147                    if isinstance(val, str) and converter is int:
148                        try:
149                            val = float(val)
150                        except ValueError:
151                            pass
152
153                    fields[field_name] = converter(val)
154                    processed_fields.add(field_name)
155                except ValueError:
156                    if not quiet:
157                        logging.warning(f'Failed to convert field \'{field_name}\': '
158                                        f'value=\'{val}\' (type={type(val).__name__}) '
159                                        f'to target_type=\'{target_type_str}\'')
160            else:
161                if not quiet:
162                    logging.warning(f'Unknown type \'{target_type_str}\' '
163                                    f'requested for field \'{field_name}\'')
164
165    # 2. Handle Lat/Lon Conversions
166    if lat_lon_specs:
167        for target_field, (val_field, dir_field) in lat_lon_specs.items():
168            if val_field not in fields or dir_field not in fields:
169                continue
170
171            val = fields[val_field]
172            direction = fields[dir_field]
173
174            decimal_degrees = convert_lat_lon(val, direction)
175
176            if decimal_degrees is not None:
177                fields[target_field] = decimal_degrees
178                processed_fields.add(target_field)
179
180                # Mark source fields as processed
181                if delete_source_fields:
182                    processed_fields.add(val_field)
183                    processed_fields.add(dir_field)
184
185                    # Delete source fields, but NOT if source == target
186                    if val_field in fields and val_field != target_field:
187                        del fields[val_field]
188                    if dir_field in fields and dir_field != target_field:
189                        del fields[dir_field]
190
191    # 3. Clean up unconverted fields
192    if delete_unconverted_fields:
193        all_fields = list(fields.keys())
194        for f in all_fields:
195            if f not in processed_fields:
196                del fields[f]
197
198    # If no fields remain, return None
199    if not fields:
200        return None
201
202    return fields