diff --git a/app/fit/models.py b/app/fit/models.py index 0b27e81..4b85484 100644 --- a/app/fit/models.py +++ b/app/fit/models.py @@ -1,4 +1,5 @@ from dataclasses import dataclass +from pathlib import Path @dataclass(frozen=True) @@ -14,3 +15,30 @@ class LocalDefinition: endian: str fields: tuple[FieldDefinition, ...] developer_field_size: int + + +GARMIN_MANUFACTURER_ID = 1 + + +@dataclass(frozen=True) +class GarminDevice: + manufacturer_id: int = GARMIN_MANUFACTURER_ID + product_id: int = 3570 + 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 DeviceFieldValue: + global_message_num: int + field_num: int + value: int | str diff --git a/app/fit/rewriter.py b/app/fit/rewriter.py index d87518c..2680c8d 100644 --- a/app/fit/rewriter.py +++ b/app/fit/rewriter.py @@ -2,7 +2,16 @@ import struct from pathlib import Path from app.fit.crc import fit_crc -from app.fit.models import FieldDefinition, LocalDefinition +from app.fit.models import ( + DeviceFieldValue, + FieldDefinition, + FitConversionResult, + GarminDevice, + LocalDefinition, +) + +FILE_ID_MESG_NUM = 0 +DEVICE_INFO_MESG_NUM = 23 class FitFormatError(ValueError): @@ -141,3 +150,155 @@ def is_fit_file(path: Path) -> bool: except (OSError, FitFormatError): return False return True + + +def convert_fit_device( + source_path: Path, output_path: Path, device: GarminDevice | None = None +) -> FitConversionResult: + resolved = device or GarminDevice() + data = bytearray(source_path.read_bytes()) + _validate_fit_container(data) + patched_count = _patch_device_metadata(data, resolved) + header_crc = _rewrite_header_crc(data) + file_crc = _rewrite_file_crc(data) + _validate_fit_container(data) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(data) + return FitConversionResult(source_path, output_path, patched_count, header_crc, file_crc) + + +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 _patch_device_metadata(data: bytearray, device: GarminDevice) -> int: + patched_count = 0 + eligible_field_count = 0 + for definition, field_offsets in _iter_data_fields(data): + field_map = {field.num: (field, offset) for field, offset in field_offsets} + if definition.global_message_num == DEVICE_INFO_MESG_NUM: + device_index_entry = field_map.get(0) + if device_index_entry is None: + continue + index_field, index_offset = device_index_entry + device_index = _read_field_value(data, index_offset, index_field, definition.endian) + if device_index != 0: + continue + + 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 _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(" bytes: + # file_id: manufacturer(1/u16), product(2/u16) + file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)]) + file_data = data(0, struct.pack(" bytes: + """device_info record entirely missing field 0 (device_index) must be left untouched.""" + file_def = definition(0, FILE_ID_MESG_NUM, [(1, 2, 0x84), (2, 2, 0x84)]) + file_data = data(0, struct.pack(" list[dict[int, int]]: + """Parse raw bytes and return one dict of {field_num: value} per device_info record, + in file order, so creator and sensor records can be distinguished positionally.""" + raw = bytearray(path.read_bytes()) + records: list[dict[int, int]] = [] + for local_def, field_offsets in _iter_data_fields(raw): + if local_def.global_message_num != DEVICE_INFO_MESG_NUM: + continue + values: dict[int, int] = {} + for field, offset in field_offsets: + raw_bytes = bytes(raw[offset : offset + field.size]) + if field.size == 1: + values[field.num] = raw_bytes[0] + elif field.size == 2: + values[field.num] = struct.unpack(f"{local_def.endian}H", raw_bytes)[0] + records.append(values) + return records + + +def test_convert_fit_device_patches_file_id(tmp_path: Path) -> None: + source = tmp_path / "source.fit" + source.write_bytes(_build_fixture()) + output = tmp_path / "output.fit" + + result = convert_fit_device(source, output) + + assert isinstance(result, FitConversionResult) + assert result.source_path == source + assert result.output_path == output + assert result.file_crc is not None + assert result.patched_field_count > 0 + + values = read_device_field_values(output) + values_by_key = {(v.global_message_num, v.field_num): v.value for v in values} + + assert values_by_key[(FILE_ID_MESG_NUM, 1)] == 1 + assert values_by_key[(FILE_ID_MESG_NUM, 2)] == 3570 + + +def test_convert_fit_device_patches_creator_and_leaves_sensor_untouched( + tmp_path: Path, +) -> None: + source = tmp_path / "source.fit" + source.write_bytes(_build_fixture()) + output = tmp_path / "output.fit" + + convert_fit_device(source, output) + + device_info_records = _read_device_info_records(output) + assert len(device_info_records) == 2 + creator_values, sensor_values = device_info_records + + # creator (device_index=0) patched to Edge 1030 Plus + assert creator_values[0] == 0 + assert creator_values[2] == 1 + assert creator_values[4] == 3570 + + # sensor (device_index=1) must remain completely untouched + assert sensor_values[0] == 1 + assert sensor_values[2] == 32 + assert sensor_values[4] == 1234 + + +def test_convert_fit_device_leaves_device_info_without_device_index_untouched( + tmp_path: Path, +) -> None: + source = tmp_path / "source.fit" + source.write_bytes(_build_no_device_index_fixture()) + output = tmp_path / "output.fit" + + convert_fit_device(source, output) + + device_info_records = _read_device_info_records(output) + assert len(device_info_records) == 1 + values = device_info_records[0] + + # device_info without a device_index field must remain completely untouched + assert values[2] == 32 + assert values[4] == 1234 + + +def test_convert_fit_device_defaults_to_edge_1030_plus() -> None: + device = GarminDevice() + assert device.manufacturer_id == 1 + assert device.product_id == 3570 + assert device.product_name == "Edge 1030 Plus" + + +def test_convert_fit_device_rejects_invalid_container(tmp_path: Path) -> None: + source = tmp_path / "bad.fit" + source.write_bytes(b"not a fit file") + output = tmp_path / "output.fit" + + with pytest.raises(FitFormatError): + convert_fit_device(source, output) + + +def test_read_device_field_values_returns_device_field_value_instances(tmp_path: Path) -> None: + source = tmp_path / "source.fit" + source.write_bytes(_build_fixture()) + + values = read_device_field_values(source) + + assert all(isinstance(v, DeviceFieldValue) for v in values) + assert any(v.global_message_num == FILE_ID_MESG_NUM for v in values)