#!/usr/bin/env python3 """ merge_component.py After splitting the HydraAix Influx CSV export into one file per field (e.g. with split.sh, giving part_001.csv, part_002.csv, ...), this script merges all the fields belonging to ONE equipment component back together into a single JSON-lines file, one JSON object per timestamp. Example: given part_002.csv contains only ACH65B_gMotorVoltage part_003.csv contains only ACH65B_ggElectronicTemperature and both have a row at the exact same _time 2026-07-10T00:01:20.90597613Z this produces one merged record: {"timestamp": "2026-07-10T00:01:20.90597613Z", "start_ts": "...", "end_ts": "...", "boat": "HydraAix", "component": "ACH65B", "gMotorVoltage": 397.2, "ggElectronicTemperature": 35} The merge key is the RAW _time STRING, compared for exact equality -- no tolerance window, no rounding. If two fields' samples are even one character apart in their timestamp string, they land in separate output records. This matches what was actually observed in the real data (fields sampled together share byte-identical _time values); if that turns out not to hold everywhere, this script will simply produce more, smaller records rather than silently merging things that don't match. ----------------------------------------------------------------------- USAGE ----------------------------------------------------------------------- Merge one specific component: python3 merge_component.py ACH65B --input-dir vessel-data2 \ --pattern "part_*.csv" --output ACH65B.jsonl Merge EVERY component found across all matching files in one pass (recommended -- scans the files once, not once per component): python3 merge_component.py --all --input-dir vessel-data2 \ --pattern "part_*.csv" --output-dir merged Fields with no underscore (no equipment prefix, e.g. "batteryStateOfChargePercentB") have no separate component to merge with -- each becomes its own single-field output file, same as running convert_helios_influx.py on that one field alone. ----------------------------------------------------------------------- OUTPUT SCHEMA (same as convert_helios_influx.py's per-row schema) ----------------------------------------------------------------------- timestamp raw _time string of the group (unmodified) start_ts raw _start string (taken from whichever row in the group had it; not re-verified across all rows in the group) end_ts raw _stop string (see above) boat _measurement value (e.g. "HydraAix") component the equipment prefix (e.g. "ACH65B") one key per field in the group, named by the field name with the component prefix stripped (e.g. "gMotorVoltage") Values are type-preserved from the CSV's #datatype annotation (double -> float, long -> int, boolean -> bool, string -> string; never silently coerced). ----------------------------------------------------------------------- IF THE SAME FIELD APPEARS TWICE AT THE SAME EXACT TIMESTAMP ----------------------------------------------------------------------- This can only happen if the same field's data is duplicated across files (e.g. overlapping exports). The later value wins and a warning is printed (use --verbose to see every occurrence, not just a summary count) -- values are never silently dropped without being counted. No external dependencies -- standard library only. """ from __future__ import annotations import argparse import csv import glob import json import os import sys from collections import defaultdict # ---------------------------------------------------------------------------- # Minimal, self-contained Influx raw CSV row reader (same parsing rules as # convert_helios_influx.py, duplicated here so this script has no import # dependency on that file and can be copied/run standalone). # ---------------------------------------------------------------------------- def convert_value(raw: str, dtype: str | None): if dtype == "double": return float(raw) if dtype in ("long", "unsignedLong"): return int(raw) if dtype == "boolean": low = raw.strip().lower() if low in ("true", "1"): return True if low in ("false", "0"): return False raise ValueError(f"Unrecognized boolean value: {raw!r}") # "string" or unknown/missing dtype: preserve as string, never guess. return raw def iter_influx_rows(path: str, measurement_filter: str): """Yields dicts: {time, start, stop, field, value, measurement} for every valid data row in an Influx raw CSV export whose _measurement matches measurement_filter. Silently-tolerant of malformed rows (skips them) and multiple concatenated annotation blocks, exactly like convert_helios_influx.py's iter_influx_rows. """ columns = None dtypes = None with open(path, newline="", encoding="utf-8", errors="replace") as f: reader = csv.reader(f) for row in reader: if not row or all(cell == "" for cell in row): continue first = row[0] if first.startswith("#group"): continue if first.startswith("#datatype"): dtypes = row continue if first.startswith("#default"): continue if len(row) > 2 and row[1] == "result" and row[2] == "table": columns = row continue if columns is None or len(row) != len(columns): continue rec = dict(zip(columns, row)) measurement = rec.get("_measurement", "") if measurement != measurement_filter: continue field = rec.get("_field") time_raw = rec.get("_time") value_raw = rec.get("_value") if field is None or time_raw is None or value_raw is None: continue value_dtype = None if dtypes is not None and "_value" in columns: try: value_dtype = dtypes[columns.index("_value")] except IndexError: value_dtype = None try: value = convert_value(value_raw, value_dtype) except ValueError: continue yield { "time": time_raw, "start": rec.get("_start"), "stop": rec.get("_stop"), "field": field, "value": value, "measurement": measurement, } def split_field(field: str): if "_" in field: component, remainder = field.split("_", 1) return component, remainder return field, None # ---------------------------------------------------------------------------- # Merge logic # ---------------------------------------------------------------------------- def merge_all_components(files, measurement, field_map, verbose): """Single pass over all files: groups rows by (component, exact timestamp string). Returns: groups: {component: {time: {"start":.., "stop":.., "measurement":.., "fields": {key: value}}}} files_per_component: {component: set(file paths that contributed)} duplicate_count: {component: int} """ groups = defaultdict(dict) files_per_component = defaultdict(set) duplicate_count = defaultdict(int) for path in files: for row in iter_influx_rows(path, measurement): component, remainder = split_field(row["field"]) key = field_map.get(row["field"], remainder) if remainder is not None else "value" if remainder is None: # No equipment prefix -- the field name itself is the # "component" (matches convert_helios_influx.py's rule). component = row["field"] t = row["time"] comp_groups = groups[component] entry = comp_groups.get(t) if entry is None: entry = {"start": row["start"], "stop": row["stop"], "measurement": row["measurement"], "fields": {}} comp_groups[t] = entry if key in entry["fields"]: duplicate_count[component] += 1 if verbose: print( f"WARNING: duplicate '{key}' at timestamp {t} for component " f"'{component}' in {path} (previous value overwritten)", file=sys.stderr, ) entry["fields"][key] = row["value"] files_per_component[component].add(path) return groups, files_per_component, duplicate_count def write_component_file(component, comp_groups, out_path): count = 0 with open(out_path, "w", encoding="utf-8") as out: for t in sorted(comp_groups.keys()): g = comp_groups[t] rec = { "timestamp": t, "start_ts": g["start"], "end_ts": g["stop"], "boat": g["measurement"], "component": component, } rec.update(g["fields"]) out.write(json.dumps(rec, sort_keys=True) + "\n") count += 1 return count def main(): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("component", nargs="?", default=None, help="Equipment component prefix to merge (e.g. ACH65B). Omit and use --all instead " "to merge every component found.") p.add_argument("--all", action="store_true", help="Merge every component found across the matched files.") p.add_argument("--input-dir", default=".", help="Directory containing the split part files. (default: %(default)s)") p.add_argument("--pattern", default="part_*.csv", help="Glob pattern for the split files. (default: %(default)s)") p.add_argument("--measurement", default="HydraAix", help="Only merge rows with this _measurement value. (default: %(default)s)") p.add_argument("--output", default=None, help="Output file path (single-component mode only). Default: .jsonl") p.add_argument("--output-dir", default="merged", help="Output directory (--all mode only). (default: %(default)s)") p.add_argument("--field-map", default=None, help="Optional JSON file mapping raw _field names to output key names.") p.add_argument("--verbose", action="store_true", help="Print every duplicate/overwrite, not just a count.") args = p.parse_args() if not args.all and not args.component: p.error("either give a component name or pass --all") files = sorted(glob.glob(os.path.join(args.input_dir, args.pattern))) if not files: print(f"ERROR: no files matched '{args.pattern}' in '{args.input_dir}'", file=sys.stderr) sys.exit(2) field_map = {} if args.field_map: with open(args.field_map, encoding="utf-8") as f: field_map = json.load(f) if args.verbose: print(f"Scanning {len(files)} file(s) matching '{args.pattern}' in '{args.input_dir}' ...", file=sys.stderr) groups, files_per_component, duplicate_count = merge_all_components(files, args.measurement, field_map, args.verbose) if not groups: print(f"No matching rows found for measurement '{args.measurement}' in any of {len(files)} file(s).", file=sys.stderr) sys.exit(1) if args.all: os.makedirs(args.output_dir, exist_ok=True) print(f"{'Component':<20} {'Files':>6} {'Records':>9} {'Duplicates':>11} Output") for component in sorted(groups.keys()): out_path = os.path.join(args.output_dir, f"{component}.jsonl") count = write_component_file(component, groups[component], out_path) print( f"{component:<20} {len(files_per_component[component]):>6} {count:>9} " f"{duplicate_count.get(component, 0):>11} {out_path}" ) else: if args.component not in groups: available = ", ".join(sorted(groups.keys())[:20]) print( f"ERROR: no fields found for component '{args.component}' in the matched files.\n" f"Components that WERE found: {available}{' ...' if len(groups) > 20 else ''}", file=sys.stderr, ) sys.exit(1) out_path = args.output or f"{args.component}.jsonl" count = write_component_file(args.component, groups[args.component], out_path) print(f"Component: {args.component}") print(f"Files contributing: {len(files_per_component[args.component])}") for f in sorted(files_per_component[args.component]): print(f" {f}") print(f"Records written: {count}") print(f"Duplicates overwritten: {duplicate_count.get(args.component, 0)}") print(f"Output: {out_path}") if __name__ == "__main__": main()