144 lines
5.2 KiB
Python
144 lines
5.2 KiB
Python
import struct
|
|
from pathlib import Path
|
|
|
|
from app.fit.crc import fit_crc
|
|
from app.fit.models import FieldDefinition, LocalDefinition
|
|
|
|
|
|
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")
|
|
|
|
|
|
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
|
|
|
|
|
|
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
|