578 lines
18 KiB
Markdown
578 lines
18 KiB
Markdown
# 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
|
|
|
|
```text
|
|
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**
|
|
|
|
```python
|
|
# 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**
|
|
|
|
```python
|
|
# 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**
|
|
|
|
```bash
|
|
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**
|
|
|
|
```python
|
|
# 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**
|
|
|
|
```python
|
|
# 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**
|
|
|
|
```python
|
|
# 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:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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**
|
|
|
|
```python
|
|
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**
|
|
|
|
```bash
|
|
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**
|
|
|
|
```python
|
|
# 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`:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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:
|
|
|
|
```python
|
|
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**
|
|
|
|
```bash
|
|
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**
|
|
|
|
```python
|
|
# 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**
|
|
|
|
```python
|
|
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**
|
|
|
|
```bash
|
|
git add tests/fit app/fit/rewriter.py
|
|
git commit -m "test: verify FIT binary preservation"
|
|
```
|