reference
This commit is contained in:
Bastian Wagner
381
reference/fit_device.py
Normal file
381
reference/fit_device.py
Normal file
@@ -0,0 +1,381 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from .fit_crc import fit_crc
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
FILE_ID_MESG_NUM = 0
|
||||
DEVICE_INFO_MESG_NUM = 23
|
||||
GARMIN_MANUFACTURER_ID = 1
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GarminDevice:
|
||||
manufacturer_id: int = GARMIN_MANUFACTURER_ID
|
||||
product_id: int = 3578
|
||||
product_name: str = "Edge 1030 Plus"
|
||||
serial_number: int | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FitConversionResult:
|
||||
source_path: Path
|
||||
output_path: Path
|
||||
patched_field_count: int
|
||||
header_crc: int | None
|
||||
file_crc: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class FieldDefinition:
|
||||
num: int
|
||||
size: int
|
||||
base_type: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class LocalDefinition:
|
||||
global_message_num: int
|
||||
endian: str
|
||||
fields: tuple[FieldDefinition, ...]
|
||||
record_size: int
|
||||
developer_field_size: int
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DeviceFieldValue:
|
||||
global_message_num: int
|
||||
field_num: int
|
||||
value: int | str
|
||||
|
||||
|
||||
class FitFormatError(ValueError):
|
||||
"""Raised when a file is not a valid enough FIT file for metadata patching."""
|
||||
|
||||
|
||||
def convert_fit_device(
|
||||
source_path: Path, output_path: Path, device: GarminDevice | None = None
|
||||
) -> FitConversionResult:
|
||||
device = device or GarminDevice()
|
||||
data = bytearray(source_path.read_bytes())
|
||||
_validate_fit_container(data)
|
||||
|
||||
patched_count = _patch_device_metadata(data, device)
|
||||
header_crc = _rewrite_header_crc(data)
|
||||
file_crc = _rewrite_file_crc(data)
|
||||
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(data)
|
||||
logger.info(
|
||||
"Converted FIT metadata for %s -> %s; patched_fields=%d",
|
||||
source_path,
|
||||
output_path,
|
||||
patched_count,
|
||||
)
|
||||
return FitConversionResult(
|
||||
source_path=source_path,
|
||||
output_path=output_path,
|
||||
patched_field_count=patched_count,
|
||||
header_crc=header_crc,
|
||||
file_crc=file_crc,
|
||||
)
|
||||
|
||||
|
||||
def is_fit_file(path: Path) -> bool:
|
||||
try:
|
||||
data = path.read_bytes()
|
||||
_validate_fit_container(data)
|
||||
except (OSError, FitFormatError):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def read_device_field_values(path: Path) -> list[DeviceFieldValue]:
|
||||
data = bytearray(path.read_bytes())
|
||||
_validate_fit_container(data)
|
||||
values: list[DeviceFieldValue] = []
|
||||
for definition, field_offsets in _iter_data_fields(data):
|
||||
for field, offset in field_offsets:
|
||||
if definition.global_message_num == FILE_ID_MESG_NUM and field.num in {
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
8,
|
||||
}:
|
||||
values.append(
|
||||
DeviceFieldValue(
|
||||
definition.global_message_num,
|
||||
field.num,
|
||||
_read_field_value(data, offset, field, definition.endian),
|
||||
)
|
||||
)
|
||||
if definition.global_message_num == DEVICE_INFO_MESG_NUM and field.num in {
|
||||
2,
|
||||
3,
|
||||
4,
|
||||
27,
|
||||
}:
|
||||
values.append(
|
||||
DeviceFieldValue(
|
||||
definition.global_message_num,
|
||||
field.num,
|
||||
_read_field_value(data, offset, field, definition.endian),
|
||||
)
|
||||
)
|
||||
return values
|
||||
|
||||
|
||||
def _validate_fit_container(data: bytearray) -> None:
|
||||
if len(data) < 14:
|
||||
raise FitFormatError("FIT file is too small")
|
||||
|
||||
header_size = data[0]
|
||||
if header_size not in {12, 14}:
|
||||
raise FitFormatError(f"Unsupported FIT header size: {header_size}")
|
||||
|
||||
if len(data) < header_size + 2:
|
||||
raise FitFormatError("FIT file is shorter than its header")
|
||||
|
||||
if bytes(data[8:12]) != b".FIT":
|
||||
raise FitFormatError("Missing .FIT signature")
|
||||
|
||||
data_size = struct.unpack_from("<I", data, 4)[0]
|
||||
expected_size = header_size + data_size + 2
|
||||
if len(data) != expected_size:
|
||||
raise FitFormatError(
|
||||
f"FIT size mismatch: header says {expected_size} bytes, file has {len(data)}"
|
||||
)
|
||||
|
||||
if header_size == 14:
|
||||
expected_header_crc = struct.unpack_from("<H", data, 12)[0]
|
||||
actual_header_crc = fit_crc(data[:12])
|
||||
if expected_header_crc != actual_header_crc:
|
||||
raise FitFormatError("FIT header CRC check failed")
|
||||
|
||||
expected_file_crc = struct.unpack_from("<H", data, len(data) - 2)[0]
|
||||
actual_file_crc = fit_crc(data[:-2])
|
||||
if expected_file_crc != actual_file_crc:
|
||||
raise FitFormatError("FIT file CRC check failed")
|
||||
|
||||
|
||||
def _patch_device_metadata(data: bytearray, device: GarminDevice) -> int:
|
||||
patched_count = 0
|
||||
eligible_field_count = 0
|
||||
for definition, field_offsets in _iter_data_fields(data):
|
||||
for field, offset in field_offsets:
|
||||
target_value: int | str | None = None
|
||||
if definition.global_message_num == FILE_ID_MESG_NUM:
|
||||
if field.num == 1:
|
||||
target_value = device.manufacturer_id
|
||||
elif field.num == 2:
|
||||
target_value = device.product_id
|
||||
elif field.num == 3 and device.serial_number is not None:
|
||||
target_value = device.serial_number
|
||||
elif field.num == 8:
|
||||
target_value = device.product_name
|
||||
elif definition.global_message_num == DEVICE_INFO_MESG_NUM:
|
||||
if field.num == 2:
|
||||
target_value = device.manufacturer_id
|
||||
elif field.num == 3 and device.serial_number is not None:
|
||||
target_value = device.serial_number
|
||||
elif field.num == 4:
|
||||
target_value = device.product_id
|
||||
elif field.num == 27:
|
||||
target_value = device.product_name
|
||||
|
||||
if target_value is not None:
|
||||
eligible_field_count += 1
|
||||
if _write_field_value(data, offset, field, definition.endian, target_value):
|
||||
patched_count += 1
|
||||
|
||||
if eligible_field_count == 0:
|
||||
raise FitFormatError("No writable file_id or device_info device fields found")
|
||||
return patched_count
|
||||
|
||||
|
||||
def _iter_data_fields(
|
||||
data: bytearray,
|
||||
) -> list[tuple[LocalDefinition, list[tuple[FieldDefinition, int]]]]:
|
||||
header_size = data[0]
|
||||
data_size = struct.unpack_from("<I", data, 4)[0]
|
||||
offset = header_size
|
||||
end_offset = header_size + data_size
|
||||
definitions: dict[int, LocalDefinition] = {}
|
||||
data_records: list[tuple[LocalDefinition, list[tuple[FieldDefinition, int]]]] = []
|
||||
|
||||
while offset < end_offset:
|
||||
record_header = data[offset]
|
||||
offset += 1
|
||||
|
||||
if record_header & 0x80:
|
||||
local_message_type = (record_header >> 5) & 0x03
|
||||
definition = definitions.get(local_message_type)
|
||||
if definition is None:
|
||||
raise FitFormatError(
|
||||
f"Compressed timestamp record used unknown local definition {local_message_type}"
|
||||
)
|
||||
field_offsets, offset = _collect_field_offsets(definition, offset)
|
||||
data_records.append((definition, field_offsets))
|
||||
continue
|
||||
|
||||
local_message_type = record_header & 0x0F
|
||||
is_definition = bool(record_header & 0x40)
|
||||
has_developer_fields = bool(record_header & 0x20)
|
||||
|
||||
if is_definition:
|
||||
definition, offset = _read_definition(
|
||||
data, offset, local_message_type, has_developer_fields
|
||||
)
|
||||
definitions[local_message_type] = definition
|
||||
continue
|
||||
|
||||
definition = definitions.get(local_message_type)
|
||||
if definition is None:
|
||||
raise FitFormatError(
|
||||
f"Data record used unknown local definition {local_message_type}"
|
||||
)
|
||||
field_offsets, offset = _collect_field_offsets(definition, offset)
|
||||
data_records.append((definition, field_offsets))
|
||||
|
||||
if offset != end_offset:
|
||||
raise FitFormatError("FIT parser did not end on data boundary")
|
||||
return data_records
|
||||
|
||||
|
||||
def _read_definition(
|
||||
data: bytearray,
|
||||
offset: int,
|
||||
local_message_type: int,
|
||||
has_developer_fields: bool,
|
||||
) -> tuple[LocalDefinition, int]:
|
||||
del local_message_type
|
||||
if offset + 5 > len(data):
|
||||
raise FitFormatError("Truncated FIT definition message")
|
||||
|
||||
offset += 1
|
||||
architecture = data[offset]
|
||||
offset += 1
|
||||
endian = ">" if architecture == 1 else "<"
|
||||
global_message_num = struct.unpack_from(f"{endian}H", data, offset)[0]
|
||||
offset += 2
|
||||
field_count = data[offset]
|
||||
offset += 1
|
||||
|
||||
fields: list[FieldDefinition] = []
|
||||
record_size = 0
|
||||
for _ in range(field_count):
|
||||
if offset + 3 > len(data):
|
||||
raise FitFormatError("Truncated FIT field definition")
|
||||
field = FieldDefinition(
|
||||
num=data[offset],
|
||||
size=data[offset + 1],
|
||||
base_type=data[offset + 2],
|
||||
)
|
||||
fields.append(field)
|
||||
record_size += field.size
|
||||
offset += 3
|
||||
|
||||
developer_field_size = 0
|
||||
if has_developer_fields:
|
||||
if offset >= len(data):
|
||||
raise FitFormatError("Truncated FIT developer field count")
|
||||
developer_field_count = data[offset]
|
||||
offset += 1
|
||||
for _ in range(developer_field_count):
|
||||
if offset + 3 > len(data):
|
||||
raise FitFormatError("Truncated FIT developer fields")
|
||||
developer_field_size += data[offset + 1]
|
||||
offset += 3
|
||||
record_size += developer_field_size
|
||||
|
||||
return (
|
||||
LocalDefinition(
|
||||
global_message_num=global_message_num,
|
||||
endian=endian,
|
||||
fields=tuple(fields),
|
||||
record_size=record_size,
|
||||
developer_field_size=developer_field_size,
|
||||
),
|
||||
offset,
|
||||
)
|
||||
|
||||
|
||||
def _collect_field_offsets(
|
||||
definition: LocalDefinition, offset: int
|
||||
) -> tuple[list[tuple[FieldDefinition, int]], int]:
|
||||
field_offsets: list[tuple[FieldDefinition, int]] = []
|
||||
current_offset = offset
|
||||
for field in definition.fields:
|
||||
field_offsets.append((field, current_offset))
|
||||
current_offset += field.size
|
||||
current_offset += definition.developer_field_size
|
||||
return field_offsets, current_offset
|
||||
|
||||
|
||||
def _read_field_value(
|
||||
data: bytearray, offset: int, field: FieldDefinition, endian: str
|
||||
) -> int | str:
|
||||
base_type = field.base_type & 0x1F
|
||||
if base_type in {0x03, 0x04, 0x0B} and field.size >= 2:
|
||||
return struct.unpack_from(f"{endian}H", data, offset)[0]
|
||||
if base_type in {0x05, 0x06, 0x0C} and field.size >= 4:
|
||||
return struct.unpack_from(f"{endian}I", data, offset)[0]
|
||||
if base_type == 0x07:
|
||||
raw = bytes(data[offset : offset + field.size])
|
||||
if 0 in raw:
|
||||
raw = raw[: raw.index(0)]
|
||||
return raw.decode("utf-8", errors="replace")
|
||||
raw = bytes(data[offset : offset + field.size])
|
||||
return int.from_bytes(raw, "little")
|
||||
|
||||
|
||||
def _write_field_value(
|
||||
data: bytearray,
|
||||
offset: int,
|
||||
field: FieldDefinition,
|
||||
endian: str,
|
||||
value: int | str,
|
||||
) -> bool:
|
||||
if isinstance(value, str):
|
||||
encoded = value.encode("utf-8")
|
||||
if not encoded or field.size == 0 or len(encoded) + 1 > field.size:
|
||||
return False
|
||||
replacement = encoded + b"\x00" + b"\x00" * (field.size - len(encoded) - 1)
|
||||
if bytes(data[offset : offset + field.size]) == replacement:
|
||||
return False
|
||||
data[offset : offset + field.size] = replacement
|
||||
return True
|
||||
|
||||
if field.size == 1:
|
||||
replacement = struct.pack("B", value)
|
||||
elif field.size == 2:
|
||||
replacement = struct.pack(f"{endian}H", value)
|
||||
elif field.size == 4:
|
||||
replacement = struct.pack(f"{endian}I", value)
|
||||
else:
|
||||
return False
|
||||
|
||||
if bytes(data[offset : offset + field.size]) == replacement:
|
||||
return False
|
||||
data[offset : offset + field.size] = replacement
|
||||
return True
|
||||
|
||||
|
||||
def _rewrite_header_crc(data: bytearray) -> int | None:
|
||||
header_size = data[0]
|
||||
if header_size != 14:
|
||||
return None
|
||||
header_crc = fit_crc(data[:12])
|
||||
struct.pack_into("<H", data, 12, header_crc)
|
||||
return header_crc
|
||||
|
||||
|
||||
def _rewrite_file_crc(data: bytearray) -> int:
|
||||
file_crc = fit_crc(data[:-2])
|
||||
struct.pack_into("<H", data, len(data) - 2, file_crc)
|
||||
return file_crc
|
||||
Reference in New Issue
Block a user