From a2387341bdb08c41a64ade29261eee9ec89db0f1 Mon Sep 17 00:00:00 2001 From: Bastian Wagner Date: Sat, 15 Aug 2026 09:04:38 +0200 Subject: [PATCH] reference --- reference/fit_crc.py | 39 ++++ reference/fit_device.py | 381 +++++++++++++++++++++++++++++++++++ reference/garmin_uploader.py | 136 +++++++++++++ 3 files changed, 556 insertions(+) create mode 100644 reference/fit_crc.py create mode 100644 reference/fit_device.py create mode 100644 reference/garmin_uploader.py diff --git a/reference/fit_crc.py b/reference/fit_crc.py new file mode 100644 index 0000000..633c347 --- /dev/null +++ b/reference/fit_crc.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +CRC_TABLE = ( + 0x0000, + 0xCC01, + 0xD801, + 0x1400, + 0xF001, + 0x3C00, + 0x2800, + 0xE401, + 0xA001, + 0x6C00, + 0x7800, + 0xB401, + 0x5000, + 0x9C01, + 0x8801, + 0x4400, +) + + +def update_crc(crc: int, byte: int) -> int: + tmp = CRC_TABLE[crc & 0xF] + crc = (crc >> 4) & 0x0FFF + crc = crc ^ tmp ^ CRC_TABLE[byte & 0xF] + + tmp = CRC_TABLE[crc & 0xF] + crc = (crc >> 4) & 0x0FFF + crc = crc ^ tmp ^ CRC_TABLE[(byte >> 4) & 0xF] + return crc & 0xFFFF + + +def fit_crc(data: bytes | bytearray | memoryview) -> int: + crc = 0 + for byte in data: + crc = update_crc(crc, byte) + return crc + diff --git a/reference/fit_device.py b/reference/fit_device.py new file mode 100644 index 0000000..f278580 --- /dev/null +++ b/reference/fit_device.py @@ -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(" 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("> 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(" int: + file_crc = fit_crc(data[:-2]) + struct.pack_into(" Any: + ... + + def upload_activity(self, activity_path: str) -> Any: + ... + + +@dataclass(frozen=True) +class UploadResult: + status: str + duplicate: bool + garmin_activity_id: str | None + raw_response: Any + + +class GarminUploadBlocked(RuntimeError): + """Raised when Garmin login needs user action such as MFA.""" + + +class GarminUploader: + def __init__( + self, + settings: Settings, + client_factory: Callable[..., GarminClientProtocol] | None = None, + ) -> None: + self.settings = settings + self._client_factory = client_factory + self._client: GarminClientProtocol | None = None + + def upload(self, fit_path: Path) -> UploadResult: + client = self._ensure_client() + try: + response = client.upload_activity(str(fit_path)) + except Exception as exc: + if _looks_duplicate_error(exc): + logger.info("Garmin already has activity for %s", fit_path) + return UploadResult( + status="duplicate", + duplicate=True, + garmin_activity_id=None, + raw_response=str(exc), + ) + raise + + return UploadResult( + status="uploaded", + duplicate=False, + garmin_activity_id=_extract_activity_id(response), + raw_response=response, + ) + + def _ensure_client(self) -> GarminClientProtocol: + if self._client is not None: + return self._client + + factory = self._client_factory or _default_garmin_factory + client = factory( + self.settings.garmin_email, + self.settings.garmin_password, + prompt_mfa=self._prompt_mfa, + ) + try: + client.login(str(self.settings.garmin_tokenstore)) + except RuntimeError: + raise + except Exception as exc: + if "mfa" in str(exc).lower(): + raise GarminUploadBlocked( + "Garmin MFA is required. Set GARMIN_MFA_CODE for one run." + ) from exc + raise + self._client = client + return client + + def _prompt_mfa(self) -> str: + if self.settings.garmin_mfa_code: + return self.settings.garmin_mfa_code + raise GarminUploadBlocked( + "Garmin requested MFA but GARMIN_MFA_CODE is not set." + ) + + +def _default_garmin_factory(*args: Any, **kwargs: Any) -> GarminClientProtocol: + from garminconnect import Garmin + + return Garmin(*args, **kwargs) + + +def _looks_duplicate_error(exc: Exception) -> bool: + text = str(exc).lower() + return any(token in text for token in ("duplicate", "already exists", "409")) + + +def _extract_activity_id(response: Any) -> str | None: + if not isinstance(response, dict): + return None + + candidates = [ + response.get("activityId"), + response.get("activity_id"), + response.get("id"), + ] + detailed_import = response.get("detailedImportResult") + if isinstance(detailed_import, dict): + candidates.extend( + [ + detailed_import.get("uploadId"), + detailed_import.get("activityId"), + ] + ) + + for key in ("successes", "success", "importedActivities"): + items = response.get(key) + if isinstance(items, list) and items: + first = items[0] + if isinstance(first, dict): + candidates.extend([first.get("activityId"), first.get("id")]) + + for candidate in candidates: + if candidate is not None: + return str(candidate) + return None +