From 88869d1f52da08b593e46576ba02567a3f334bf3 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Mon, 29 Jun 2026 11:13:54 -0700 Subject: [PATCH 01/15] python(feat): add ULog data import support --- python/CHANGELOG.md | 20 ++ python/examples/data_import/ulog/.env.example | 5 + python/examples/data_import/ulog/main.py | 67 +++++ .../data_import/ulog/requirements.txt | 2 + .../low_level_wrappers/data_imports.py | 3 + python/lib/sift_client/_internal/util/ulog.py | 245 ++++++++++++++++++ .../sift_client/_tests/_internal/test_ulog.py | 229 ++++++++++++++++ .../_tests/resources/test_data_imports.py | 104 ++++++++ .../lib/sift_client/resources/data_imports.py | 29 ++- .../lib/sift_client/sift_types/data_import.py | 146 +++++++++++ python/pyproject.toml | 14 +- python/uv.lock | 29 ++- 12 files changed, 883 insertions(+), 10 deletions(-) create mode 100644 python/examples/data_import/ulog/.env.example create mode 100644 python/examples/data_import/ulog/main.py create mode 100644 python/examples/data_import/ulog/requirements.txt create mode 100644 python/lib/sift_client/_internal/util/ulog.py create mode 100644 python/lib/sift_client/_tests/_internal/test_ulog.py diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index ed70f3f66c..c275ffdde1 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -61,6 +61,26 @@ Templates can be fetched by ID or client key, and rules can be attached by rule Breaking change: `client.reports.create_from_template` now takes `report_template` (a `ReportTemplate` or ID string) and `run` (a `Run` or ID string) instead of `report_template_id` and `run_id`, matching the other `create_from_*` methods. +#### ULog imports + +Added PX4 ULog (`.ulg`) as a supported data import format. ULog files are self-describing, so `client.data_import.import_from_path("flight.ulg", asset=asset)` auto-detects every channel from the embedded schema with no column mapping. Detection runs client-side and requires the `ulog` extra (`pip install sift-stack-py[ulog]`). + +Call `detect_config` to inspect and patch the import before uploading, the same as the other formats: + +```python +config = client.data_import.detect_config("flight.ulg") + +# Import only the accelerometer channels +config.data = [d for d in config.data if d.channel.startswith("sensor_accel_0.")] + +# Anchor the timeline for a log with no GPS fix, and import run metadata +config.relative_start_time = datetime(2026, 1, 1, tzinfo=timezone.utc) +config.info_keys = ["ver_sw"] +config.param_keys = ["BAT1_CAPACITY"] + +client.data_import.import_from_path("flight.ulg", asset=asset, config=config) +``` + #### Resource and principal attributes (ABAC) Added a public API for attribute based access control (ABAC) attributes. `client.resource_attributes` manages attribute keys assigned to entities (assets, channels, runs), and `client.principal_attributes` manages attribute keys assigned to principals (users and user groups). Both are available synchronously and asynchronously via `client.async_`. diff --git a/python/examples/data_import/ulog/.env.example b/python/examples/data_import/ulog/.env.example new file mode 100644 index 0000000000..e4e7e11eb4 --- /dev/null +++ b/python/examples/data_import/ulog/.env.example @@ -0,0 +1,5 @@ +SIFT_GRPC_URI= +SIFT_REST_URI= +SIFT_API_KEY= +ASSET_NAME= +ULOG_PATH=sample_data.ulg diff --git a/python/examples/data_import/ulog/main.py b/python/examples/data_import/ulog/main.py new file mode 100644 index 0000000000..aa4ffeb2d9 --- /dev/null +++ b/python/examples/data_import/ulog/main.py @@ -0,0 +1,67 @@ +"""Import a PX4 ULog (.ulg) file into Sift. + +ULog files are self-describing, so detection enumerates every channel from the +embedded schema and you supply no column mapping. Channel names follow PX4's +convention of message, multi-instance index, and field, e.g. "sensor_accel_0.x". + +Point ULOG_PATH at your own .ulg flight log. +""" + +import os + +from dotenv import load_dotenv +from sift_client import SiftClient + +if __name__ == "__main__": + load_dotenv() + + grpc_uri = os.getenv("SIFT_GRPC_URI") + assert grpc_uri, "expected 'SIFT_GRPC_URI' environment variable to be set" + + rest_uri = os.getenv("SIFT_REST_URI") + assert rest_uri, "expected 'SIFT_REST_URI' environment variable to be set" + + apikey = os.getenv("SIFT_API_KEY") + assert apikey, "expected 'SIFT_API_KEY' environment variable to be set" + + asset_name = os.getenv("ASSET_NAME") + assert asset_name, "expected 'ASSET_NAME' environment variable to be set" + + ulog_path = os.getenv("ULOG_PATH", "sample_data.ulg") + + client = SiftClient(api_key=apikey, grpc_url=grpc_uri, rest_url=rest_uri) + + # Auto-detect the config and import the file. + import_job = client.data_import.import_from_path( + ulog_path, + asset=asset_name, + ) + + import_job.wait_until_complete() + + # If auto-detect doesn't quite match your file, inspect the config and patch + # it before importing. Common fixes: drop channels you don't need, rename or + # retype a channel, set a start time for logs without a GPS fix, or pick run + # metadata to import. + # + # from datetime import datetime, timezone + # + # config = client.data_import.detect_config(ulog_path) + # print(config) # inspect every detected channel + # + # # Example: import only the accelerometer channels + # config.data = [d for d in config.data if d.channel.startswith("sensor_accel_0.")] + # + # # Example: anchor the timeline when the log has no GPS fix + # config.relative_start_time = datetime(2026, 1, 1, tzinfo=timezone.utc) + # + # # Example: import firmware version and a parameter as run metadata + # config.info_keys = ["ver_sw"] + # config.param_keys = ["BAT1_CAPACITY"] + # + # import_job = client.data_import.import_from_path( + # ulog_path, + # asset=asset_name, + # config=config, + # ) + # import_job.wait_until_complete() diff --git a/python/examples/data_import/ulog/requirements.txt b/python/examples/data_import/ulog/requirements.txt new file mode 100644 index 0000000000..6952567f92 --- /dev/null +++ b/python/examples/data_import/ulog/requirements.txt @@ -0,0 +1,2 @@ +python-dotenv +sift-stack-py[ulog] diff --git a/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py b/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py index b0219124ba..cd430a773d 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/data_imports.py @@ -20,6 +20,7 @@ ParquetFlatDatasetImportConfig, ParquetSingleChannelPerRowImportConfig, TdmsImportConfig, + UlogImportConfig, ) from sift_client.transport import WithGrpcClient @@ -44,6 +45,8 @@ def _set_config_on_request( request.tdms_config.CopyFrom(config._to_proto()) elif isinstance(config, Hdf5ImportConfig): request.hdf5_config.CopyFrom(config._to_proto()) + elif isinstance(config, UlogImportConfig): + request.ulog_config.CopyFrom(config._to_proto()) else: raise TypeError(f"Unsupported import config type: {type(config).__name__}") diff --git a/python/lib/sift_client/_internal/util/ulog.py b/python/lib/sift_client/_internal/util/ulog.py new file mode 100644 index 0000000000..4c91ff9bac --- /dev/null +++ b/python/lib/sift_client/_internal/util/ulog.py @@ -0,0 +1,245 @@ +"""ULog (PX4 .ulg) channel detection. + +ULog files are self-describing: the embedded schema names every channel and its +C type, so detection enumerates channels with no user input. This mirrors the +server-side importer so a detected config imports identically. + +Detection never decodes data records. It walks the message stream reading only +the AddLogged ('A') subscriptions and logged-string ('L'/'C') markers, and +parses the Definitions section (via pyulog, header-only) for the message +formats. The channel naming, type mapping, and field flattening match the +server importer field for field. +""" + +from __future__ import annotations + +import contextlib +import struct +import sys +from pathlib import Path +from typing import TYPE_CHECKING + +from pyulog import ULog + +from sift_client.sift_types.channel import ChannelDataType +from sift_client.sift_types.data_import import UlogDataColumn, UlogImportConfig + +if TYPE_CHECKING: + from typing import BinaryIO + +# ULog C scalar types to Sift channel types. Narrow integers widen to 32-bit +# and char/char[N] become STRING, matching the server importer and the other +# import formats. +ULOG_TO_SIFT_TYPE: dict[str, ChannelDataType] = { + "int8_t": ChannelDataType.INT_32, + "int16_t": ChannelDataType.INT_32, + "int32_t": ChannelDataType.INT_32, + "int64_t": ChannelDataType.INT_64, + "uint8_t": ChannelDataType.UINT_32, + "uint16_t": ChannelDataType.UINT_32, + "uint32_t": ChannelDataType.UINT_32, + "uint64_t": ChannelDataType.UINT_64, + "float": ChannelDataType.FLOAT, + "double": ChannelDataType.DOUBLE, + "bool": ChannelDataType.BOOL, + "char": ChannelDataType.STRING, +} + +# Channel that collects printf-style logged strings ('L' and tagged 'C'). +LOG_MESSAGES_CHANNEL = "log_messages" + +ULOG_MAGIC = b"\x55\x4c\x6f\x67\x01\x12\x35" +ULOG_HEADER_SIZE = 16 +# Full sync message: 3-byte header (size=8, type 'S') + the 8 sync magic bytes. +SYNC_MESSAGE = struct.pack(" bool: + """True if any dotted segment of the key is a ULog padding field.""" + return any(seg.startswith("_padding") for seg in channel_key.split(".")) + + +def _find_next_sync(f: BinaryIO, start: int) -> int: + """Absolute offset of the next sync marker at or after ``start``, or -1.""" + chunk_size = 1 << 16 + overlap = len(SYNC_MESSAGE) - 1 + f.seek(start) + chunk_start = start + carry = b"" + while True: + chunk = f.read(chunk_size) + if not chunk: + return -1 + buf = carry + chunk + idx = buf.find(SYNC_MESSAGE) + if idx != -1: + return chunk_start - len(carry) + idx + chunk_start += len(chunk) + carry = buf[-overlap:] + + +class _ScanResult: + """Structural findings the header parse does not surface: which + ``(message_name, multi_id)`` pairs were logged and whether the file carries + logged-string channels.""" + + def __init__(self) -> None: + self.subscriptions: list[tuple[str, int]] = [] + self.has_untagged_logs = False + self.log_tags: set[int] = set() + + +def _scan_subscriptions(path: Path) -> _ScanResult: + """Walk the message stream for subscriptions and logged-string markers. + + Reads only 'A', 'L', and 'C' payloads and skips 'D' data records, so the + cost is one streaming pass with no data decode. Tolerates a truncated final + record and reframes past garbage at the next sync marker, so a partially + corrupt log still enumerates its channels. + """ + result = _ScanResult() + size = path.stat().st_size + if size < ULOG_HEADER_SIZE: + raise ValueError(f"'{path.name}' is not a ULog file (invalid size).") + + with open(path, "rb") as f: + if f.read(7) != ULOG_MAGIC[:7]: + raise ValueError(f"'{path.name}' is not a ULog file (bad magic bytes).") + + pos = ULOG_HEADER_SIZE + f.seek(pos) + while pos < size: + if pos + 3 > size: + break # truncated message header + msg_size, msg_type = struct.unpack(" size: + break # truncated final record + + consumed = 0 + if msg_type == ord("A"): + # 'A' (add subscription): multi_id[1], msg_id[2], message_name. + if msg_size >= 4: + payload = f.read(msg_size) + consumed = msg_size + multi_id = payload[0] + name = payload[3:msg_size].decode("utf-8", errors="replace") + result.subscriptions.append((name, multi_id)) + elif msg_type == ord("L"): + result.has_untagged_logs = True + elif msg_type == ord("C"): + # 'C' (tagged logged string): log_level[1], tag[2], ... + if msg_size >= 3: + (tag,) = struct.unpack_from(" list[tuple[str, str]]: + """Flatten one message format into ``(field_key, c_type)`` leaf entries. + + Scalars yield one entry; ``type[N]`` arrays yield one per element with the + index in brackets (``gyro_rad[0]``); ``char``/``char[N]`` collapse to a + single ``char`` entry; nested message types recurse to dotted leaf paths. + """ + flattened: list[tuple[str, str]] = [] + + def walk(prefix: str, type_name: str) -> None: + for field_type, array_size, field_name in message_formats[type_name].fields: + if field_type == "char": + # char and char[N] both collapse to a single STRING channel. + flattened.append((f"{prefix}{field_name}", "char")) + elif field_type in ULOG_TO_SIFT_TYPE: + if array_size > 0: + flattened.extend( + (f"{prefix}{field_name}[{i}]", field_type) for i in range(array_size) + ) + else: + flattened.append((f"{prefix}{field_name}", field_type)) + else: # nested message type + if array_size > 0: + for i in range(array_size): + walk(f"{prefix}{field_name}[{i}].", field_type) + else: + walk(f"{prefix}{field_name}.", field_type) + + walk("", message_name) + return flattened + + +def detect_ulog_fields(message_formats: dict, scan: _ScanResult) -> dict[str, str]: + """Enumerate importable channels as ``{channel_key: c_type}``. + + Each logged ``(message_name, multi_id)`` contributes its non-timestamp, + non-padding fields under the ``_.`` prefix. + Logged-string channels are appended as ``log_messages`` and + ``log_messages_``. + """ + fields: dict[str, str] = {} + for message_name, multi_id in scan.subscriptions: + # Truncated headers can leave a subscription without a format. + if message_name not in message_formats: + continue + # No top-level timestamp means no usable time axis. + if not any(f[2] == "timestamp" for f in message_formats[message_name].fields): + continue + prefix = f"{message_name}_{multi_id}" + for key, type_str in expand_message_fields(message_formats, message_name): + # timestamp is the time axis; _padding fields are alignment bytes. + if key == "timestamp" or _is_padding(key): + continue + fields[f"{prefix}.{key}"] = type_str + if scan.has_untagged_logs: + fields[LOG_MESSAGES_CHANNEL] = "char" + for tag in sorted(scan.log_tags): + fields[f"{LOG_MESSAGES_CHANNEL}_{tag}"] = "char" + return fields + + +def _parse_header(path: Path) -> ULog: + """Parse only the ULog Definitions section (no data records).""" + # pyulog logs format notes to stdout; keep the client's stdout clean. + with contextlib.redirect_stdout(sys.stderr): + return ULog(str(path), parse_header_only=True) + + +def detect_ulog_config(file_path: str | Path, asset_name: str = "") -> UlogImportConfig: + """Detect a ULog import config by enumerating the file's channels. + + Args: + file_path: Path to the ``.ulg`` file. + asset_name: The asset name to set on the config. + + Returns: + A UlogImportConfig whose ``data`` lists every detected channel with its + default name (the channel key) and mapped data type. Drop, rename, or + retype entries before importing; an empty ``data`` list imports all + channels with these same defaults server-side. + """ + path = Path(file_path) + scan = _scan_subscriptions(path) + header = _parse_header(path) + detected = detect_ulog_fields(header.message_formats, scan) + data = [ + UlogDataColumn(channel=key, name=key, data_type=ULOG_TO_SIFT_TYPE[c_type]) + for key, c_type in detected.items() + ] + return UlogImportConfig(asset_name=asset_name, data=data) diff --git a/python/lib/sift_client/_tests/_internal/test_ulog.py b/python/lib/sift_client/_tests/_internal/test_ulog.py new file mode 100644 index 0000000000..011b66f949 --- /dev/null +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -0,0 +1,229 @@ +"""Tests for ULog channel detection.""" + +from __future__ import annotations + +import struct + +import pytest + +from sift_client._internal.util.ulog import ( + ULOG_HEADER_SIZE, + ULOG_MAGIC, + ULOG_TO_SIFT_TYPE, + _scan_subscriptions, + _ScanResult, + detect_ulog_config, + detect_ulog_fields, + expand_message_fields, +) +from sift_client.sift_types.channel import ChannelDataType + + +class _FakeFormat: + """Stand-in for pyulog's MessageFormat: a list of (type, array_size, name).""" + + def __init__(self, fields): + self.fields = fields + + +def _header() -> bytes: + # 7 magic bytes + 1 version byte + uint64 start timestamp. + return ULOG_MAGIC[:7] + b"\x00" + struct.pack(" bytes: + return struct.pack(" bytes: + payload = bytes([multi_id]) + struct.pack(" bytes: + return _message("D", struct.pack(" bytes: + # 'L': log_level[1], timestamp[8], message. + return _message("L", b"\x06" + struct.pack(" bytes: + # 'C': log_level[1], tag[2], timestamp[8], message. + return _message("C", b"\x06" + struct.pack(" ImportConfig: """Auto-detect import configuration from a file. - Reads a sample of the file, sends it to the server's DetectConfig - endpoint, and returns the detected configuration. The file format - is inferred from the file extension when ``data_type`` is not - provided. + Returns the detected configuration, inferring the file format from the + extension when ``data_type`` is not provided. CSV and Parquet are + detected by sending a sample of the file to the server's DetectConfig + endpoint; TDMS, HDF5, and ULog are detected locally on the client. - CSV, Parquet, HDF5, and TDMS files are supported for + CSV, Parquet, HDF5, TDMS, and ULog files are supported for auto-detection. For CSV files, the server scans the first two rows for an optional @@ -267,7 +268,10 @@ async def detect_config( config = await self._detect_config_for_type(path, data_type_key) if time_format is not None: _apply_time_format(config, time_format) - elif not isinstance(config, TdmsImportConfig) and _get_time_format(config) is None: + elif ( + not isinstance(config, (TdmsImportConfig, UlogImportConfig)) + and _get_time_format(config) is None + ): _apply_time_format(config, TimeFormat.ABSOLUTE_UNIX_NANOSECONDS) return config @@ -298,6 +302,15 @@ async def _detect_config_for_type( "Install it via `pip install sift-stack-py[tdms]`." ) from e return await run_sync_function(lambda: detect_tdms_config(path)) + if data_type_key == DataTypeKey.ULOG: + try: + from sift_client._internal.util.ulog import detect_ulog_config + except ImportError as e: + raise RuntimeError( + "pyulog is required for ULog import. " + "Install it via `pip install sift-stack-py[ulog]`." + ) from e + return await run_sync_function(lambda: detect_ulog_config(path)) is_parquet = data_type_key in ( DataTypeKey.PARQUET_FLATDATASET, @@ -335,7 +348,7 @@ def _read_sample() -> bytes: raise ValueError( f"No supported configuration detected for '{path.name}'. " - "Only CSV, Parquet, HDF5, and TDMS are supported by auto-detection." + "Only CSV, Parquet, HDF5, TDMS, and ULog are supported by auto-detection." ) diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 88d0d06256..b160aa515a 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -14,6 +14,7 @@ DATA_TYPE_KEY_PARQUET_FLATDATASET, DATA_TYPE_KEY_PARQUET_SINGLE_CHANNEL_PER_ROW, DATA_TYPE_KEY_TDMS, + DATA_TYPE_KEY_ULOG, PARQUET_COMPLEX_TYPES_IMPORT_MODE_BOTH, PARQUET_COMPLEX_TYPES_IMPORT_MODE_BYTES, PARQUET_COMPLEX_TYPES_IMPORT_MODE_IGNORE, @@ -24,6 +25,8 @@ TDMS_FALLBACK_METHOD_FAIL_ON_ERROR, TDMS_FALLBACK_METHOD_IGNORE_ERROR, TDMS_FALLBACK_METHOD_UNSPECIFIED, + ULOG_PARSE_ERROR_POLICY_FAIL_ON_ERROR, + ULOG_PARSE_ERROR_POLICY_IGNORE_ERROR, ) from sift.data_imports.v2.data_imports_pb2 import CsvConfig as CsvConfigProto from sift.data_imports.v2.data_imports_pb2 import CsvTimeColumn as CsvTimeColumnProto @@ -47,6 +50,8 @@ from sift.data_imports.v2.data_imports_pb2 import TDMSConfig as TDMSConfigProto from sift.data_imports.v2.data_imports_pb2 import TdmsDataConfig as TdmsDataConfigProto from sift.data_imports.v2.data_imports_pb2 import TimeFormat as TimeFormatProto +from sift.data_imports.v2.data_imports_pb2 import UlogConfig as UlogConfigProto +from sift.data_imports.v2.data_imports_pb2 import UlogDataConfig as UlogDataConfigProto from sift_client._internal.util.timestamp import to_pb_timestamp from sift_client.sift_types.channel import ChannelDataType @@ -79,6 +84,7 @@ class DataTypeKey(Enum): HDF5_ONE_D = "hdf5_one_d" HDF5_TWO_D = "hdf5_two_d" HDF5_COMPOUND = "hdf5_compound" + ULOG = "ulog" DATA_TYPE_KEY_TO_PROTO = { @@ -89,12 +95,14 @@ class DataTypeKey(Enum): DataTypeKey.HDF5_ONE_D: DATA_TYPE_KEY_HDF5, DataTypeKey.HDF5_TWO_D: DATA_TYPE_KEY_HDF5, DataTypeKey.HDF5_COMPOUND: DATA_TYPE_KEY_HDF5, + DataTypeKey.ULOG: DATA_TYPE_KEY_ULOG, } EXTENSION_TO_DATA_TYPE_KEY: dict[str, DataTypeKey] = { ".csv": DataTypeKey.CSV, ".tdms": DataTypeKey.TDMS, + ".ulg": DataTypeKey.ULOG, } @@ -841,10 +849,148 @@ def _to_proto(self) -> Hdf5ConfigProto: return proto +class UlogParseErrorPolicy(Enum): + """How the importer handles recoverable ULog parse errors. + + Recoverable errors include a truncated final record, a data record + referencing an unbound message id, and garbage bytes skipped to the next + sync marker. The policy is enforced server-side at import time. + """ + + FAIL_ON_ERROR = ULOG_PARSE_ERROR_POLICY_FAIL_ON_ERROR + """Fail the import on any recoverable parse error.""" + + IGNORE_ERROR = ULOG_PARSE_ERROR_POLICY_IGNORE_ERROR + """Import the records that parsed; skipped records are logged.""" + + +class UlogDataColumn(DataColumnBase): + """A channel to import from a ULog file. + + Attributes: + channel: The full ULog channel name, formed from the message name, its + multi-instance index, and the field (e.g. ``"sensor_accel_0.x"``). + This selects the source field; the inherited ``name`` is the Sift + channel name it is imported as and defaults to ``channel``. + """ + + channel: str + + @model_validator(mode="before") + @classmethod + def _default_name_to_channel(cls, data: object) -> object: + # The Sift name defaults to the ULog channel key, matching the + # import-all default, so a selection can be built from `channel` alone. + if isinstance(data, dict) and not data.get("name") and data.get("channel"): + data["name"] = data["channel"] + return data + + +class UlogImportConfig(ImportConfigBase): + """Configuration for importing a PX4 ULog (``.ulg``) file. + + ULog files are self-describing, so ``detect_config`` enumerates every + channel from the embedded schema and you supply no column mapping. Inspect + the detected ``data`` and drop, rename, or retype channels before importing. + + Attributes: + data: Channels to import. Empty imports every detected channel with its + defaults; if non-empty, only the listed channels are imported. + relative_start_time: Log-start UTC. ULog timestamps are relative to a + boot clock; the timeline is anchored to absolute time by the log's + GPS fix when present, otherwise by this value. The import fails if + neither is available, so set this for logs without a GPS fix. + info_keys: Info (``I``/``M``) message keys to import as run metadata, + stored as ``info.``. Requires ``run_name`` or ``run_id``. + param_keys: Parameter (``P``) names to import as run metadata, stored + as ``param.``. Requires ``run_name`` or ``run_id``. + parse_error_policy: How to handle recoverable parse errors. Defaults to + failing the import. + """ + + data: list[UlogDataColumn] = [] + relative_start_time: datetime | None = None + info_keys: list[str] = [] + param_keys: list[str] = [] + parse_error_policy: UlogParseErrorPolicy = UlogParseErrorPolicy.FAIL_ON_ERROR + + def __getitem__(self, name: str) -> UlogDataColumn: + """Look up a data column by Sift channel name. + + Example:: + + config["sensor_accel_0.x"].data_type = ChannelDataType.DOUBLE + """ + for dc in self.data: + if dc.name == name: + return dc + raise KeyError(f"No data column named '{name}'") + + def _to_proto(self) -> UlogConfigProto: + proto = UlogConfigProto( + asset_name=self.asset_name, + run_name=self.run_name or "", + run_id=self.run_id or "", + info_keys=self.info_keys, + param_keys=self.param_keys, + parse_error_policy=self.parse_error_policy.value, + ) + if self.relative_start_time is not None: + proto.relative_start_time.CopyFrom(to_pb_timestamp(self.relative_start_time)) + for dc in self.data: + proto.data.append( + UlogDataConfigProto( + channel=dc.channel, + channel_config=ChannelConfigProto( + name=dc.name, + data_type=dc.data_type.value, + units=dc.units, + description=dc.description, + ), + ) + ) + return proto + + @classmethod + def _from_proto(cls, proto: UlogConfigProto) -> UlogImportConfig: + """Create from a proto UlogConfig (e.g. from a GetDataImport response).""" + relative_start_time = None + if proto.HasField("relative_start_time"): + from datetime import timezone + + relative_start_time = proto.relative_start_time.ToDatetime(tzinfo=timezone.utc) + + parse_error_policy = UlogParseErrorPolicy.FAIL_ON_ERROR + if proto.parse_error_policy == ULOG_PARSE_ERROR_POLICY_IGNORE_ERROR: + parse_error_policy = UlogParseErrorPolicy.IGNORE_ERROR + + data = [ + UlogDataColumn( + channel=d.channel, + name=d.channel_config.name, + data_type=ChannelDataType(d.channel_config.data_type), + units=d.channel_config.units, + description=d.channel_config.description, + ) + for d in proto.data + ] + return cls( + asset_name=proto.asset_name, + run_name=proto.run_name or None, + run_id=proto.run_id or None, + data=data, + relative_start_time=relative_start_time, + info_keys=list(proto.info_keys), + param_keys=list(proto.param_keys), + parse_error_policy=parse_error_policy, + ) + + ImportConfig = Union[ CsvImportConfig, ParquetFlatDatasetImportConfig, ParquetSingleChannelPerRowImportConfig, TdmsImportConfig, Hdf5ImportConfig, + UlogImportConfig, ] diff --git a/python/pyproject.toml b/python/pyproject.toml index dfe94c0430..ed45037b01 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -67,6 +67,7 @@ all = [ 'polars~=1.8', 'pyOpenSSL<24.0.0', 'pyarrow>=17.0.0', + 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", 'sift-stream-bindings==0.4.1', 'types-pyOpenSSL<24.0.0', @@ -109,6 +110,7 @@ dev-all = [ 'pytest-mock==3.14.0', 'pytest-randomly==3.15.0', 'pytest==8.2.2', + 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", 'ruff~=0.12.10', 'sift-stream-bindings==0.4.1', @@ -164,6 +166,7 @@ docs-build = [ 'pytest-mock==3.14.0', 'pytest-randomly==3.15.0', 'pytest==8.2.2', + 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", 'ruff~=0.12.10', 'sift-stream-bindings==0.4.1', @@ -174,6 +177,7 @@ file-imports = [ 'h5py~=3.11', 'npTDMS~=1.9', 'polars~=1.8', + 'pyulog~=1.2.2', "rosbags~=0.0 ; python_full_version >= '3.8.2'", ] hdf5 = [ @@ -197,6 +201,9 @@ sift-stream-bindings = [ tdms = [ 'npTDMS~=1.9', ] +ulog = [ + 'pyulog~=1.2.2', +] [tool.sift.extras] # Extras configurations defined here will be available in [project.optional-dependencies] @@ -235,6 +242,7 @@ tdms = ["npTDMS~=1.9"] rosbags = ["rosbags~=0.0 ; python_full_version >= '3.8.2'"] sift-stream = ["sift-stream-bindings==0.4.1"] hdf5 = ["h5py~=3.11", "polars~=1.8"] # polars is only used by sift_py; remove once sift_py is fully deprecated +ulog = ["pyulog~=1.2.2"] data-review = ["pyarrow>=17.0.0"] [tool.sift.extras.combine] @@ -245,7 +253,7 @@ dev = ["development"] sift-stream-bindings = ["sift-stream"] # combinations -file-imports = ["tdms", "rosbags", "hdf5"] +file-imports = ["tdms", "rosbags", "hdf5", "ulog"] all = ["openssl", "sift-stream", "file-imports", "data-review"] dev-all = ["development", "all", "build"] @@ -321,6 +329,10 @@ module = "grpc_testing" ignore_missing_imports = true ignore_errors = true +[[tool.mypy.overrides]] +module = "pyulog" +ignore_missing_imports = true + [[tool.mypy.overrides]] module = "grpc" ignore_missing_imports = true diff --git a/python/uv.lock b/python/uv.lock index 7a0c68645b..0fafe266dd 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -3702,6 +3702,21 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/dd/96da98f892250475bdf2328112d7468abdd4acc7b902b6af23f4ed958ea0/pytz-2026.2-py2.py3-none-any.whl", hash = "sha256:04156e608bee23d3792fd45c94ae47fae1036688e75032eea2e3bf0323d1f126", size = 510141, upload-time = "2026-05-04T01:35:27.408Z" }, ] +[[package]] +name = "pyulog" +version = "1.2.3" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "numpy", version = "1.24.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "numpy", version = "2.0.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.10.*'" }, + { name = "numpy", version = "2.4.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e0/9a/5fc70f3ac7033e07cb4fad927a9c18b6547bb310aa40f536ecd480a7cd46/pyulog-1.2.3.tar.gz", hash = "sha256:ea67e8284f33af9c6a2ce6072aad59c207323651243648661c68141e9a802734", size = 81146, upload-time = "2026-06-16T07:08:13.187Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/a6/c7/b69618b216523e3f3d13425a15df47fea16a968d01dbefe949f9fd365898/pyulog-1.2.3-py3-none-any.whl", hash = "sha256:bffbada69dc852a3ab796ec30a35bdb6594edc5f2ebebcb2831c3be9780ac64d", size = 52285, upload-time = "2026-06-16T07:08:11.856Z" }, +] + [[package]] name = "pyyaml" version = "6.0.3" @@ -4400,6 +4415,7 @@ all = [ { name = "pyarrow", version = "21.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, { name = "pyarrow", version = "24.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, { name = "pyopenssl" }, + { name = "pyulog" }, { name = "rosbags", version = "0.9.23", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.8.2' and python_full_version < '3.10'" }, { name = "rosbags", version = "0.11.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, { name = "sift-stream-bindings" }, @@ -4453,6 +4469,7 @@ dev-all = [ { name = "pytest-dotenv" }, { name = "pytest-mock" }, { name = "pytest-randomly" }, + { name = "pyulog" }, { name = "rosbags", version = "0.9.23", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.8.2' and python_full_version < '3.10'" }, { name = "rosbags", version = "0.11.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, { name = "ruff" }, @@ -4517,6 +4534,7 @@ docs-build = [ { name = "pytest-dotenv" }, { name = "pytest-mock" }, { name = "pytest-randomly" }, + { name = "pyulog" }, { name = "rosbags", version = "0.9.23", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.8.2' and python_full_version < '3.10'" }, { name = "rosbags", version = "0.11.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, { name = "ruff" }, @@ -4532,6 +4550,7 @@ file-imports = [ { name = "polars", version = "1.8.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, { name = "polars", version = "1.36.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, { name = "polars", version = "1.40.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "pyulog" }, { name = "rosbags", version = "0.9.23", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.8.2' and python_full_version < '3.10'" }, { name = "rosbags", version = "0.11.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, ] @@ -4561,6 +4580,9 @@ sift-stream-bindings = [ tdms = [ { name = "nptdms" }, ] +ulog = [ + { name = "pyulog" }, +] [package.metadata] requires-dist = [ @@ -4661,6 +4683,11 @@ requires-dist = [ { name = "pytest-randomly", marker = "extra == 'dev-all'", specifier = "==3.15.0" }, { name = "pytest-randomly", marker = "extra == 'development'", specifier = "==3.15.0" }, { name = "pytest-randomly", marker = "extra == 'docs-build'", specifier = "==3.15.0" }, + { name = "pyulog", marker = "extra == 'all'", specifier = "~=1.2.2" }, + { name = "pyulog", marker = "extra == 'dev-all'", specifier = "~=1.2.2" }, + { name = "pyulog", marker = "extra == 'docs-build'", specifier = "~=1.2.2" }, + { name = "pyulog", marker = "extra == 'file-imports'", specifier = "~=1.2.2" }, + { name = "pyulog", marker = "extra == 'ulog'", specifier = "~=1.2.2" }, { name = "pyyaml", specifier = "~=6.0" }, { name = "rapidyaml", specifier = "~=0.11" }, { name = "requests", specifier = "~=2.25" }, @@ -4693,7 +4720,7 @@ requires-dist = [ { name = "types-requests", specifier = "~=2.25" }, { name = "typing-extensions", specifier = "~=4.6" }, ] -provides-extras = ["all", "build", "data-review", "dev", "dev-all", "development", "docs", "docs-build", "file-imports", "hdf5", "openssl", "rosbags", "sift-stream", "sift-stream-bindings", "tdms"] +provides-extras = ["all", "build", "data-review", "dev", "dev-all", "development", "docs", "docs-build", "file-imports", "hdf5", "openssl", "rosbags", "sift-stream", "sift-stream-bindings", "tdms", "ulog"] [[package]] name = "sift-stream-bindings" From 7a65bf6348e05acad81231ab0b8111008063aeb5 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Mon, 29 Jun 2026 11:36:18 -0700 Subject: [PATCH 02/15] pyright --- python/lib/sift_client/sift_types/data_import.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index b160aa515a..a7a750c304 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -875,6 +875,9 @@ class UlogDataColumn(DataColumnBase): """ channel: str + # Declared optional so a column can be built from `channel` alone; the + # validator below fills it in. Inherited `name` is otherwise required. + name: str = "" @model_validator(mode="before") @classmethod From 684d363e874266ef7068a01751b88ccb7804f169 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Mon, 29 Jun 2026 11:42:22 -0700 Subject: [PATCH 03/15] clarify time-format docs --- python/lib/sift_client/_internal/util/ulog.py | 61 ++++++++----------- .../lib/sift_client/resources/data_imports.py | 27 ++++---- .../lib/sift_client/sift_types/data_import.py | 46 +++++++------- 3 files changed, 60 insertions(+), 74 deletions(-) diff --git a/python/lib/sift_client/_internal/util/ulog.py b/python/lib/sift_client/_internal/util/ulog.py index 4c91ff9bac..ccd6332d1e 100644 --- a/python/lib/sift_client/_internal/util/ulog.py +++ b/python/lib/sift_client/_internal/util/ulog.py @@ -1,14 +1,9 @@ -"""ULog (PX4 .ulg) channel detection. +"""Detect channels in PX4 ULog (``.ulg``) files. -ULog files are self-describing: the embedded schema names every channel and its -C type, so detection enumerates channels with no user input. This mirrors the -server-side importer so a detected config imports identically. - -Detection never decodes data records. It walks the message stream reading only -the AddLogged ('A') subscriptions and logged-string ('L'/'C') markers, and -parses the Definitions section (via pyulog, header-only) for the message -formats. The channel naming, type mapping, and field flattening match the -server importer field for field. +Detection reads the ULog definitions and lightweight message records to find +logged topics, multi-instance IDs, and logged-string channels. It does not +decode data records. Channel keys, type mapping, and flattened field names +match the server importer. """ from __future__ import annotations @@ -27,9 +22,8 @@ if TYPE_CHECKING: from typing import BinaryIO -# ULog C scalar types to Sift channel types. Narrow integers widen to 32-bit -# and char/char[N] become STRING, matching the server importer and the other -# import formats. +# Map ULog C scalars to Sift channel types. Smaller ints widen to 32-bit, and +# char fields import as strings. ULOG_TO_SIFT_TYPE: dict[str, ChannelDataType] = { "int8_t": ChannelDataType.INT_32, "int16_t": ChannelDataType.INT_32, @@ -45,7 +39,7 @@ "char": ChannelDataType.STRING, } -# Channel that collects printf-style logged strings ('L' and tagged 'C'). +# Base channel for ULog logged strings; tagged logs use ``log_messages_``. LOG_MESSAGES_CHANNEL = "log_messages" ULOG_MAGIC = b"\x55\x4c\x6f\x67\x01\x12\x35" @@ -58,12 +52,12 @@ def _is_padding(channel_key: str) -> bool: - """True if any dotted segment of the key is a ULog padding field.""" + """Return whether a channel key contains a ULog padding field.""" return any(seg.startswith("_padding") for seg in channel_key.split(".")) def _find_next_sync(f: BinaryIO, start: int) -> int: - """Absolute offset of the next sync marker at or after ``start``, or -1.""" + """Return the next sync marker offset at or after ``start``, or -1.""" chunk_size = 1 << 16 overlap = len(SYNC_MESSAGE) - 1 f.seek(start) @@ -82,9 +76,7 @@ def _find_next_sync(f: BinaryIO, start: int) -> int: class _ScanResult: - """Structural findings the header parse does not surface: which - ``(message_name, multi_id)`` pairs were logged and whether the file carries - logged-string channels.""" + """Logged topics and log-string markers found while scanning records.""" def __init__(self) -> None: self.subscriptions: list[tuple[str, int]] = [] @@ -93,12 +85,11 @@ def __init__(self) -> None: def _scan_subscriptions(path: Path) -> _ScanResult: - """Walk the message stream for subscriptions and logged-string markers. + """Collect logged topics and log-string markers without decoding data. - Reads only 'A', 'L', and 'C' payloads and skips 'D' data records, so the - cost is one streaming pass with no data decode. Tolerates a truncated final - record and reframes past garbage at the next sync marker, so a partially - corrupt log still enumerates its channels. + The scan skips data records and stops cleanly on a truncated tail. If + framing is lost, it resumes at the next sync marker so later channels can + still be detected. """ result = _ScanResult() size = path.stat().st_size @@ -154,11 +145,10 @@ def _scan_subscriptions(path: Path) -> _ScanResult: def expand_message_fields(message_formats: dict, message_name: str) -> list[tuple[str, str]]: - """Flatten one message format into ``(field_key, c_type)`` leaf entries. + """Flatten a message format into ``(field_key, c_type)`` leaf entries. - Scalars yield one entry; ``type[N]`` arrays yield one per element with the - index in brackets (``gyro_rad[0]``); ``char``/``char[N]`` collapse to a - single ``char`` entry; nested message types recurse to dotted leaf paths. + Arrays expand to ``field[i]``, nested messages use dotted paths, and + ``char[N]`` stays one string field. """ flattened: list[tuple[str, str]] = [] @@ -186,11 +176,10 @@ def walk(prefix: str, type_name: str) -> None: def detect_ulog_fields(message_formats: dict, scan: _ScanResult) -> dict[str, str]: - """Enumerate importable channels as ``{channel_key: c_type}``. + """Return importable channels as ``{channel_key: c_type}``. - Each logged ``(message_name, multi_id)`` contributes its non-timestamp, - non-padding fields under the ``_.`` prefix. - Logged-string channels are appended as ``log_messages`` and + Logged topics use ``_.``. The timestamp axis and + padding fields are excluded; logged strings become ``log_messages`` or ``log_messages_``. """ fields: dict[str, str] = {} @@ -229,10 +218,10 @@ def detect_ulog_config(file_path: str | Path, asset_name: str = "") -> UlogImpor asset_name: The asset name to set on the config. Returns: - A UlogImportConfig whose ``data`` lists every detected channel with its - default name (the channel key) and mapped data type. Drop, rename, or - retype entries before importing; an empty ``data`` list imports all - channels with these same defaults server-side. + A config whose ``data`` lists detected channels with default Sift names + and data types. Remove entries to skip channels, or edit entries before + importing. Leaving ``data`` empty lets the server import all channels + with the same defaults. """ path = Path(file_path) scan = _scan_subscriptions(path) diff --git a/python/lib/sift_client/resources/data_imports.py b/python/lib/sift_client/resources/data_imports.py index d84b81ef8c..f49c2b2782 100644 --- a/python/lib/sift_client/resources/data_imports.py +++ b/python/lib/sift_client/resources/data_imports.py @@ -112,12 +112,12 @@ async def import_from_path( multiple supported layouts (Parquet, HDF5) where the file extension alone is ambiguous. Only used when ``config`` is not provided. - time_format: Time format override. When provided, takes - precedence over the format returned by detection. When - omitted, the returned config uses the detected format if - available, falling back to - ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. Only used when - ``config`` is not provided. + time_format: Time format override for CSV, Parquet, HDF5, and TDMS. + Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the + detected format if available, otherwise + ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its + detected/default time handling. Only used when ``config`` is + not provided. run: ``Run`` object or run ID string to import into an existing run. Mutually exclusive with ``run_name``. run_name: Name for a new run. Defaults to the filename if @@ -245,11 +245,11 @@ async def detect_config( data_type: Explicit data type key. Required for formats with multiple supported layouts (Parquet, HDF5) where the file extension alone is ambiguous. - time_format: Time format override. When provided, takes - precedence over the format returned by detection. When - omitted, the returned config uses the detected format if - available, falling back to - ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. + time_format: Time format override for CSV, Parquet, HDF5, and TDMS. + Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the + detected format if available, otherwise + ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its + detected/default time handling. Returns: The detected import config. @@ -354,8 +354,9 @@ def _read_sample() -> bytes: def _apply_time_format(config: ImportConfig, time_format: TimeFormat) -> None: """Set the time format on a detected config, dispatching by config type. - CSV and Parquet store the format under ``time_column.format``; TDMS and - HDF5 store it on ``time_format`` directly. + + CSV and Parquet store the format under ``time_column.format``. TDMS and + HDF5 store it on ``time_format``. ULog has no configurable time format. """ if isinstance( config, diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index a7a750c304..6fc2f811ee 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -850,40 +850,36 @@ def _to_proto(self) -> Hdf5ConfigProto: class UlogParseErrorPolicy(Enum): - """How the importer handles recoverable ULog parse errors. + """Controls how ULog import handles records the parser can skip. - Recoverable errors include a truncated final record, a data record - referencing an unbound message id, and garbage bytes skipped to the next - sync marker. The policy is enforced server-side at import time. + Recoverable errors include a truncated final record or bytes skipped while + resynchronizing the log. The policy is enforced server-side at import time. """ FAIL_ON_ERROR = ULOG_PARSE_ERROR_POLICY_FAIL_ON_ERROR - """Fail the import on any recoverable parse error.""" + """Fail the import on the first recoverable parse error.""" IGNORE_ERROR = ULOG_PARSE_ERROR_POLICY_IGNORE_ERROR - """Import the records that parsed; skipped records are logged.""" + """Skip bad records and import the rest of the file.""" class UlogDataColumn(DataColumnBase): - """A channel to import from a ULog file. + """A single ULog channel selection. Attributes: - channel: The full ULog channel name, formed from the message name, its - multi-instance index, and the field (e.g. ``"sensor_accel_0.x"``). - This selects the source field; the inherited ``name`` is the Sift - channel name it is imported as and defaults to ``channel``. + channel: Source channel key, such as ``"sensor_accel_0.x"``. The key is + ``_.`` and is returned by + ``detect_config``. + name: Sift channel name to create. Defaults to ``channel``. """ channel: str - # Declared optional so a column can be built from `channel` alone; the - # validator below fills it in. Inherited `name` is otherwise required. + # Allow building a column from ``channel`` alone; the validator fills ``name``. name: str = "" @model_validator(mode="before") @classmethod def _default_name_to_channel(cls, data: object) -> object: - # The Sift name defaults to the ULog channel key, matching the - # import-all default, so a selection can be built from `channel` alone. if isinstance(data, dict) and not data.get("name") and data.get("channel"): data["name"] = data["channel"] return data @@ -892,17 +888,17 @@ def _default_name_to_channel(cls, data: object) -> object: class UlogImportConfig(ImportConfigBase): """Configuration for importing a PX4 ULog (``.ulg``) file. - ULog files are self-describing, so ``detect_config`` enumerates every - channel from the embedded schema and you supply no column mapping. Inspect - the detected ``data`` and drop, rename, or retype channels before importing. + ULog files describe their own channels. Leave ``data`` empty to import every + detected channel, or call ``detect_config`` and edit the returned ``data`` + list to skip, rename, retype, or annotate channels before importing. Attributes: - data: Channels to import. Empty imports every detected channel with its - defaults; if non-empty, only the listed channels are imported. - relative_start_time: Log-start UTC. ULog timestamps are relative to a - boot clock; the timeline is anchored to absolute time by the log's - GPS fix when present, otherwise by this value. The import fails if - neither is available, so set this for logs without a GPS fix. + data: Channel selections. Empty imports all detected channels with + default names and data types. If non-empty, only these channels are + imported. + relative_start_time: Absolute log start time to use when the log does + not contain a GPS-derived UTC reference. ULog timestamps are + relative to boot time. Set this for logs without that reference. info_keys: Info (``I``/``M``) message keys to import as run metadata, stored as ``info.``. Requires ``run_name`` or ``run_id``. param_keys: Parameter (``P``) names to import as run metadata, stored @@ -918,7 +914,7 @@ class UlogImportConfig(ImportConfigBase): parse_error_policy: UlogParseErrorPolicy = UlogParseErrorPolicy.FAIL_ON_ERROR def __getitem__(self, name: str) -> UlogDataColumn: - """Look up a data column by Sift channel name. + """Look up a configured ULog channel by Sift channel name. Example:: From d4ff99de6c98495347035c162c6b9f7dceea3efb Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Mon, 29 Jun 2026 11:52:36 -0700 Subject: [PATCH 04/15] stubs --- .../resources/sync_stubs/__init__.pyi | 34 +++++++++---------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index d977109be3..0c975255ff 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -696,12 +696,12 @@ class DataImportAPI: ) -> ImportConfig: """Auto-detect import configuration from a file. - Reads a sample of the file, sends it to the server's DetectConfig - endpoint, and returns the detected configuration. The file format - is inferred from the file extension when ``data_type`` is not - provided. + Returns the detected configuration, inferring the file format from the + extension when ``data_type`` is not provided. CSV and Parquet are + detected by sending a sample of the file to the server's DetectConfig + endpoint; TDMS, HDF5, and ULog are detected locally on the client. - CSV, Parquet, HDF5, and TDMS files are supported for + CSV, Parquet, HDF5, TDMS, and ULog files are supported for auto-detection. For CSV files, the server scans the first two rows for an optional @@ -731,11 +731,11 @@ class DataImportAPI: data_type: Explicit data type key. Required for formats with multiple supported layouts (Parquet, HDF5) where the file extension alone is ambiguous. - time_format: Time format override. When provided, takes - precedence over the format returned by detection. When - omitted, the returned config uses the detected format if - available, falling back to - ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. + time_format: Time format override for CSV, Parquet, HDF5, and TDMS. + Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the + detected format if available, otherwise + ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its + detected/default time handling. Returns: The detected import config. @@ -788,7 +788,7 @@ class DataImportAPI: completion before proceeding. When ``config`` is omitted the file format is auto-detected via - ``detect_config`` (CSV, Parquet, HDF5, and TDMS). + ``detect_config`` (CSV, Parquet, HDF5, TDMS, and ULog). When ``asset`` is provided it overrides the config value; otherwise the config's ``asset_name`` is used. If neither ``run`` nor ``run_name`` is provided (and none is @@ -833,12 +833,12 @@ class DataImportAPI: multiple supported layouts (Parquet, HDF5) where the file extension alone is ambiguous. Only used when ``config`` is not provided. - time_format: Time format override. When provided, takes - precedence over the format returned by detection. When - omitted, the returned config uses the detected format if - available, falling back to - ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. Only used when - ``config`` is not provided. + time_format: Time format override for CSV, Parquet, HDF5, and TDMS. + Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the + detected format if available, otherwise + ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its + detected/default time handling. Only used when ``config`` is + not provided. run: ``Run`` object or run ID string to import into an existing run. Mutually exclusive with ``run_name``. run_name: Name for a new run. Defaults to the filename if From 0130dc8e5bb76bd8d7cd677e7476645b0f96bc32 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Wed, 1 Jul 2026 18:22:55 -0700 Subject: [PATCH 05/15] update relative_start_time docs --- python/CHANGELOG.md | 3 ++- python/examples/data_import/ulog/main.py | 3 ++- python/lib/sift_client/sift_types/data_import.py | 7 ++++--- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index c275ffdde1..26906973c9 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -73,7 +73,8 @@ config = client.data_import.detect_config("flight.ulg") # Import only the accelerometer channels config.data = [d for d in config.data if d.channel.startswith("sensor_accel_0.")] -# Anchor the timeline for a log with no GPS fix, and import run metadata +# Anchor the timeline at an explicit log start time; this takes precedence +# over the log's GPS fix and is required for logs without one config.relative_start_time = datetime(2026, 1, 1, tzinfo=timezone.utc) config.info_keys = ["ver_sw"] config.param_keys = ["BAT1_CAPACITY"] diff --git a/python/examples/data_import/ulog/main.py b/python/examples/data_import/ulog/main.py index aa4ffeb2d9..7d0a1a657d 100644 --- a/python/examples/data_import/ulog/main.py +++ b/python/examples/data_import/ulog/main.py @@ -52,7 +52,8 @@ # # Example: import only the accelerometer channels # config.data = [d for d in config.data if d.channel.startswith("sensor_accel_0.")] # - # # Example: anchor the timeline when the log has no GPS fix + # # Example: anchor the timeline at an explicit log start time; this takes + # # precedence over the log's GPS fix and is required for logs without one # config.relative_start_time = datetime(2026, 1, 1, tzinfo=timezone.utc) # # # Example: import firmware version and a parameter as run metadata diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 6fc2f811ee..112d66e5dc 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -896,9 +896,10 @@ class UlogImportConfig(ImportConfigBase): data: Channel selections. Empty imports all detected channels with default names and data types. If non-empty, only these channels are imported. - relative_start_time: Absolute log start time to use when the log does - not contain a GPS-derived UTC reference. ULog timestamps are - relative to boot time. Set this for logs without that reference. + relative_start_time: Absolute UTC time of log start. ULog timestamps + are relative to boot time. When set, this anchors the timeline and + takes precedence over the log's GPS fix; otherwise the GPS fix + anchors the timeline. The import fails if neither is available. info_keys: Info (``I``/``M``) message keys to import as run metadata, stored as ``info.``. Requires ``run_name`` or ``run_id``. param_keys: Parameter (``P``) names to import as run metadata, stored From 7cad55f46c5ff6a4d993dec81054478f76a15d3c Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Thu, 2 Jul 2026 00:08:03 -0700 Subject: [PATCH 06/15] update example import --- python/examples/data_import/ulog/.env.example | 1 - python/examples/data_import/ulog/main.py | 10 ++++------ python/examples/data_import/ulog/sample_data.ulg | Bin 0 -> 2064 bytes python/lib/sift_client/sift_types/data_import.py | 1 - 4 files changed, 4 insertions(+), 8 deletions(-) create mode 100644 python/examples/data_import/ulog/sample_data.ulg diff --git a/python/examples/data_import/ulog/.env.example b/python/examples/data_import/ulog/.env.example index e4e7e11eb4..0c61389ff4 100644 --- a/python/examples/data_import/ulog/.env.example +++ b/python/examples/data_import/ulog/.env.example @@ -2,4 +2,3 @@ SIFT_GRPC_URI= SIFT_REST_URI= SIFT_API_KEY= ASSET_NAME= -ULOG_PATH=sample_data.ulg diff --git a/python/examples/data_import/ulog/main.py b/python/examples/data_import/ulog/main.py index 7d0a1a657d..2b6c9f6ece 100644 --- a/python/examples/data_import/ulog/main.py +++ b/python/examples/data_import/ulog/main.py @@ -4,7 +4,7 @@ embedded schema and you supply no column mapping. Channel names follow PX4's convention of message, multi-instance index, and field, e.g. "sensor_accel_0.x". -Point ULOG_PATH at your own .ulg flight log. +Swap sample_data.ulg for your own .ulg flight log. """ import os @@ -27,13 +27,11 @@ asset_name = os.getenv("ASSET_NAME") assert asset_name, "expected 'ASSET_NAME' environment variable to be set" - ulog_path = os.getenv("ULOG_PATH", "sample_data.ulg") - client = SiftClient(api_key=apikey, grpc_url=grpc_uri, rest_url=rest_uri) # Auto-detect the config and import the file. import_job = client.data_import.import_from_path( - ulog_path, + "sample_data.ulg", asset=asset_name, ) @@ -46,7 +44,7 @@ # # from datetime import datetime, timezone # - # config = client.data_import.detect_config(ulog_path) + # config = client.data_import.detect_config("sample_data.ulg") # print(config) # inspect every detected channel # # # Example: import only the accelerometer channels @@ -61,7 +59,7 @@ # config.param_keys = ["BAT1_CAPACITY"] # # import_job = client.data_import.import_from_path( - # ulog_path, + # "sample_data.ulg", # asset=asset_name, # config=config, # ) diff --git a/python/examples/data_import/ulog/sample_data.ulg b/python/examples/data_import/ulog/sample_data.ulg new file mode 100644 index 0000000000000000000000000000000000000000..7076558cf81a1cebece0af885e35df9670ff2109 GIT binary patch literal 2064 zcma*ne{2(F7zgmTAafDvA}A5)#%0Q6gN?$V1KRi2+e)|TAj51#0$Y1@*D}1d*xrBv zvVns92mvQQi$H+|1Pcm*4QlUfp%JMaW{OBquyeu)qM#s}@gv@zzNeD#hyA0s-RFIt zd;5Lgdn-y^4pskgs>V%QP|s&LDuC<%_!^JQ9?t1;xh<^C#_<#DYMg>@tVN)Nnp(~y zu(fmbcHYGb)Bxt_-rKGMrlC)g}?9?qthpO8C7ege-5|NiZAlD;f2 z=z9vs%OgCj!0~*Iz{Aodo#)~O*1=ikk+!xvx0`bc z7FIvY<>D!u%jx87g52|2RHC-cX5BN!RZ{afx5YDmbdEMhI~bKcr%befDIaYqWXhPr zlJe;*2OyM!7&+Si=GITXqf+k3aXcgc-wQoD*@4J}kg`NQOVbvM;ZuKT!T?mN7nZC0 z({y~IC5l5w2!!@*1E^oR6IV4vaN2wV)~xLS>Y^?@yeNQcO9-IDI{?zI_oul+KR%X2 zz?bI(D61GrXNz9km8l62MJ5$efa-Y3TwIidk{`Y`P+OebiqI|QX_%f-+6ASDJ81e! z$_l(nPkP2pdjNiKFQLy)slaw7+z1n$NK?r#@F~J+Z9YYO2BygAb?u8 z3TFyg;`EaQ;`JW^^wD)0%B4y%BaoU%QvvG>d61??PBS0UyuE}iH!3WF= zvHX3~lX~X}z{P1(#inasF`G|Kq^W>+>zj<-ewU;f>UR`|?`DpPc@DpLKZo?OZa4;T zdwaL&pA`_7{7lkBL*zx3Klu*_OTC*SV)}jpzFo%w{0(EIv`|#kxd<4C2LNVt7Dy*! zF>#QIK>W}r08cJ^MWPQz#Wf=dG>#JiT6>pD&3r_BaRdP>C;=F4CaGyeK^}wCVvCu! z*nN^1Dmw+k*A}a!(52;KN+;>#J0Ah)KUosBKg<-{GnDnO{0!jWm1^;)Wtn0&r>uYH zX@CjoD}&wDSAvr^6KI@s2B7$kWLPxH9^BYUfch;8a97uWlRM25)+t*lG(QHg+7`uC z(|E&!7E)n-*apyNZ4zy^UNdBDB701M_`V8A%p@mP35ne&Z!B8=C6ubyq|+bY$ijPV zq^GZMJAhU_o{nT!Vf8Sw#}q(lG2`8;ViIqxvFR%qetDvqP7+pNrh_CFfBqbRv$2%E zHeefW*-4;r=QoLp9WT>=uJ+?*hX&qQ<$fvatgdl79(`k}jBlaHy5C3-SrWi``J`LO ze*s`ipHcLC%@O=#76EJSMSyjVfwb#P6n8574+Xvh=)LhK-YuW8s!626XS@UuZu5kU>zyI e+<19zi(qcO3e<4YQtNNR2)liue9|T}s=onaH)D7J literal 0 HcmV?d00001 diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 112d66e5dc..8395610906 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -874,7 +874,6 @@ class UlogDataColumn(DataColumnBase): """ channel: str - # Allow building a column from ``channel`` alone; the validator fills ``name``. name: str = "" @model_validator(mode="before") From 3d8dc69d255daab5f8375617a5f80c2145fc3b9f Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Tue, 7 Jul 2026 11:03:04 -0700 Subject: [PATCH 07/15] use pyulog for channel detection --- python/CHANGELOG.md | 2 +- python/examples/data_import/ulog/main.py | 4 +- python/lib/sift_client/_internal/util/ulog.py | 168 +++------- .../sift_client/_tests/_internal/test_ulog.py | 302 ++++++++++++------ .../lib/sift_client/resources/data_imports.py | 16 +- .../resources/sync_stubs/__init__.pyi | 11 +- 6 files changed, 265 insertions(+), 238 deletions(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 26906973c9..7c900e7ef0 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -63,7 +63,7 @@ Breaking change: `client.reports.create_from_template` now takes `report_templat #### ULog imports -Added PX4 ULog (`.ulg`) as a supported data import format. ULog files are self-describing, so `client.data_import.import_from_path("flight.ulg", asset=asset)` auto-detects every channel from the embedded schema with no column mapping. Detection runs client-side and requires the `ulog` extra (`pip install sift-stack-py[ulog]`). +Added PX4 ULog (`.ulg`) as a supported data import format. ULog files are self-describing, so `client.data_import.import_from_path("flight.ulg", asset=asset)` imports every channel in the file with no column mapping. Detection runs client-side and requires the `ulog` extra (`pip install sift-stack-py[ulog]`). Call `detect_config` to inspect and patch the import before uploading, the same as the other formats: diff --git a/python/examples/data_import/ulog/main.py b/python/examples/data_import/ulog/main.py index 2b6c9f6ece..ee4a439170 100644 --- a/python/examples/data_import/ulog/main.py +++ b/python/examples/data_import/ulog/main.py @@ -1,7 +1,7 @@ """Import a PX4 ULog (.ulg) file into Sift. -ULog files are self-describing, so detection enumerates every channel from the -embedded schema and you supply no column mapping. Channel names follow PX4's +ULog files are self-describing, so the import needs no column mapping: every +channel in the file is imported. Channel names follow PX4's convention of message, multi-instance index, and field, e.g. "sensor_accel_0.x". Swap sample_data.ulg for your own .ulg flight log. diff --git a/python/lib/sift_client/_internal/util/ulog.py b/python/lib/sift_client/_internal/util/ulog.py index ccd6332d1e..a2dd76d258 100644 --- a/python/lib/sift_client/_internal/util/ulog.py +++ b/python/lib/sift_client/_internal/util/ulog.py @@ -1,27 +1,26 @@ """Detect channels in PX4 ULog (``.ulg``) files. -Detection reads the ULog definitions and lightweight message records to find -logged topics, multi-instance IDs, and logged-string channels. It does not -decode data records. Channel keys, type mapping, and flattened field names -match the server importer. +Detection parses the file with pyulog and enumerates the decoded topics, +multi-instance IDs, and logged-string channels. + +Warning: pyulog silently drops what it cannot decode, so malformed records +and topics that never logged a decodable record are missing from the detected +channels. An import config with an empty ``data`` list still imports every +channel in the file. """ from __future__ import annotations import contextlib -import struct import sys +import warnings from pathlib import Path -from typing import TYPE_CHECKING from pyulog import ULog from sift_client.sift_types.channel import ChannelDataType from sift_client.sift_types.data_import import UlogDataColumn, UlogImportConfig -if TYPE_CHECKING: - from typing import BinaryIO - # Map ULog C scalars to Sift channel types. Smaller ints widen to 32-bit, and # char fields import as strings. ULOG_TO_SIFT_TYPE: dict[str, ChannelDataType] = { @@ -42,108 +41,12 @@ # Base channel for ULog logged strings; tagged logs use ``log_messages_``. LOG_MESSAGES_CHANNEL = "log_messages" -ULOG_MAGIC = b"\x55\x4c\x6f\x67\x01\x12\x35" -ULOG_HEADER_SIZE = 16 -# Full sync message: 3-byte header (size=8, type 'S') + the 8 sync magic bytes. -SYNC_MESSAGE = struct.pack(" bool: """Return whether a channel key contains a ULog padding field.""" return any(seg.startswith("_padding") for seg in channel_key.split(".")) -def _find_next_sync(f: BinaryIO, start: int) -> int: - """Return the next sync marker offset at or after ``start``, or -1.""" - chunk_size = 1 << 16 - overlap = len(SYNC_MESSAGE) - 1 - f.seek(start) - chunk_start = start - carry = b"" - while True: - chunk = f.read(chunk_size) - if not chunk: - return -1 - buf = carry + chunk - idx = buf.find(SYNC_MESSAGE) - if idx != -1: - return chunk_start - len(carry) + idx - chunk_start += len(chunk) - carry = buf[-overlap:] - - -class _ScanResult: - """Logged topics and log-string markers found while scanning records.""" - - def __init__(self) -> None: - self.subscriptions: list[tuple[str, int]] = [] - self.has_untagged_logs = False - self.log_tags: set[int] = set() - - -def _scan_subscriptions(path: Path) -> _ScanResult: - """Collect logged topics and log-string markers without decoding data. - - The scan skips data records and stops cleanly on a truncated tail. If - framing is lost, it resumes at the next sync marker so later channels can - still be detected. - """ - result = _ScanResult() - size = path.stat().st_size - if size < ULOG_HEADER_SIZE: - raise ValueError(f"'{path.name}' is not a ULog file (invalid size).") - - with open(path, "rb") as f: - if f.read(7) != ULOG_MAGIC[:7]: - raise ValueError(f"'{path.name}' is not a ULog file (bad magic bytes).") - - pos = ULOG_HEADER_SIZE - f.seek(pos) - while pos < size: - if pos + 3 > size: - break # truncated message header - msg_size, msg_type = struct.unpack(" size: - break # truncated final record - - consumed = 0 - if msg_type == ord("A"): - # 'A' (add subscription): multi_id[1], msg_id[2], message_name. - if msg_size >= 4: - payload = f.read(msg_size) - consumed = msg_size - multi_id = payload[0] - name = payload[3:msg_size].decode("utf-8", errors="replace") - result.subscriptions.append((name, multi_id)) - elif msg_type == ord("L"): - result.has_untagged_logs = True - elif msg_type == ord("C"): - # 'C' (tagged logged string): log_level[1], tag[2], ... - if msg_size >= 3: - (tag,) = struct.unpack_from(" list[tuple[str, str]]: """Flatten a message format into ``(field_key, c_type)`` leaf entries. @@ -175,44 +78,50 @@ def walk(prefix: str, type_name: str) -> None: return flattened -def detect_ulog_fields(message_formats: dict, scan: _ScanResult) -> dict[str, str]: +def detect_ulog_fields(ulog: ULog) -> dict[str, str]: """Return importable channels as ``{channel_key: c_type}``. - Logged topics use ``_.``. The timestamp axis and - padding fields are excluded; logged strings become ``log_messages`` or - ``log_messages_``. + Decoded topics use ``_.``. The timestamp axis + and padding fields are excluded; logged strings become ``log_messages`` + or ``log_messages_``. """ fields: dict[str, str] = {} - for message_name, multi_id in scan.subscriptions: - # Truncated headers can leave a subscription without a format. - if message_name not in message_formats: - continue + for dataset in ulog.data_list: # No top-level timestamp means no usable time axis. - if not any(f[2] == "timestamp" for f in message_formats[message_name].fields): + if not any(f[2] == "timestamp" for f in ulog.message_formats[dataset.name].fields): continue - prefix = f"{message_name}_{multi_id}" - for key, type_str in expand_message_fields(message_formats, message_name): + prefix = f"{dataset.name}_{dataset.multi_id}" + for key, type_str in expand_message_fields(ulog.message_formats, dataset.name): # timestamp is the time axis; _padding fields are alignment bytes. if key == "timestamp" or _is_padding(key): continue fields[f"{prefix}.{key}"] = type_str - if scan.has_untagged_logs: + if ulog.logged_messages: fields[LOG_MESSAGES_CHANNEL] = "char" - for tag in sorted(scan.log_tags): + for tag in sorted(ulog.logged_messages_tagged): fields[f"{LOG_MESSAGES_CHANNEL}_{tag}"] = "char" return fields -def _parse_header(path: Path) -> ULog: - """Parse only the ULog Definitions section (no data records).""" - # pyulog logs format notes to stdout; keep the client's stdout clean. +def _parse_ulog(path: Path) -> ULog: + """Fully parse the file with pyulog, data records included.""" + # pyulog logs parse notes to stdout; keep the client's stdout clean. with contextlib.redirect_stdout(sys.stderr): - return ULog(str(path), parse_header_only=True) + try: + return ULog(str(path)) + except Exception as e: + raise ValueError(f"'{path.name}' could not be parsed as ULog: {e}") from e def detect_ulog_config(file_path: str | Path, asset_name: str = "") -> UlogImportConfig: """Detect a ULog import config by enumerating the file's channels. + Channels come from what pyulog decodes, so malformed records and topics + without decodable records are missing from the result; a warning is + emitted when pyulog reports corruption. On files with parse errors the + result can also differ from the channels the import finds; if the import + rejects the returned channel list, clear ``data`` to import all channels. + Args: file_path: Path to the ``.ulg`` file. asset_name: The asset name to set on the config. @@ -220,13 +129,18 @@ def detect_ulog_config(file_path: str | Path, asset_name: str = "") -> UlogImpor Returns: A config whose ``data`` lists detected channels with default Sift names and data types. Remove entries to skip channels, or edit entries before - importing. Leaving ``data`` empty lets the server import all channels - with the same defaults. + importing. Leaving ``data`` empty imports all channels with the same + defaults. """ path = Path(file_path) - scan = _scan_subscriptions(path) - header = _parse_header(path) - detected = detect_ulog_fields(header.message_formats, scan) + ulog = _parse_ulog(path) + if ulog.file_corruption: + warnings.warn( + f"'{path.name}' has records pyulog could not decode and dropped; " + "the detected channels may be incomplete.", + stacklevel=2, + ) + detected = detect_ulog_fields(ulog) data = [ UlogDataColumn(channel=key, name=key, data_type=ULOG_TO_SIFT_TYPE[c_type]) for key, c_type in detected.items() diff --git a/python/lib/sift_client/_tests/_internal/test_ulog.py b/python/lib/sift_client/_tests/_internal/test_ulog.py index 011b66f949..34c86edd9f 100644 --- a/python/lib/sift_client/_tests/_internal/test_ulog.py +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -7,11 +7,7 @@ import pytest from sift_client._internal.util.ulog import ( - ULOG_HEADER_SIZE, - ULOG_MAGIC, ULOG_TO_SIFT_TYPE, - _scan_subscriptions, - _ScanResult, detect_ulog_config, detect_ulog_fields, expand_message_fields, @@ -26,22 +22,57 @@ def __init__(self, fields): self.fields = fields +class _FakeDataset: + """Stand-in for pyulog's Data: a decoded topic instance.""" + + def __init__(self, name, multi_id): + self.name = name + self.multi_id = multi_id + + +class _FakeUlog: + """Stand-in for a parsed pyulog ULog: only the attributes detection reads.""" + + def __init__( + self, + message_formats=None, + data_list=None, + logged_messages=None, + logged_messages_tagged=None, + ): + self.message_formats = message_formats or {} + self.data_list = data_list or [] + self.logged_messages = logged_messages or [] + self.logged_messages_tagged = logged_messages_tagged or {} + + +# Builders for real ULog bytes, parsed by pyulog in the end-to-end tests. + + def _header() -> bytes: # 7 magic bytes + 1 version byte + uint64 start timestamp. - return ULOG_MAGIC[:7] + b"\x00" + struct.pack(" bytes: return struct.pack(" bytes: + # 'B': compat[8], incompat[8], appended offsets[24]; must follow the header. + return _message("B", b"\x00" * 40) + + +def _format(definition: str) -> bytes: + return _message("F", definition.encode()) + + def _add_logged(multi_id: int, msg_id: int, name: str) -> bytes: - payload = bytes([multi_id]) + struct.pack(" bytes: - return _message("D", struct.pack(" bytes: + return _message("D", struct.pack(" bytes: @@ -54,55 +85,8 @@ def _tagged_logged_string(tag: int, text: str = "tagged") -> bytes: return _message("C", b"\x06" + struct.pack(" Date: Tue, 7 Jul 2026 11:10:34 -0700 Subject: [PATCH 08/15] consolidate detection tests --- .../sift_client/_tests/_internal/test_ulog.py | 94 ++++++++----------- 1 file changed, 37 insertions(+), 57 deletions(-) diff --git a/python/lib/sift_client/_tests/_internal/test_ulog.py b/python/lib/sift_client/_tests/_internal/test_ulog.py index 34c86edd9f..5455505f1f 100644 --- a/python/lib/sift_client/_tests/_internal/test_ulog.py +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -7,7 +7,6 @@ import pytest from sift_client._internal.util.ulog import ( - ULOG_TO_SIFT_TYPE, detect_ulog_config, detect_ulog_fields, expand_message_fields, @@ -89,6 +88,17 @@ def _tagged_logged_string(tag: int, text: str = "tagged") -> bytes: ACCEL_RECORD = struct.pack(" bytes: + """A minimal well-formed log: one topic, one record.""" + return ( + _header() + + _flag_bits() + + ACCEL_FORMAT + + _add_logged(0, 0, "sensor_accel") + + _data_record(0, ACCEL_RECORD) + ) + + class TestExpandMessageFields: def test_scalars(self): formats = { @@ -151,15 +161,6 @@ def test_skips_timestamp_and_padding(self): ) assert detect_ulog_fields(ulog) == {"sensor_accel_0.x": "float"} - def test_multi_id_in_prefix(self): - ulog = _FakeUlog( - message_formats={ - "sensor_accel": _FakeFormat([("uint64_t", 0, "timestamp"), ("float", 0, "x")]) - }, - data_list=[_FakeDataset("sensor_accel", 0), _FakeDataset("sensor_accel", 1)], - ) - assert set(detect_ulog_fields(ulog)) == {"sensor_accel_0.x", "sensor_accel_1.x"} - def test_skips_message_without_timestamp(self): ulog = _FakeUlog( message_formats={"no_time": _FakeFormat([("float", 0, "x")])}, @@ -177,27 +178,10 @@ def test_log_message_channels_sorted_by_tag(self): ] -class TestTypeMapping: - def test_narrow_ints_widen(self): - assert ULOG_TO_SIFT_TYPE["int8_t"] == ChannelDataType.INT_32 - assert ULOG_TO_SIFT_TYPE["uint16_t"] == ChannelDataType.UINT_32 - assert ULOG_TO_SIFT_TYPE["int64_t"] == ChannelDataType.INT_64 - assert ULOG_TO_SIFT_TYPE["uint64_t"] == ChannelDataType.UINT_64 - - def test_char_is_string(self): - assert ULOG_TO_SIFT_TYPE["char"] == ChannelDataType.STRING - - class TestDetectUlogConfig: def test_detects_channel_per_field(self, tmp_path): path = tmp_path / "log.ulg" - path.write_bytes( - _header() - + _flag_bits() - + ACCEL_FORMAT - + _add_logged(0, 0, "sensor_accel") - + _data_record(0, ACCEL_RECORD) - ) + path.write_bytes(_accel_log()) config = detect_ulog_config(path, asset_name="drone") assert config.asset_name == "drone" @@ -241,31 +225,40 @@ def test_log_message_channels(self, tmp_path): ("log_messages_5", ChannelDataType.STRING), } - def test_bool_and_char_fields(self, tmp_path): + def test_maps_every_ulog_scalar_type(self, tmp_path): + # Narrow ints widen to 32-bit, char[N] imports as one string. path = tmp_path / "log.ulg" + fmt = _format( + "types:uint64_t timestamp;int8_t i8;int16_t i16;int32_t i32;int64_t i64;" + "uint8_t u8;uint16_t u16;uint32_t u32;uint64_t u64;" + "float f;double d;bool b;char[4] s;" + ) + record = struct.pack( + " Date: Thu, 9 Jul 2026 13:50:22 -0700 Subject: [PATCH 09/15] split ULog channel selector into message_name/instance/field_name --- python/CHANGELOG.md | 2 +- python/examples/data_import/ulog/main.py | 2 +- python/lib/sift_client/_internal/util/ulog.py | 49 ++++++++---- .../sift_client/_tests/_internal/test_ulog.py | 74 ++++++++++--------- .../_tests/resources/test_data_imports.py | 60 +++++++++++++-- .../lib/sift_client/sift_types/data_import.py | 42 ++++++++--- 6 files changed, 161 insertions(+), 68 deletions(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 7c900e7ef0..a993cc373e 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -71,7 +71,7 @@ Call `detect_config` to inspect and patch the import before uploading, the same config = client.data_import.detect_config("flight.ulg") # Import only the accelerometer channels -config.data = [d for d in config.data if d.channel.startswith("sensor_accel_0.")] +config.data = [d for d in config.data if d.message_name == "sensor_accel"] # Anchor the timeline at an explicit log start time; this takes precedence # over the log's GPS fix and is required for logs without one diff --git a/python/examples/data_import/ulog/main.py b/python/examples/data_import/ulog/main.py index ee4a439170..8243e32eac 100644 --- a/python/examples/data_import/ulog/main.py +++ b/python/examples/data_import/ulog/main.py @@ -48,7 +48,7 @@ # print(config) # inspect every detected channel # # # Example: import only the accelerometer channels - # config.data = [d for d in config.data if d.channel.startswith("sensor_accel_0.")] + # config.data = [d for d in config.data if d.message_name == "sensor_accel"] # # # Example: anchor the timeline at an explicit log start time; this takes # # precedence over the log's GPS fix and is required for logs without one diff --git a/python/lib/sift_client/_internal/util/ulog.py b/python/lib/sift_client/_internal/util/ulog.py index a2dd76d258..b8d9735961 100644 --- a/python/lib/sift_client/_internal/util/ulog.py +++ b/python/lib/sift_client/_internal/util/ulog.py @@ -15,6 +15,7 @@ import sys import warnings from pathlib import Path +from typing import NamedTuple from pyulog import ULog @@ -42,9 +43,20 @@ LOG_MESSAGES_CHANNEL = "log_messages" -def _is_padding(channel_key: str) -> bool: - """Return whether a channel key contains a ULog padding field.""" - return any(seg.startswith("_padding") for seg in channel_key.split(".")) +class DetectedUlogChannel(NamedTuple): + """The (message_name, instance, field_name) selector of a detected channel, + plus its ULog C type. Log-message channels select by channel name with an + empty field.""" + + message_name: str + instance: int + field_name: str + c_type: str + + +def _is_padding(field_name: str) -> bool: + """Return whether a field name contains a ULog padding segment.""" + return any(seg.startswith("_padding") for seg in field_name.split(".")) def expand_message_fields(message_formats: dict, message_name: str) -> list[tuple[str, str]]: @@ -78,29 +90,32 @@ def walk(prefix: str, type_name: str) -> None: return flattened -def detect_ulog_fields(ulog: ULog) -> dict[str, str]: - """Return importable channels as ``{channel_key: c_type}``. +def detect_ulog_fields(ulog: ULog) -> dict[str, DetectedUlogChannel]: + """Return importable channels as ``{channel_name: DetectedUlogChannel}``. Decoded topics use ``_.``. The timestamp axis and padding fields are excluded; logged strings become ``log_messages`` or ``log_messages_``. """ - fields: dict[str, str] = {} + channels: dict[str, DetectedUlogChannel] = {} for dataset in ulog.data_list: # No top-level timestamp means no usable time axis. if not any(f[2] == "timestamp" for f in ulog.message_formats[dataset.name].fields): continue prefix = f"{dataset.name}_{dataset.multi_id}" - for key, type_str in expand_message_fields(ulog.message_formats, dataset.name): + for field_name, type_str in expand_message_fields(ulog.message_formats, dataset.name): # timestamp is the time axis; _padding fields are alignment bytes. - if key == "timestamp" or _is_padding(key): + if field_name == "timestamp" or _is_padding(field_name): continue - fields[f"{prefix}.{key}"] = type_str + channels[f"{prefix}.{field_name}"] = DetectedUlogChannel( + dataset.name, dataset.multi_id, field_name, type_str + ) if ulog.logged_messages: - fields[LOG_MESSAGES_CHANNEL] = "char" + channels[LOG_MESSAGES_CHANNEL] = DetectedUlogChannel(LOG_MESSAGES_CHANNEL, 0, "", "char") for tag in sorted(ulog.logged_messages_tagged): - fields[f"{LOG_MESSAGES_CHANNEL}_{tag}"] = "char" - return fields + name = f"{LOG_MESSAGES_CHANNEL}_{tag}" + channels[name] = DetectedUlogChannel(name, 0, "", "char") + return channels def _parse_ulog(path: Path) -> ULog: @@ -142,7 +157,13 @@ def detect_ulog_config(file_path: str | Path, asset_name: str = "") -> UlogImpor ) detected = detect_ulog_fields(ulog) data = [ - UlogDataColumn(channel=key, name=key, data_type=ULOG_TO_SIFT_TYPE[c_type]) - for key, c_type in detected.items() + UlogDataColumn( + message_name=ch.message_name, + instance=ch.instance, + field_name=ch.field_name, + name=name, + data_type=ULOG_TO_SIFT_TYPE[ch.c_type], + ) + for name, ch in detected.items() ] return UlogImportConfig(asset_name=asset_name, data=data) diff --git a/python/lib/sift_client/_tests/_internal/test_ulog.py b/python/lib/sift_client/_tests/_internal/test_ulog.py index 5455505f1f..e50095ec2d 100644 --- a/python/lib/sift_client/_tests/_internal/test_ulog.py +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -7,6 +7,7 @@ import pytest from sift_client._internal.util.ulog import ( + DetectedUlogChannel, detect_ulog_config, detect_ulog_fields, expand_message_fields, @@ -159,7 +160,9 @@ def test_skips_timestamp_and_padding(self): }, data_list=[_FakeDataset("sensor_accel", 0)], ) - assert detect_ulog_fields(ulog) == {"sensor_accel_0.x": "float"} + assert detect_ulog_fields(ulog) == { + "sensor_accel_0.x": DetectedUlogChannel("sensor_accel", 0, "x", "float") + } def test_skips_message_without_timestamp(self): ulog = _FakeUlog( @@ -171,10 +174,10 @@ def test_skips_message_without_timestamp(self): def test_log_message_channels_sorted_by_tag(self): ulog = _FakeUlog(logged_messages=["boot"], logged_messages_tagged={2: [], 9: [], 5: []}) assert list(detect_ulog_fields(ulog).items()) == [ - ("log_messages", "char"), - ("log_messages_2", "char"), - ("log_messages_5", "char"), - ("log_messages_9", "char"), + ("log_messages", DetectedUlogChannel("log_messages", 0, "", "char")), + ("log_messages_2", DetectedUlogChannel("log_messages_2", 0, "", "char")), + ("log_messages_5", DetectedUlogChannel("log_messages_5", 0, "", "char")), + ("log_messages_9", DetectedUlogChannel("log_messages_9", 0, "", "char")), ] @@ -185,11 +188,13 @@ def test_detects_channel_per_field(self, tmp_path): config = detect_ulog_config(path, asset_name="drone") assert config.asset_name == "drone" - channels = {(d.channel, d.name, d.data_type) for d in config.data} + channels = { + (d.message_name, d.instance, d.field_name, d.name, d.data_type) for d in config.data + } assert channels == { - ("sensor_accel_0.x", "sensor_accel_0.x", ChannelDataType.FLOAT), - ("sensor_accel_0.y", "sensor_accel_0.y", ChannelDataType.FLOAT), - ("sensor_accel_0.z", "sensor_accel_0.z", ChannelDataType.FLOAT), + ("sensor_accel", 0, "x", "sensor_accel_0.x", ChannelDataType.FLOAT), + ("sensor_accel", 0, "y", "sensor_accel_0.y", ChannelDataType.FLOAT), + ("sensor_accel", 0, "z", "sensor_accel_0.z", ChannelDataType.FLOAT), } def test_multi_instance_topics_stay_separate(self, tmp_path): @@ -205,24 +210,27 @@ def test_multi_instance_topics_stay_separate(self, tmp_path): ) config = detect_ulog_config(path) - assert {d.channel for d in config.data} == { - "sensor_accel_0.x", - "sensor_accel_0.y", - "sensor_accel_0.z", - "sensor_accel_1.x", - "sensor_accel_1.y", - "sensor_accel_1.z", + assert all(d.message_name == "sensor_accel" for d in config.data) + assert {(d.instance, d.field_name) for d in config.data} == { + (0, "x"), + (0, "y"), + (0, "z"), + (1, "x"), + (1, "y"), + (1, "z"), } - def test_log_message_channels(self, tmp_path): + def test_log_message_channels_select_by_name_with_empty_field(self, tmp_path): path = tmp_path / "log.ulg" path.write_bytes(_header() + _flag_bits() + _logged_string() + _tagged_logged_string(5)) config = detect_ulog_config(path) - channels = {(d.channel, d.data_type) for d in config.data} + channels = { + (d.message_name, d.instance, d.field_name, d.name, d.data_type) for d in config.data + } assert channels == { - ("log_messages", ChannelDataType.STRING), - ("log_messages_5", ChannelDataType.STRING), + ("log_messages", 0, "", "log_messages", ChannelDataType.STRING), + ("log_messages_5", 0, "", "log_messages_5", ChannelDataType.STRING), } def test_maps_every_ulog_scalar_type(self, tmp_path): @@ -241,19 +249,19 @@ def test_maps_every_ulog_scalar_type(self, tmp_path): ) config = detect_ulog_config(path) - assert {(d.channel, d.data_type) for d in config.data} == { - ("types_0.i8", ChannelDataType.INT_32), - ("types_0.i16", ChannelDataType.INT_32), - ("types_0.i32", ChannelDataType.INT_32), - ("types_0.i64", ChannelDataType.INT_64), - ("types_0.u8", ChannelDataType.UINT_32), - ("types_0.u16", ChannelDataType.UINT_32), - ("types_0.u32", ChannelDataType.UINT_32), - ("types_0.u64", ChannelDataType.UINT_64), - ("types_0.f", ChannelDataType.FLOAT), - ("types_0.d", ChannelDataType.DOUBLE), - ("types_0.b", ChannelDataType.BOOL), - ("types_0.s", ChannelDataType.STRING), + assert {(d.field_name, d.data_type) for d in config.data} == { + ("i8", ChannelDataType.INT_32), + ("i16", ChannelDataType.INT_32), + ("i32", ChannelDataType.INT_32), + ("i64", ChannelDataType.INT_64), + ("u8", ChannelDataType.UINT_32), + ("u16", ChannelDataType.UINT_32), + ("u32", ChannelDataType.UINT_32), + ("u64", ChannelDataType.UINT_64), + ("f", ChannelDataType.FLOAT), + ("d", ChannelDataType.DOUBLE), + ("b", ChannelDataType.BOOL), + ("s", ChannelDataType.STRING), } def test_clean_file_does_not_warn(self, tmp_path, recwarn): diff --git a/python/lib/sift_client/_tests/resources/test_data_imports.py b/python/lib/sift_client/_tests/resources/test_data_imports.py index 02aa9e65a3..b01f64fc71 100644 --- a/python/lib/sift_client/_tests/resources/test_data_imports.py +++ b/python/lib/sift_client/_tests/resources/test_data_imports.py @@ -321,12 +321,14 @@ def _config(self): run_name="run1", data=[ UlogDataColumn( - channel="sensor_accel_0.x", + message_name="sensor_accel", + field_name="x", name="sensor_accel_0.x", data_type=ChannelDataType.FLOAT, ), UlogDataColumn( - channel="vehicle_status_0.nav_state", + message_name="vehicle_status", + field_name="nav_state", name="nav_state", data_type=ChannelDataType.UINT_32, units="enum", @@ -343,9 +345,12 @@ def test_to_proto(self): assert proto.asset_name == "my_asset" assert proto.run_name == "run1" assert len(proto.data) == 2 - assert proto.data[0].channel == "sensor_accel_0.x" + assert proto.data[0].message_name == "sensor_accel" + assert proto.data[0].instance == 0 + assert proto.data[0].field_name == "x" assert proto.data[0].channel_config.name == "sensor_accel_0.x" - assert proto.data[1].channel == "vehicle_status_0.nav_state" + assert proto.data[1].message_name == "vehicle_status" + assert proto.data[1].field_name == "nav_state" assert proto.data[1].channel_config.name == "nav_state" assert proto.data[1].channel_config.units == "enum" assert list(proto.info_keys) == ["ver_sw"] @@ -382,7 +387,9 @@ def test_from_proto_round_trip(self): assert restored.param_keys == config.param_keys assert restored.parse_error_policy == UlogParseErrorPolicy.IGNORE_ERROR assert len(restored.data) == 2 - assert restored.data[1].channel == "vehicle_status_0.nav_state" + assert restored.data[1].message_name == "vehicle_status" + assert restored.data[1].instance == 0 + assert restored.data[1].field_name == "nav_state" assert restored.data[1].name == "nav_state" assert restored.data[1].data_type == ChannelDataType.UINT_32 assert restored.data[1].units == "enum" @@ -391,13 +398,52 @@ def test_run_id_takes_precedence(self): proto = UlogImportConfig(asset_name="a", run_name="ignored", run_id="run_123")._to_proto() assert proto.run_id == "run_123" + def test_channel_derived_from_selector(self): + col = UlogDataColumn( + message_name="sensor_accel", + instance=1, + field_name="x", + data_type=ChannelDataType.FLOAT, + ) + assert col.channel == "sensor_accel_1.x" + + def test_nonzero_instance_round_trips_through_proto(self): + config = UlogImportConfig( + asset_name="a", + data=[ + UlogDataColumn( + message_name="sensor_accel", + instance=1, + field_name="x", + data_type=ChannelDataType.FLOAT, + ) + ], + ) + proto = config._to_proto() + assert proto.data[0].instance == 1 + restored = UlogImportConfig._from_proto(proto) + assert restored.data[0].instance == 1 + assert restored.data[0].name == "sensor_accel_1.x" + + def test_log_message_channel_is_message_name(self): + col = UlogDataColumn(message_name="log_messages_5", data_type=ChannelDataType.STRING) + assert col.channel == "log_messages_5" + assert col.name == "log_messages_5" + proto_data = UlogImportConfig(asset_name="a", data=[col])._to_proto().data[0] + assert proto_data.message_name == "log_messages_5" + assert proto_data.instance == 0 + assert proto_data.field_name == "" + def test_name_defaults_to_channel(self): - col = UlogDataColumn(channel="sensor_accel_0.x", data_type=ChannelDataType.FLOAT) + col = UlogDataColumn( + message_name="sensor_accel", field_name="x", data_type=ChannelDataType.FLOAT + ) assert col.name == "sensor_accel_0.x" def test_explicit_name_overrides_channel(self): col = UlogDataColumn( - channel="vehicle_status_0.nav_state", + message_name="vehicle_status", + field_name="nav_state", name="nav_state", data_type=ChannelDataType.UINT_32, ) diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 8395610906..53bb556945 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -866,22 +866,36 @@ class UlogParseErrorPolicy(Enum): class UlogDataColumn(DataColumnBase): """A single ULog channel selection. + Channels are selected by message name, instance, and field, as returned + by ``detect_config``. Log-message channels (``"log_messages"``, + ``"log_messages_"``) set ``message_name`` to the channel name and + leave ``field_name`` empty. + Attributes: - channel: Source channel key, such as ``"sensor_accel_0.x"``. The key is - ``_.`` and is returned by - ``detect_config``. + message_name: The ULog message name (e.g. ``"sensor_accel"``). + instance: The message instance for multi-instance topics. Defaults to 0. + field_name: The flattened ULog field name (e.g. ``"x"``, ``"esc[0].v"``). + Empty for log-message channels. name: Sift channel name to create. Defaults to ``channel``. """ - channel: str + message_name: str + instance: int = 0 + field_name: str = "" name: str = "" - @model_validator(mode="before") - @classmethod - def _default_name_to_channel(cls, data: object) -> object: - if isinstance(data, dict) and not data.get("name") and data.get("channel"): - data["name"] = data["channel"] - return data + @property + def channel(self) -> str: + """The full channel name this selection maps to, e.g. ``"sensor_accel_0.x"``.""" + if not self.field_name: + return self.message_name + return f"{self.message_name}_{self.instance}.{self.field_name}" + + @model_validator(mode="after") + def _default_name_to_channel(self) -> UlogDataColumn: + if not self.name: + self.name = self.channel + return self class UlogImportConfig(ImportConfigBase): @@ -939,7 +953,9 @@ def _to_proto(self) -> UlogConfigProto: for dc in self.data: proto.data.append( UlogDataConfigProto( - channel=dc.channel, + message_name=dc.message_name, + instance=dc.instance, + field_name=dc.field_name, channel_config=ChannelConfigProto( name=dc.name, data_type=dc.data_type.value, @@ -965,7 +981,9 @@ def _from_proto(cls, proto: UlogConfigProto) -> UlogImportConfig: data = [ UlogDataColumn( - channel=d.channel, + message_name=d.message_name, + instance=d.instance, + field_name=d.field_name, name=d.channel_config.name, data_type=ChannelDataType(d.channel_config.data_type), units=d.channel_config.units, From 4cc04ff3f8a01fbbb91542e67a5875c293d62a19 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Thu, 9 Jul 2026 14:37:01 -0700 Subject: [PATCH 10/15] update ULog comments and docstrings --- python/CHANGELOG.md | 2 ++ python/examples/data_import/ulog/main.py | 4 ++-- python/lib/sift_client/_internal/util/ulog.py | 10 ++++------ python/lib/sift_client/_tests/_internal/test_ulog.py | 4 ++-- python/lib/sift_client/resources/data_imports.py | 10 ++++++---- python/lib/sift_client/sift_types/data_import.py | 12 +++++++----- 6 files changed, 23 insertions(+), 19 deletions(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index a993cc373e..93dc78cdd8 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -76,6 +76,8 @@ config.data = [d for d in config.data if d.message_name == "sensor_accel"] # Anchor the timeline at an explicit log start time; this takes precedence # over the log's GPS fix and is required for logs without one config.relative_start_time = datetime(2026, 1, 1, tzinfo=timezone.utc) + +# Import firmware version and a parameter as run metadata config.info_keys = ["ver_sw"] config.param_keys = ["BAT1_CAPACITY"] diff --git a/python/examples/data_import/ulog/main.py b/python/examples/data_import/ulog/main.py index 8243e32eac..5a32af5fd0 100644 --- a/python/examples/data_import/ulog/main.py +++ b/python/examples/data_import/ulog/main.py @@ -1,8 +1,8 @@ """Import a PX4 ULog (.ulg) file into Sift. ULog files are self-describing, so the import needs no column mapping: every -channel in the file is imported. Channel names follow PX4's -convention of message, multi-instance index, and field, e.g. "sensor_accel_0.x". +channel in the file is imported. Channel names combine the ULog message, +multi-instance index, and field, e.g. "sensor_accel_0.x". Swap sample_data.ulg for your own .ulg flight log. """ diff --git a/python/lib/sift_client/_internal/util/ulog.py b/python/lib/sift_client/_internal/util/ulog.py index b8d9735961..d0aa954735 100644 --- a/python/lib/sift_client/_internal/util/ulog.py +++ b/python/lib/sift_client/_internal/util/ulog.py @@ -3,10 +3,8 @@ Detection parses the file with pyulog and enumerates the decoded topics, multi-instance IDs, and logged-string channels. -Warning: pyulog silently drops what it cannot decode, so malformed records -and topics that never logged a decodable record are missing from the detected -channels. An import config with an empty ``data`` list still imports every -channel in the file. +See ``detect_ulog_config`` for caveats on malformed files and empty +``data`` behavior. """ from __future__ import annotations @@ -60,7 +58,7 @@ def _is_padding(field_name: str) -> bool: def expand_message_fields(message_formats: dict, message_name: str) -> list[tuple[str, str]]: - """Flatten a message format into ``(field_key, c_type)`` leaf entries. + """Flatten a message format into ``(field_name, c_type)`` leaf entries. Arrays expand to ``field[i]``, nested messages use dotted paths, and ``char[N]`` stays one string field. @@ -119,7 +117,7 @@ def detect_ulog_fields(ulog: ULog) -> dict[str, DetectedUlogChannel]: def _parse_ulog(path: Path) -> ULog: - """Fully parse the file with pyulog, data records included.""" + """Fully parse the file with pyulog, raising ValueError if parsing fails.""" # pyulog logs parse notes to stdout; keep the client's stdout clean. with contextlib.redirect_stdout(sys.stderr): try: diff --git a/python/lib/sift_client/_tests/_internal/test_ulog.py b/python/lib/sift_client/_tests/_internal/test_ulog.py index e50095ec2d..4b971782ac 100644 --- a/python/lib/sift_client/_tests/_internal/test_ulog.py +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -16,7 +16,7 @@ class _FakeFormat: - """Stand-in for pyulog's MessageFormat: a list of (type, array_size, name).""" + """Stand-in for pyulog's MessageFormat: `fields` is a list of (type, array_size, name).""" def __init__(self, fields): self.fields = fields @@ -46,7 +46,7 @@ def __init__( self.logged_messages_tagged = logged_messages_tagged or {} -# Builders for real ULog bytes, parsed by pyulog in the end-to-end tests. +# Builders for real ULog bytes; the TestDetectUlogConfig cases parse these with pyulog. def _header() -> bytes: diff --git a/python/lib/sift_client/resources/data_imports.py b/python/lib/sift_client/resources/data_imports.py index 5e943127de..c2a19c0970 100644 --- a/python/lib/sift_client/resources/data_imports.py +++ b/python/lib/sift_client/resources/data_imports.py @@ -143,9 +143,10 @@ async def import_from_path( time_format=time_format, ) if isinstance(config, UlogImportConfig): - # ULog imports every channel without a channel list; a detected - # list would only filter that, and can be rejected on a damaged - # file where detection diverges. + # An empty channel list imports every channel. Keeping the + # detected list adds nothing and can fail the import when + # detection misreads a damaged file and lists channels the + # file does not contain. config.data = [] if asset is not None: @@ -245,7 +246,8 @@ async def detect_config( For ULog files, ``data`` lists the channels pyulog decodes from the file. When imported, a non-empty ``data`` list restricts the import - to those channels; clear it to import every channel in the file. + to exactly those channels; the import fails if a listed channel is + not in the file. Clear ``data`` to import every channel. For file types with multiple supported layouts (Parquet, HDF5), ``data_type`` must be specified explicitly. diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 53bb556945..cda440b77d 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -850,10 +850,11 @@ def _to_proto(self) -> Hdf5ConfigProto: class UlogParseErrorPolicy(Enum): - """Controls how ULog import handles records the parser can skip. + """Controls how ULog import handles recoverable parse errors. Recoverable errors include a truncated final record or bytes skipped while - resynchronizing the log. The policy is enforced server-side at import time. + resynchronizing the log. The policy applies when the file is imported, not + during ``detect_config``. """ FAIL_ON_ERROR = ULOG_PARSE_ERROR_POLICY_FAIL_ON_ERROR @@ -868,15 +869,16 @@ class UlogDataColumn(DataColumnBase): Channels are selected by message name, instance, and field, as returned by ``detect_config``. Log-message channels (``"log_messages"``, - ``"log_messages_"``) set ``message_name`` to the channel name and - leave ``field_name`` empty. + ``"log_messages_"``) set ``message_name`` to the channel name, with + ``instance`` 0 and an empty ``field_name``. Attributes: message_name: The ULog message name (e.g. ``"sensor_accel"``). instance: The message instance for multi-instance topics. Defaults to 0. field_name: The flattened ULog field name (e.g. ``"x"``, ``"esc[0].v"``). Empty for log-message channels. - name: Sift channel name to create. Defaults to ``channel``. + name: Sift channel name to create. Defaults to the full channel name + (the ``channel`` property), e.g. ``"sensor_accel_0.x"``. """ message_name: str From f004bbaba01b01dada4d153dcdacced969f424af Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Thu, 9 Jul 2026 14:52:08 -0700 Subject: [PATCH 11/15] stub --- python/lib/sift_client/resources/sync_stubs/__init__.pyi | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index b1f8f358c3..c799e8a85d 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -725,7 +725,8 @@ class DataImportAPI: For ULog files, ``data`` lists the channels pyulog decodes from the file. When imported, a non-empty ``data`` list restricts the import - to those channels; clear it to import every channel in the file. + to exactly those channels; the import fails if a listed channel is + not in the file. Clear ``data`` to import every channel. For file types with multiple supported layouts (Parquet, HDF5), ``data_type`` must be specified explicitly. From 8584318d1f28965501f4ff5d0487618a6b9840b5 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Fri, 10 Jul 2026 11:55:53 -0700 Subject: [PATCH 12/15] rename ULogDataColumn.channel to default_channel_name --- .../lib/sift_client/_tests/_internal/test_ulog.py | 2 +- .../_tests/resources/test_data_imports.py | 6 +++--- python/lib/sift_client/sift_types/data_import.py | 14 ++++++++------ 3 files changed, 12 insertions(+), 10 deletions(-) diff --git a/python/lib/sift_client/_tests/_internal/test_ulog.py b/python/lib/sift_client/_tests/_internal/test_ulog.py index 4b971782ac..61c141d19f 100644 --- a/python/lib/sift_client/_tests/_internal/test_ulog.py +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -302,7 +302,7 @@ def test_unknown_msg_id_warns_but_keeps_valid_channels(self, tmp_path): with pytest.warns(UserWarning, match="could not decode"): config = detect_ulog_config(path) - assert {d.channel for d in config.data} == { + assert {d.default_channel_name for d in config.data} == { "sensor_accel_0.x", "sensor_accel_0.y", "sensor_accel_0.z", diff --git a/python/lib/sift_client/_tests/resources/test_data_imports.py b/python/lib/sift_client/_tests/resources/test_data_imports.py index b01f64fc71..c960437a4c 100644 --- a/python/lib/sift_client/_tests/resources/test_data_imports.py +++ b/python/lib/sift_client/_tests/resources/test_data_imports.py @@ -405,7 +405,7 @@ def test_channel_derived_from_selector(self): field_name="x", data_type=ChannelDataType.FLOAT, ) - assert col.channel == "sensor_accel_1.x" + assert col.default_channel_name == "sensor_accel_1.x" def test_nonzero_instance_round_trips_through_proto(self): config = UlogImportConfig( @@ -427,7 +427,7 @@ def test_nonzero_instance_round_trips_through_proto(self): def test_log_message_channel_is_message_name(self): col = UlogDataColumn(message_name="log_messages_5", data_type=ChannelDataType.STRING) - assert col.channel == "log_messages_5" + assert col.default_channel_name == "log_messages_5" assert col.name == "log_messages_5" proto_data = UlogImportConfig(asset_name="a", data=[col])._to_proto().data[0] assert proto_data.message_name == "log_messages_5" @@ -451,7 +451,7 @@ def test_explicit_name_overrides_channel(self): def test_getitem(self): col = self._config()["nav_state"] - assert col.channel == "vehicle_status_0.nav_state" + assert col.default_channel_name == "vehicle_status_0.nav_state" def test_getitem_not_found(self): with pytest.raises(KeyError, match="nonexistent"): diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index cda440b77d..1508e4420c 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -877,8 +877,8 @@ class UlogDataColumn(DataColumnBase): instance: The message instance for multi-instance topics. Defaults to 0. field_name: The flattened ULog field name (e.g. ``"x"``, ``"esc[0].v"``). Empty for log-message channels. - name: Sift channel name to create. Defaults to the full channel name - (the ``channel`` property), e.g. ``"sensor_accel_0.x"``. + name: Sift channel name to create. Defaults to ``default_channel_name``, + e.g. ``"sensor_accel_0.x"``. """ message_name: str @@ -887,16 +887,18 @@ class UlogDataColumn(DataColumnBase): name: str = "" @property - def channel(self) -> str: - """The full channel name this selection maps to, e.g. ``"sensor_accel_0.x"``.""" + def default_channel_name(self) -> str: + """The default Sift channel name for this selection, + ``_.`` (e.g. ``"sensor_accel_0.x"``), + or just ``message_name`` for log-message channels.""" if not self.field_name: return self.message_name return f"{self.message_name}_{self.instance}.{self.field_name}" @model_validator(mode="after") - def _default_name_to_channel(self) -> UlogDataColumn: + def _apply_default_name(self) -> UlogDataColumn: if not self.name: - self.name = self.channel + self.name = self.default_channel_name return self From 2bc2ef51b87e3de21088ced4d47fe924a9ec1263 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Fri, 10 Jul 2026 11:57:33 -0700 Subject: [PATCH 13/15] formatting --- python/lib/sift_client/sift_types/data_import.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 1508e4420c..57dc15f900 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -890,7 +890,8 @@ class UlogDataColumn(DataColumnBase): def default_channel_name(self) -> str: """The default Sift channel name for this selection, ``_.`` (e.g. ``"sensor_accel_0.x"``), - or just ``message_name`` for log-message channels.""" + or just ``message_name`` for log-message channels. + """ if not self.field_name: return self.message_name return f"{self.message_name}_{self.instance}.{self.field_name}" From 210f9c6498ea81c504809bfa3d5bce7633cbeb6d Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Fri, 10 Jul 2026 11:58:38 -0700 Subject: [PATCH 14/15] update docstrings for empty data config --- python/lib/sift_client/sift_types/data_import.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 57dc15f900..870596b331 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -911,9 +911,9 @@ class UlogImportConfig(ImportConfigBase): list to skip, rename, retype, or annotate channels before importing. Attributes: - data: Channel selections. Empty imports all detected channels with - default names and data types. If non-empty, only these channels are - imported. + data: Channel selections. If empty, imports all detected channels with + default names and data types. If non-empty, imports only these + channels. relative_start_time: Absolute UTC time of log start. ULog timestamps are relative to boot time. When set, this anchors the timeline and takes precedence over the log's GPS fix; otherwise the GPS fix From d2cf0dc7d7c1ca6c26b8ad7163f210cc8646bd50 Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Fri, 10 Jul 2026 12:12:06 -0700 Subject: [PATCH 15/15] detect_ulog_fields return UlogDataColumn directly --- python/lib/sift_client/_internal/util/ulog.py | 57 +++++++------------ .../sift_client/_tests/_internal/test_ulog.py | 25 ++++---- 2 files changed, 37 insertions(+), 45 deletions(-) diff --git a/python/lib/sift_client/_internal/util/ulog.py b/python/lib/sift_client/_internal/util/ulog.py index d0aa954735..4c568f013d 100644 --- a/python/lib/sift_client/_internal/util/ulog.py +++ b/python/lib/sift_client/_internal/util/ulog.py @@ -13,7 +13,6 @@ import sys import warnings from pathlib import Path -from typing import NamedTuple from pyulog import ULog @@ -41,17 +40,6 @@ LOG_MESSAGES_CHANNEL = "log_messages" -class DetectedUlogChannel(NamedTuple): - """The (message_name, instance, field_name) selector of a detected channel, - plus its ULog C type. Log-message channels select by channel name with an - empty field.""" - - message_name: str - instance: int - field_name: str - c_type: str - - def _is_padding(field_name: str) -> bool: """Return whether a field name contains a ULog padding segment.""" return any(seg.startswith("_padding") for seg in field_name.split(".")) @@ -88,31 +76,41 @@ def walk(prefix: str, type_name: str) -> None: return flattened -def detect_ulog_fields(ulog: ULog) -> dict[str, DetectedUlogChannel]: - """Return importable channels as ``{channel_name: DetectedUlogChannel}``. +def detect_ulog_fields(ulog: ULog) -> list[UlogDataColumn]: + """Return importable channels as ``UlogDataColumn``s with default names + and data types. - Decoded topics use ``_.``. The timestamp axis + Decoded topics become ``_.``. The timestamp axis and padding fields are excluded; logged strings become ``log_messages`` or ``log_messages_``. """ - channels: dict[str, DetectedUlogChannel] = {} + channels: list[UlogDataColumn] = [] for dataset in ulog.data_list: # No top-level timestamp means no usable time axis. if not any(f[2] == "timestamp" for f in ulog.message_formats[dataset.name].fields): continue - prefix = f"{dataset.name}_{dataset.multi_id}" for field_name, type_str in expand_message_fields(ulog.message_formats, dataset.name): # timestamp is the time axis; _padding fields are alignment bytes. if field_name == "timestamp" or _is_padding(field_name): continue - channels[f"{prefix}.{field_name}"] = DetectedUlogChannel( - dataset.name, dataset.multi_id, field_name, type_str + channels.append( + UlogDataColumn( + message_name=dataset.name, + instance=dataset.multi_id, + field_name=field_name, + data_type=ULOG_TO_SIFT_TYPE[type_str], + ) ) if ulog.logged_messages: - channels[LOG_MESSAGES_CHANNEL] = DetectedUlogChannel(LOG_MESSAGES_CHANNEL, 0, "", "char") - for tag in sorted(ulog.logged_messages_tagged): - name = f"{LOG_MESSAGES_CHANNEL}_{tag}" - channels[name] = DetectedUlogChannel(name, 0, "", "char") + channels.append( + UlogDataColumn(message_name=LOG_MESSAGES_CHANNEL, data_type=ChannelDataType.STRING) + ) + channels.extend( + UlogDataColumn( + message_name=f"{LOG_MESSAGES_CHANNEL}_{tag}", data_type=ChannelDataType.STRING + ) + for tag in sorted(ulog.logged_messages_tagged) + ) return channels @@ -153,15 +151,4 @@ def detect_ulog_config(file_path: str | Path, asset_name: str = "") -> UlogImpor "the detected channels may be incomplete.", stacklevel=2, ) - detected = detect_ulog_fields(ulog) - data = [ - UlogDataColumn( - message_name=ch.message_name, - instance=ch.instance, - field_name=ch.field_name, - name=name, - data_type=ULOG_TO_SIFT_TYPE[ch.c_type], - ) - for name, ch in detected.items() - ] - return UlogImportConfig(asset_name=asset_name, data=data) + return UlogImportConfig(asset_name=asset_name, data=detect_ulog_fields(ulog)) diff --git a/python/lib/sift_client/_tests/_internal/test_ulog.py b/python/lib/sift_client/_tests/_internal/test_ulog.py index 61c141d19f..897b5d8aa7 100644 --- a/python/lib/sift_client/_tests/_internal/test_ulog.py +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -7,12 +7,12 @@ import pytest from sift_client._internal.util.ulog import ( - DetectedUlogChannel, detect_ulog_config, detect_ulog_fields, expand_message_fields, ) from sift_client.sift_types.channel import ChannelDataType +from sift_client.sift_types.data_import import UlogDataColumn class _FakeFormat: @@ -160,24 +160,29 @@ def test_skips_timestamp_and_padding(self): }, data_list=[_FakeDataset("sensor_accel", 0)], ) - assert detect_ulog_fields(ulog) == { - "sensor_accel_0.x": DetectedUlogChannel("sensor_accel", 0, "x", "float") - } + assert detect_ulog_fields(ulog) == [ + UlogDataColumn( + message_name="sensor_accel", + instance=0, + field_name="x", + data_type=ChannelDataType.FLOAT, + ) + ] def test_skips_message_without_timestamp(self): ulog = _FakeUlog( message_formats={"no_time": _FakeFormat([("float", 0, "x")])}, data_list=[_FakeDataset("no_time", 0)], ) - assert detect_ulog_fields(ulog) == {} + assert detect_ulog_fields(ulog) == [] def test_log_message_channels_sorted_by_tag(self): ulog = _FakeUlog(logged_messages=["boot"], logged_messages_tagged={2: [], 9: [], 5: []}) - assert list(detect_ulog_fields(ulog).items()) == [ - ("log_messages", DetectedUlogChannel("log_messages", 0, "", "char")), - ("log_messages_2", DetectedUlogChannel("log_messages_2", 0, "", "char")), - ("log_messages_5", DetectedUlogChannel("log_messages_5", 0, "", "char")), - ("log_messages_9", DetectedUlogChannel("log_messages_9", 0, "", "char")), + assert detect_ulog_fields(ulog) == [ + UlogDataColumn(message_name="log_messages", data_type=ChannelDataType.STRING), + UlogDataColumn(message_name="log_messages_2", data_type=ChannelDataType.STRING), + UlogDataColumn(message_name="log_messages_5", data_type=ChannelDataType.STRING), + UlogDataColumn(message_name="log_messages_9", data_type=ChannelDataType.STRING), ]