mywhoosh2garmin/docs/superpowers/plans/2026-08-15-fit-rewriter.md

FIT Edge 1030 Plus Rewriter Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Implement a binary-preserving FIT metadata rewriter that changes only Garmin creator-device metadata plus required CRC bytes, targeting Garmin Edge 1030 Plus product ID 3570.

Architecture: Parse FIT definition/data records only far enough to locate file_id and creator device_info fields, patch bytes in place, recalculate header/file CRC, and validate the output. Do not fully decode/re-encode the activity, so unknown/developer fields and all non-target bytes remain untouched.

Tech Stack: Python 3.12 standard library (struct, dataclasses, pathlib), pytest.

Global Constraints

• Garmin manufacturer ID is 1.
• Garmin Edge 1030 Plus product ID is 3570.
• Product name is Edge 1030 Plus.
• Support FIT header sizes 12 and 14.
• Validate declared data size and .FIT signature.
• Validate header CRC when a 14-byte header is present.
• Validate file CRC before and after modification.
• Support little- and big-endian definition architectures, compressed timestamp records, developer fields, and changing local message definitions.
• Patch file_id device fields when present.
• Patch device_info only when it is safely identified as the creator (device_index == 0).
• If creator-specific device_info cannot be identified safely, leave it unchanged rather than rewriting all device records.
• Preserve every non-target byte except CRC fields.

---

File Structure

app/fit/
  __init__.py
  crc.py
  models.py
  rewriter.py
tests/fit/
  builders.py
  test_crc.py
  test_rewriter_validation.py
  test_rewriter_patching.py
  test_rewriter_preservation.py

Task 1: Implement Garmin FIT CRC

Files:

• Create: app/fit/crc.py
• Create: tests/fit/test_crc.py

Interfaces:

• Produces: fit_crc(data: bytes | bytearray | memoryview) -> int.

• Step 1: Write failing CRC vector tests

# tests/fit/test_crc.py
from app.fit.crc import fit_crc


def test_empty_crc_is_zero() -> None:
    assert fit_crc(b"") == 0


def test_crc_is_incremental_equivalent() -> None:
    payload = b".FIT-device-metadata"
    assert fit_crc(payload) == fit_crc(memoryview(payload))
    assert 0 <= fit_crc(payload) <= 0xFFFF

• Step 2: Run and verify failure

Run: pytest tests/fit/test_crc.py -v

Expected: import failure.

• Step 3: Implement the FIT nibble-table CRC algorithm

# app/fit/crc.py
CRC_TABLE = (
    0x0000, 0xCC01, 0xD801, 0x1400,
    0xF001, 0x3C00, 0x2800, 0xE401,
    0xA001, 0x6C00, 0x7800, 0xB401,
    0x5000, 0x9C01, 0x8801, 0x4400,
)


def fit_crc(data: bytes | bytearray | memoryview) -> int:
    crc = 0
    for byte in data:
        tmp = CRC_TABLE[crc & 0xF]
        crc = (crc >> 4) & 0x0FFF
        crc ^= tmp ^ CRC_TABLE[byte & 0xF]
        tmp = CRC_TABLE[crc & 0xF]
        crc = (crc >> 4) & 0x0FFF
        crc ^= tmp ^ CRC_TABLE[(byte >> 4) & 0xF]
    return crc & 0xFFFF

• Step 4: Run tests

Run: pytest tests/fit/test_crc.py -v

Expected: PASS.

• Step 5: Commit

git add app/fit/crc.py tests/fit/test_crc.py
git commit -m "feat: add FIT CRC calculation"

Task 2: Parse and validate the FIT container safely

Files:

• Create: app/fit/models.py
• Create: app/fit/rewriter.py
• Create: tests/fit/builders.py
• Create: tests/fit/test_rewriter_validation.py

Interfaces:

• Produces: FitFormatError, FieldDefinition, LocalDefinition, _validate_fit_container(), _iter_data_fields().

• _iter_data_fields(data) returns data records with exact field offsets without decoding unrelated values.

• Step 1: Create a deterministic minimal FIT fixture builder

# tests/fit/builders.py
import struct

from app.fit.crc import fit_crc


def make_fit(data_records: bytes, *, header_size: int = 14) -> bytes:
    if header_size not in {12, 14}:
        raise ValueError(header_size)
    header = bytearray(header_size)
    header[0] = header_size
    header[1] = 0x20
    struct.pack_into("<H", header, 2, 0x0100)
    struct.pack_into("<I", header, 4, len(data_records))
    header[8:12] = b".FIT"
    if header_size == 14:
        struct.pack_into("<H", header, 12, fit_crc(header[:12]))
    body = header + data_records
    return bytes(body + struct.pack("<H", fit_crc(body)))


