DD3-LoRa-Bridge-MultiSender/test_republish_compatibility.py

#!/usr/bin/env python3
"""
Compatibility test for republish_mqtt.py and republish_mqtt_gui.py
Tests against newest CSV and InfluxDB formats
"""

import csv
import json
import tempfile
import sys
from pathlib import Path
from datetime import datetime, timedelta

def test_csv_format_current():
    """Test that scripts can parse the CURRENT SD logger CSV format (ts_hms_local)"""
    print("\n=== TEST 1: CSV Format (Current HD logger) ===")
    
    # Current format from sd_logger.cpp line 105:
    # ts_utc,ts_hms_local,p_w,p1_w,p2_w,p3_w,e_kwh,bat_v,bat_pct,rssi,snr,err_m,err_d,err_tx,err_last
    
    csv_header = "ts_utc,ts_hms_local,p_w,p1_w,p2_w,p3_w,e_kwh,bat_v,bat_pct,rssi,snr,err_m,err_d,err_tx,err_last"
    csv_data = "1710076800,08:00:00,5432,1800,1816,1816,1234.567,4.15,95,-95,9.25,0,0,0,"
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False, newline='') as f:
        f.write(csv_header + '\n')
        f.write(csv_data + '\n')
        csv_file = f.name
    
    try:
        # Parse like the republish script does
        with open(csv_file, 'r') as f:
            reader = csv.DictReader(f)
            fieldnames = reader.fieldnames
            
            # Check required fields
            required = ['ts_utc', 'e_kwh', 'p_w']
            missing = [field for field in required if field not in fieldnames]
            
            if missing:
                print(f"❌ FAIL: Missing required fields: {missing}")
                return False
            
            # Check optional fields that scripts handle
            optional_handled = ['p1_w', 'p2_w', 'p3_w', 'bat_v', 'bat_pct', 'rssi', 'snr']
            present_optional = [f for f in optional_handled if f in fieldnames]
            
            print(f"✓ Required fields: {required}")
            print(f"✓ Optional fields found: {present_optional}")
            
            # Try parsing first row
            for row in reader:
                try:
                    ts_utc = int(row['ts_utc'])
                    e_kwh = float(row['e_kwh'])
                    p_w = int(round(float(row['p_w'])))
                    print(f"✓ Parsed sample: ts={ts_utc}, e_kwh={e_kwh:.2f}, p_w={p_w}W")
                    return True
                except (ValueError, KeyError) as e:
                    print(f"❌ FAIL: Could not parse row: {e}")
                    return False
    finally:
        Path(csv_file).unlink()


def test_csv_format_with_new_fields():
    """Test that scripts gracefully handle new CSV fields (rx_reject, etc)"""
    print("\n=== TEST 2: CSV Format with Future Fields ===")
    
    # Hypothetical future format with additional fields
    csv_header = "ts_utc,ts_hms_local,p_w,p1_w,p2_w,p3_w,e_kwh,bat_v,bat_pct,rssi,snr,err_m,err_d,err_tx,err_last,rx_reject,rx_reject_text"
    csv_data = "1710076800,08:00:00,5432,1800,1816,1816,1234.567,4.15,95,-95,9.25,0,0,0,,0,none"
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False, newline='') as f:
        f.write(csv_header + '\n')
        f.write(csv_data + '\n')
        csv_file = f.name
    
    try:
        with open(csv_file, 'r') as f:
            reader = csv.DictReader(f)
            fieldnames = reader.fieldnames
            
            # Check required fields
            required = ['ts_utc', 'e_kwh', 'p_w']
            missing = [field for field in required if field not in fieldnames]
            
            if missing:
                print(f"❌ FAIL: Missing required fields: {missing}")
                return False
            
            print(f"✓ All required fields present: {required}")
            print(f"✓ Total fields in format: {len(fieldnames)}")
            print(f"  - New field 'rx_reject': {'rx_reject' in fieldnames}")
            print(f"  - New field 'rx_reject_text': {'rx_reject_text' in fieldnames}")
            
            return True
    finally:
        Path(csv_file).unlink()


