feat: patch FIT creator as Edge 1030 Plus

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

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("<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

tests/fit/test_rewriter_patching.py

@@ -0,0 +1,150 @@
+import struct
+from pathlib import Path
+
+import pytest
+
+from app.fit.models import DeviceFieldValue, FitConversionResult, GarminDevice
+from app.fit.rewriter import (
+    FitFormatError,
+    _iter_data_fields,
+    convert_fit_device,
+    read_device_field_values,
+)
+from tests.fit.builders import data, definition, make_fit
+
+FILE_ID_MESG_NUM = 0
+DEVICE_INFO_MESG_NUM = 23
+
+
+def _build_fixture() -> 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("<HH", 255, 999))
+
+    # device_info: device_index(0/u8), manufacturer(2/u16), product(4/u16)
+    device_def = definition(1, DEVICE_INFO_MESG_NUM, [(0, 1, 0x02), (2, 2, 0x84), (4, 2, 0x84)])
+    creator = data(1, struct.pack("<BHH", 0, 255, 999))
+    sensor = data(1, struct.pack("<BHH", 1, 32, 1234))
+
+    records = file_def + file_data + device_def + creator + sensor
+    return make_fit(records)
+
+
+def _build_no_device_index_fixture() -> 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("<HH", 255, 999))
+
+    # device_info WITHOUT device_index field at all: manufacturer(2/u16), product(4/u16)
+    device_def = definition(1, DEVICE_INFO_MESG_NUM, [(2, 2, 0x84), (4, 2, 0x84)])
+    no_index_device = data(1, struct.pack("<HH", 32, 1234))
+
+    records = file_def + file_data + device_def + no_index_device
+    return make_fit(records)
+
+
+def _read_device_info_records(path: Path) -> 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)