def definition(local: int, global_num: int, fields: list[tuple[int, int, int]], *, endian: str = "<") -> bytes:
    architecture = 1 if endian == ">" else 0
    payload = bytearray([0, architecture])
    payload.extend(struct.pack(f"{endian}H", global_num))
    payload.append(len(fields))
    for num, size, base_type in fields:
        payload.extend(bytes([num, size, base_type]))
    return bytes([0x40 | local]) + bytes(payload)


def data(local: int, payload: bytes) -> bytes:
    return bytes([local]) + payload

• Step 2: Write validation tests

# tests/fit/test_rewriter_validation.py
from pathlib import Path

import pytest

from app.fit.rewriter import FitFormatError, is_fit_file
from tests.fit.builders import make_fit


def test_valid_12_and_14_byte_headers(tmp_path: Path) -> None:
    for size in (12, 14):
        path = tmp_path / f"valid-{size}.fit"
        path.write_bytes(make_fit(b"", header_size=size))
        assert is_fit_file(path) is True


def test_bad_file_crc_is_rejected(tmp_path: Path) -> None:
    payload = bytearray(make_fit(b""))
    payload[-1] ^= 0xFF
    path = tmp_path / "bad.fit"
    path.write_bytes(payload)
    assert is_fit_file(path) is False

• Step 3: Run and verify failure

Run: pytest tests/fit/test_rewriter_validation.py -v

Expected: import failure.

• Step 4: Implement the parser structures and validation

# app/fit/models.py
from dataclasses import dataclass


@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, ...]
    developer_field_size: int

Implement in app/fit/rewriter.py the validated logic from the known working implementation:

class FitFormatError(ValueError):
    pass


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]
        if expected_header_crc != fit_crc(data[:12]):
            raise FitFormatError("FIT header CRC check failed")
    expected_file_crc = struct.unpack_from("<H", data, len(data) - 2)[0]
    if expected_file_crc != fit_crc(data[:-2]):
        raise FitFormatError("FIT file CRC check failed")

Implement the parser helpers explicitly:

def _read_definition(
    data: bytearray,
    offset: int,
    has_developer_fields: bool,
    end_offset: int,
) -> tuple[LocalDefinition, int]:
    if offset + 5 > end_offset:
        raise FitFormatError("Truncated FIT definition message")
    offset += 1  # reserved byte
    architecture = data[offset]
    offset += 1
    if architecture not in {0, 1}:
        raise FitFormatError(f"Unsupported FIT architecture: {architecture}")
    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] = []
    for _ in range(field_count):
        if offset + 3 > end_offset:
            raise FitFormatError("Truncated FIT field definition")
        fields.append(FieldDefinition(data[offset], data[offset + 1], data[offset + 2]))
        offset += 3

    developer_field_size = 0
    if has_developer_fields:
        if offset >= end_offset:
            raise FitFormatError("Truncated FIT developer field count")
        developer_count = data[offset]
        offset += 1
        for _ in range(developer_count):
            if offset + 3 > end_offset:
                raise FitFormatError("Truncated FIT developer field definition")
            developer_field_size += data[offset + 1]
            offset += 3

    return LocalDefinition(global_message_num, endian, tuple(fields), developer_field_size), offset


def _collect_field_offsets(
    definition: LocalDefinition,
    offset: int,
    end_offset: int,
) -> tuple[list[tuple[FieldDefinition, int]], int]:
    result: list[tuple[FieldDefinition, int]] = []
    current = offset
    for field in definition.fields:
        if current + field.size > end_offset:
            raise FitFormatError("Truncated FIT data record")
        result.append((field, current))
        current += field.size
    if current + definition.developer_field_size > end_offset:
        raise FitFormatError("Truncated FIT developer field payload")
    current += definition.developer_field_size
    return result, current


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] = {}
    records: list[tuple[LocalDefinition, list[tuple[FieldDefinition, int]]]] = []

    while offset < end_offset:
        record_header = data[offset]
        offset += 1

        if record_header & 0x80:
            local = (record_header >> 5) & 0x03
            definition = definitions.get(local)
            if definition is None:
                raise FitFormatError(f"Compressed timestamp record used unknown local definition {local}")
            field_offsets, offset = _collect_field_offsets(definition, offset, end_offset)
            records.append((definition, field_offsets))
            continue

        local = 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, has_developer_fields, end_offset)
            definitions[local] = definition
            continue

        definition = definitions.get(local)
        if definition is None:
            raise FitFormatError(f"Data record used unknown local definition {local}")
        field_offsets, offset = _collect_field_offsets(definition, offset, end_offset)
        records.append((definition, field_offsets))

    if offset != end_offset:
        raise FitFormatError("FIT parser did not end on data boundary")
    return records

• Step 5: Expose is_fit_file and verify parser boundary failures

def is_fit_file(path: Path) -> bool:
    try:
        data = bytearray(path.read_bytes())
        _validate_fit_container(data)
        _iter_data_fields(data)
    except (OSError, FitFormatError):
        return False
    return True

• Step 6: Run validation tests

Run: pytest tests/fit/test_rewriter_validation.py -v

Expected: PASS.

• Step 7: Commit

git add app/fit/models.py app/fit/rewriter.py tests/fit/builders.py tests/fit/test_rewriter_validation.py
git commit -m "feat: parse and validate FIT containers"

Task 3: Patch file_id and only creator device_info

Files:

• Modify: app/fit/models.py
• Modify: app/fit/rewriter.py
• Create: tests/fit/test_rewriter_patching.py

Interfaces:

• Produces: GarminDevice, FitConversionResult, DeviceFieldValue, convert_fit_device(), read_device_field_values().

• Default device is Garmin manufacturer 1, Edge 1030 Plus product 3570, name Edge 1030 Plus.

• Step 1: Add model types

# app/fit/models.py additions
from pathlib import Path

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

• Step 2: Write a fixture containing one creator and one sensor device_info

Use these FIT field numbers in tests/fit/test_rewriter_patching.py:

FILE_ID_MESG_NUM = 0
DEVICE_INFO_MESG_NUM = 23

# 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))

Write assertions that file_id and creator become (1, 3570) while sensor remains (32, 1234).

• Step 3: Run the test and verify failure

Run: pytest tests/fit/test_rewriter_patching.py -v

Expected: missing conversion functions.

• Step 4: Implement conservative creator detection

In _patch_device_metadata, process each device_info record as a unit:

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

Only after this check may fields 2, 3, 4, and 27 be patched. Do not patch a device_info message that lacks field 0.

Always patch eligible file_id fields 1, 2, optional 3, and 8.

• Step 5: Implement numeric/string read-write helpers and conversion entry point

Use the supplied working _read_field_value() and _write_field_value() behavior, then expose:

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)

• Step 6: Run patching tests

Run: pytest tests/fit/test_rewriter_patching.py -v

Expected: PASS.

• Step 7: Commit

git add app/fit tests/fit/test_rewriter_patching.py
git commit -m "feat: patch FIT creator as Edge 1030 Plus"

Task 4: Prove binary preservation and advanced record support

Files:

• Create: tests/fit/test_rewriter_preservation.py
• Modify: tests/fit/builders.py
• Modify: app/fit/rewriter.py only if a test exposes an actual parser defect

Interfaces:

• No new public interface; this task hardens convert_fit_device().

• Step 1: Write a preservation test that records changed byte positions

# tests/fit/test_rewriter_preservation.py
from pathlib import Path

from app.fit.rewriter import convert_fit_device


def test_only_target_fields_and_crcs_change(tmp_path: Path, complex_fit_bytes: bytes) -> None:
    source = tmp_path / "source.fit"
    output = tmp_path / "output.fit"
    source.write_bytes(complex_fit_bytes)

    convert_fit_device(source, output)

    before = source.read_bytes()
    after = output.read_bytes()
    assert len(before) == len(after)

    changed = {index for index, (a, b) in enumerate(zip(before, after)) if a != b}
    expected_metadata_offsets = set(find_expected_device_metadata_offsets(before))
    crc_offsets = {12, 13, len(before) - 2, len(before) - 1}
    assert changed <= expected_metadata_offsets | crc_offsets

The fixture helper find_expected_device_metadata_offsets() must use the test fixture's known construction offsets, not production parser code, so the test is independent.

• Step 2: Add fixtures for compressed timestamps, developer fields, and local-definition replacement

Construct one synthetic FIT data section containing:

1. a normal file_id definition/data pair;
2. a device_info definition with creator record;
3. a record definition with one developer field and one data record;
4. a compressed-timestamp data header referring to a known local definition;
5. a later replacement definition for the same local message number.

The test passes if conversion completes, output CRC validates, and non-target payload bytes remain identical.

• Step 3: Run the focused advanced tests

Run: pytest tests/fit/test_rewriter_preservation.py -v

Expected: PASS. If a parser guard fails, fix only the smallest parser defect required by the test.

• Step 4: Add rejection tests for malformed/truncated definitions

def test_truncated_definition_is_non_recoverable(tmp_path: Path) -> None:
    path = tmp_path / "truncated.fit"
    path.write_bytes(make_fit(bytes([0x40, 0x00, 0x00])))
    assert is_fit_file(path) is False

Also assert that convert_fit_device() raises FitFormatError for the same input.

• Step 5: Run the entire FIT suite

Run: pytest tests/fit -v

Expected: PASS.

• Step 6: Commit

git add tests/fit app/fit/rewriter.py
git commit -m "test: verify FIT binary preservation"