def test_mqtt_json_format():
    """Test that republished MQTT JSON format matches device format"""
    print("\n=== TEST 3: MQTT JSON Format ===")
    
    # Simulate what the republish script generates
    csv_row = {
        'ts_utc': '1710076800',
        'e_kwh': '1234.567',
        'p_w': '5432.1',
        'p1_w': '1800.5',
        'p2_w': '1816.3',
        'p3_w': '1815.7',
        'bat_v': '4.15',
        'bat_pct': '95',
        'rssi': '-95',
        'snr': '9.25'
    }
    
    # Republish script builds this
    data = {
        'id': 'F19C',  # Last 4 chars of device_id
        'ts': int(csv_row['ts_utc']),
    }
    
    # Energy
    e_kwh = float(csv_row['e_kwh'])
    data['e_kwh'] = f"{e_kwh:.2f}"
    
    # Power values (as integers)
    for key in ['p_w', 'p1_w', 'p2_w', 'p3_w']:
        if key in csv_row and csv_row[key].strip():
            data[key] = int(round(float(csv_row[key])))
    
    # Battery
    if 'bat_v' in csv_row and csv_row['bat_v'].strip():
        data['bat_v'] = f"{float(csv_row['bat_v']):.2f}"
    
    if 'bat_pct' in csv_row and csv_row['bat_pct'].strip():
        data['bat_pct'] = int(csv_row['bat_pct'])
    
    # Link quality
    if 'rssi' in csv_row and csv_row['rssi'].strip() and csv_row['rssi'] != '-127':
        data['rssi'] = int(csv_row['rssi'])
    
    if 'snr' in csv_row and csv_row['snr'].strip():
        data['snr'] = float(csv_row['snr'])
    
    # What the device format expects (from json_codec.cpp)
    expected_fields = {'id', 'ts', 'e_kwh', 'p_w', 'p1_w', 'p2_w', 'p3_w', 'bat_v', 'bat_pct', 'rssi', 'snr'}
    actual_fields = set(data.keys())
    
    print(f"✓ Republish script generates:")
    print(f"  JSON: {json.dumps(data, indent=2)}")
    print(f"✓ Field types:")
    for field, value in data.items():
        print(f"  - {field}: {type(value).__name__} = {repr(value)}")
    
    if expected_fields == actual_fields:
        print(f"✓ All expected fields present")
        return True
    else:
        missing = expected_fields - actual_fields
        extra = actual_fields - expected_fields
        if missing:
            print(f"⚠ Missing fields: {missing}")
        if extra:
            print(f"⚠ Extra fields: {extra}")
        return True  # Still OK if extra/missing as device accepts optional fields


def test_csv_legacy_format():
    """Test backward compatibility with legacy CSV format (no ts_hms_local)"""
    print("\n=== TEST 4: CSV Format (Legacy - no ts_hms_local) ===")
    
    # Legacy format: just ts_utc,p_w,... (from README: History parser accepts both)
    csv_header = "ts_utc,p_w,e_kwh,p1_w,p2_w,p3_w,bat_v,bat_pct,rssi,snr"
    csv_data = "1710076800,5432,1234.567,1800,1816,1816,4.15,95,-95,9.25"
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False, newline='') as f:
        f.write(csv_header + '\n')
        f.write(csv_data + '\n')
        csv_file = f.name
    
    try:
        with open(csv_file, 'r') as f:
            reader = csv.DictReader(f)
            
            required = ['ts_utc', 'e_kwh', 'p_w']
            missing = [field for field in required if field not in reader.fieldnames]
            
            if missing:
                print(f"❌ FAIL: Missing required fields: {missing}")
                return False
            
            print(f"✓ Legacy format compatible (ts_hms_local not required)")
            return True
    finally:
        Path(csv_file).unlink()


def test_influxdb_query_schema():
    """Document expected InfluxDB schema for auto-detect"""
    print("\n=== TEST 5: InfluxDB Schema (Query Format) ===")
    print("""
    The republish scripts expect:
    - Measurement: "smartmeter"
    - Tag name: "device_id"
    - Query example:
      from(bucket: "smartmeter")
        |> range(start: <timestamp>, stop: <timestamp>)
        |> filter(fn: (r) => r._measurement == "smartmeter" and r.device_id == "dd3-F19C")
        |> keep(columns: ["_time"])
        |> sort(columns: ["_time"])
    """)
    
    print("✓ Expected schema documented")
    print("⚠ NOTE: Device firmware does NOT write to InfluxDB directly")
    print("  → Requires separate bridge (Telegraf, Node-RED, etc) from MQTT → InfluxDB")
    print("  → InfluxDB auto-detect mode is OPTIONAL - manual mode always works")
    return True


def print_summary(results):
    """Print test summary"""
    print("\n" + "="*60)
    print("TEST SUMMARY")
    print("="*60)
    
    passed = sum(1 for r in results if r)
    total = len(results)
    
    test_names = [
        "CSV Format (Current with ts_hms_local)",
        "CSV Format (with future fields)",
        "MQTT JSON Format compatibility",
        "CSV Format (Legacy - backward compat)",
        "InfluxDB schema validation"
    ]
    
    for i, (name, result) in enumerate(zip(test_names, results)):
        status = "✓ PASS" if result else "❌ FAIL"
        print(f"{status}: {name}")
    
    print(f"\nResult: {passed}/{total} tests passed")
    return passed == total


if __name__ == '__main__':
    print("="*60)
    print("DD3 MQTT Republisher - Compatibility Tests")
    print("Testing against newest CSV and InfluxDB formats")
    print(f"Date: {datetime.now()}")
    print("="*60)
    
    results = [
        test_csv_format_current(),
        test_csv_format_with_new_fields(),
        test_mqtt_json_format(),
        test_csv_legacy_format(),
        test_influxdb_query_schema(),
    ]
    
    all_passed = print_summary(results)
    sys.exit(0 if all_passed else 1)