paulg/venus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
9756538f16ae5a1aae7d35a32183404125730e63
venus/axiom-nmea/debug/battery_debug.py
dev 9756538f16 Initial commit: Venus OS boat addons monorepo 
Organizes 11 projects for Cerbo GX/Venus OS into a single repository:
- axiom-nmea: Raymarine LightHouse protocol decoder
- dbus-generator-ramp: Generator current ramp controller
- dbus-lightning: Blitzortung lightning monitor
- dbus-meteoblue-forecast: Meteoblue weather forecast
- dbus-no-foreign-land: noforeignland.com tracking
- dbus-tides: Tide prediction from depth + harmonics
- dbus-vrm-history: VRM cloud history proxy
- dbus-windy-station: Windy.com weather upload
- mfd-custom-app: MFD app deployment package
- venus-html5-app: Custom Victron HTML5 app fork
- watermaker: Watermaker PLC control UI

Adds root README, .gitignore, project template, and per-project
.gitignore files. Sensitive config files excluded via .gitignore
with .example templates provided.

Made-with: Cursor
2026-03-16 17:04:16 +00:00

454 lines
16 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Battery Debug - Dump raw protobuf entries to find battery data fields.
Scans for fields that might contain battery data (voltage, current, SoC).
Supports deep nesting (e.g., Field 14.3.4 for engine battery).
"""
import struct
import socket
import time
import threading
from typing import Dict, List, Any, Optional
WIRE_VARINT = 0
WIRE_FIXED64 = 1
WIRE_LENGTH = 2
WIRE_FIXED32 = 5
HEADER_SIZE = 20
# Fields to specifically look for battery data
# Field 14 = Engine data (contains battery at 14.3.4)
# Field 20 = House batteries
BATTERY_CANDIDATE_FIELDS = {14, 17, 18, 19, 20, 21, 22, 23, 24, 25}
MULTICAST_GROUPS = [
("226.192.206.102", 2565), # Main sensor data
("239.2.1.1", 2154), # May contain additional sensor data
]
class ProtobufParser:
def __init__(self, data: bytes):
self.data = data
self.pos = 0
def remaining(self):
return len(self.data) - self.pos
def read_varint(self) -> int:
result = 0
shift = 0
while self.pos < len(self.data):
byte = self.data[self.pos]
self.pos += 1
result |= (byte & 0x7F) << shift
if not (byte & 0x80):
break
shift += 7
return result
def read_fixed32(self) -> bytes:
val = self.data[self.pos:self.pos + 4]
self.pos += 4
return val
def read_fixed64(self) -> bytes:
val = self.data[self.pos:self.pos + 8]
self.pos += 8
return val
def read_length_delimited(self) -> bytes:
length = self.read_varint()
val = self.data[self.pos:self.pos + length]
self.pos += length
return val
def parse_all_fields(self) -> Dict[int, List[Any]]:
"""Parse and collect all fields, grouping repeated fields."""
fields = {}
while self.pos < len(self.data):
if self.remaining() < 1:
break
try:
start_pos = self.pos
tag = self.read_varint()
field_num = tag >> 3
wire_type = tag & 0x07
if field_num == 0 or field_num > 1000:
break
if wire_type == WIRE_VARINT:
value = ('varint', self.read_varint())
elif wire_type == WIRE_FIXED64:
raw = self.read_fixed64()
try:
d = struct.unpack('<d', raw)[0]
value = ('double', d, raw.hex())
except:
value = ('fixed64', raw.hex())
elif wire_type == WIRE_LENGTH:
raw = self.read_length_delimited()
value = ('length', raw)
elif wire_type == WIRE_FIXED32:
raw = self.read_fixed32()
try:
f = struct.unpack('<f', raw)[0]
value = ('float', f, raw.hex())
except:
value = ('fixed32', raw.hex())
else:
break
if field_num not in fields:
fields[field_num] = []
fields[field_num].append(value)
except:
break
return fields
def parse_nested_deep(self, data: bytes, path: str = "", depth: int = 0, max_depth: int = 5) -> List[tuple]:
"""Recursively parse nested protobuf and return list of (path, type, value) tuples."""
results = []
pos = 0
if depth > max_depth:
return results
while pos < len(data):
if pos >= len(data):
break
try:
# Read tag
tag_byte = data[pos]
pos += 1
# Handle multi-byte varints for tag
tag = tag_byte & 0x7F
shift = 7
while tag_byte & 0x80 and pos < len(data):
tag_byte = data[pos]
pos += 1
tag |= (tag_byte & 0x7F) << shift
shift += 7
field_num = tag >> 3
wire_type = tag & 0x07
if field_num == 0 or field_num > 100:
break
field_path = f"{path}.{field_num}" if path else str(field_num)
if wire_type == WIRE_VARINT:
# Read varint value
val = 0
shift = 0
while pos < len(data):
byte = data[pos]
pos += 1
val |= (byte & 0x7F) << shift
if not (byte & 0x80):
break
shift += 7
results.append((field_path, 'varint', val))
elif wire_type == WIRE_FIXED32:
raw = data[pos:pos + 4]
pos += 4
try:
f = struct.unpack('<f', raw)[0]
if f == f: # not NaN
results.append((field_path, 'float', f))
except:
pass
elif wire_type == WIRE_FIXED64:
raw = data[pos:pos + 8]
pos += 8
try:
d = struct.unpack('<d', raw)[0]
if d == d: # not NaN
results.append((field_path, 'double', d))
except:
pass
elif wire_type == WIRE_LENGTH:
# Read length
length = 0
shift = 0
while pos < len(data):
byte = data[pos]
pos += 1
length |= (byte & 0x7F) << shift
if not (byte & 0x80):
break
shift += 7
raw = data[pos:pos + length]
pos += length
# Try to parse as nested message
if len(raw) >= 2:
nested_results = self.parse_nested_deep(raw, field_path, depth + 1, max_depth)
if nested_results:
results.extend(nested_results)
else:
# Couldn't parse as nested, store as bytes
results.append((field_path, 'bytes', raw.hex()[:40]))
else:
results.append((field_path, 'bytes', raw.hex()))
else:
break
except Exception as e:
break
return results
def parse_nested_entry(self, data: bytes) -> dict:
"""Parse a nested protobuf entry and return all fields (single level)."""
entry = {'fields': {}}
pos = 0
while pos < len(data):
if pos >= len(data):
break
try:
# Read tag
tag_byte = data[pos]
pos += 1
# Handle multi-byte varints for tag
tag = tag_byte & 0x7F
shift = 7
while tag_byte & 0x80 and pos < len(data):
tag_byte = data[pos]
pos += 1
tag |= (tag_byte & 0x7F) << shift
shift += 7
field_num = tag >> 3
wire_type = tag & 0x07
if field_num == 0 or field_num > 100:
break
if wire_type == WIRE_VARINT:
# Read varint value
val = 0
shift = 0
while pos < len(data):
byte = data[pos]
pos += 1
val |= (byte & 0x7F) << shift
if not (byte & 0x80):
break
shift += 7
entry['fields'][field_num] = ('varint', val)
elif wire_type == WIRE_FIXED32:
raw = data[pos:pos + 4]
pos += 4
try:
f = struct.unpack('<f', raw)[0]
entry['fields'][field_num] = ('float', f, raw.hex())
except:
entry['fields'][field_num] = ('fixed32', raw.hex())
elif wire_type == WIRE_FIXED64:
raw = data[pos:pos + 8]
pos += 8
try:
d = struct.unpack('<d', raw)[0]
entry['fields'][field_num] = ('double', d, raw.hex())
except:
entry['fields'][field_num] = ('fixed64', raw.hex())
elif wire_type == WIRE_LENGTH:
# Read length
length = 0
shift = 0
while pos < len(data):
byte = data[pos]
pos += 1
length |= (byte & 0x7F) << shift
if not (byte & 0x80):
break
shift += 7
raw = data[pos:pos + length]
pos += length
entry['fields'][field_num] = ('bytes', len(raw), raw.hex()[:40], raw)
else:
break
except Exception as e:
entry['parse_error'] = str(e)
break
return entry
def is_voltage_like(val: float) -> Optional[str]:
"""Check if a float value looks like a battery voltage."""
if 10.0 <= val <= 16.0:
return "12V system"
if 20.0 <= val <= 32.0:
return "24V system"
if 40.0 <= val <= 60.0:
return "48V system"
return None
def print_nested_field(raw_data: bytes, field_num: int, indent: str = " "):
"""Print a nested field with deep parsing."""
parser = ProtobufParser(raw_data)
deep_results = parser.parse_nested_deep(raw_data, str(field_num))
# Group results by depth for better display
print(f"{indent}Deep parse of Field {field_num} (length={len(raw_data)}):")
for path, vtype, value in deep_results:
depth = path.count('.')
sub_indent = indent + " " * depth
if vtype == 'float':
voltage_hint = is_voltage_like(value)
if voltage_hint:
print(f"{sub_indent}Field {path}: {value:.2f} ({vtype}) <- VOLTAGE? ({voltage_hint})")
elif value != value: # NaN
print(f"{sub_indent}Field {path}: nan ({vtype})")
else:
print(f"{sub_indent}Field {path}: {value:.4f} ({vtype})")
elif vtype == 'double':
voltage_hint = is_voltage_like(value)
if voltage_hint:
print(f"{sub_indent}Field {path}: {value:.2f} ({vtype}) <- VOLTAGE? ({voltage_hint})")
else:
print(f"{sub_indent}Field {path}: {value:.4f} ({vtype})")
elif vtype == 'varint':
print(f"{sub_indent}Field {path}: {value} ({vtype})")
else:
print(f"{sub_indent}Field {path}: {value} ({vtype})")
def scan_packet(data: bytes, group: str, port: int, target_field: Optional[int] = None):
"""Scan a packet and dump potential battery-related fields."""
if len(data) < HEADER_SIZE + 5:
return
proto_data = data[HEADER_SIZE:]
parser = ProtobufParser(proto_data)
all_fields = parser.parse_all_fields()
print(f"\n{'='*70}")
print(f"Packet from {group}:{port} (size: {len(data)} bytes)")
print(f"Top-level fields present: {sorted(all_fields.keys())}")
print(f"{'='*70}")
# If a specific field is targeted, only show that one
fields_to_check = {target_field} if target_field else BATTERY_CANDIDATE_FIELDS
# Check candidate fields for battery data
for field_num in sorted(fields_to_check):
if field_num in all_fields:
entries = all_fields[field_num]
print(f"\n Field {field_num}: {len(entries)} entries")
for i, entry in enumerate(entries):
if entry[0] == 'length':
raw_data = entry[1]
print(f"\n Entry {i+1}:")
print_nested_field(raw_data, field_num, " ")
elif entry[0] in ('float', 'double'):
voltage_hint = is_voltage_like(entry[1])
if voltage_hint:
print(f" Value: {entry[1]:.2f} <- VOLTAGE? ({voltage_hint})")
else:
print(f" Value: {entry[1]:.4f}")
else:
print(f" Value: {entry}")
# Deep scan ALL fields for voltage-like values
print(f"\n Deep scanning all fields for voltage-like values:")
found_any = False
for field_num, entries in sorted(all_fields.items()):
for entry in entries:
if entry[0] == 'length':
raw_data = entry[1]
deep_parser = ProtobufParser(raw_data)
deep_results = deep_parser.parse_nested_deep(raw_data, str(field_num))
for path, vtype, value in deep_results:
if vtype in ('float', 'double') and is_voltage_like(value):
print(f" Field {path}: {value:.2f}V ({is_voltage_like(value)})")
found_any = True
elif entry[0] == 'float' and is_voltage_like(entry[1]):
print(f" Field {field_num}: {entry[1]:.2f}V ({is_voltage_like(entry[1])})")
found_any = True
elif entry[0] == 'double' and is_voltage_like(entry[1]):
print(f" Field {field_num}: {entry[1]:.2f}V ({is_voltage_like(entry[1])})")
found_any = True
if not found_any:
print(f" (no voltage-like values found)")
def main():
import argparse
parser = argparse.ArgumentParser(description="Debug potential battery data fields")
parser.add_argument('-i', '--interface', required=True, help='Interface IP')
parser.add_argument('-t', '--time', type=int, default=5, help='Capture time (seconds)')
parser.add_argument('-g', '--group', default="226.192.206.102", help='Multicast group')
parser.add_argument('-p', '--port', type=int, default=2565, help='Port')
parser.add_argument('-f', '--field', type=int, help='Focus on specific field number (e.g., 14)')
args = parser.parse_args()
print(f"Scanning for battery data fields...")
if args.field:
print(f"Focusing on Field {args.field} with deep nesting")
else:
print(f"Looking at fields: {sorted(BATTERY_CANDIDATE_FIELDS)}")
print(f"Target voltages: Aft House ~26.3V, Stern House ~27.2V, Port Engine ~26.5V")
print(f"\nCapturing from {args.group}:{args.port} for {args.time} seconds...")
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
if hasattr(socket, 'SO_REUSEPORT'):
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
sock.bind(('', args.port))
mreq = struct.pack("4s4s", socket.inet_aton(args.group), socket.inet_aton(args.interface))
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
sock.settimeout(1.0)
seen_sizes = set()
end_time = time.time() + args.time
try:
while time.time() < end_time:
try:
data, _ = sock.recvfrom(65535)
# Only process each unique packet size once
if len(data) not in seen_sizes:
seen_sizes.add(len(data))
scan_packet(data, args.group, args.port, args.field)
except socket.timeout:
continue
except KeyboardInterrupt:
pass
finally:
sock.close()
print("\n\nDone.")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink