diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index ed70f3f66..93dc78cdd 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -61,6 +61,29 @@ 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)` 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: + +```python +config = client.data_import.detect_config("flight.ulg") + +# Import only the accelerometer channels +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"] + +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 000000000..0c61389ff --- /dev/null +++ b/python/examples/data_import/ulog/.env.example @@ -0,0 +1,4 @@ +SIFT_GRPC_URI= +SIFT_REST_URI= +SIFT_API_KEY= +ASSET_NAME= diff --git a/python/examples/data_import/ulog/main.py b/python/examples/data_import/ulog/main.py new file mode 100644 index 000000000..5a32af5fd --- /dev/null +++ b/python/examples/data_import/ulog/main.py @@ -0,0 +1,66 @@ +"""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 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. +""" + +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" + + 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( + "sample_data.ulg", + 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("sample_data.ulg") + # print(config) # inspect every detected channel + # + # # Example: import only the accelerometer channels + # 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 + # 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( + # "sample_data.ulg", + # 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 000000000..6952567f9 --- /dev/null +++ b/python/examples/data_import/ulog/requirements.txt @@ -0,0 +1,2 @@ +python-dotenv +sift-stack-py[ulog] 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 000000000..7076558cf Binary files /dev/null and b/python/examples/data_import/ulog/sample_data.ulg differ 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 b0219124b..cd430a773 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 000000000..4c568f013 --- /dev/null +++ b/python/lib/sift_client/_internal/util/ulog.py @@ -0,0 +1,154 @@ +"""Detect channels in PX4 ULog (``.ulg``) files. + +Detection parses the file with pyulog and enumerates the decoded topics, +multi-instance IDs, and logged-string channels. + +See ``detect_ulog_config`` for caveats on malformed files and empty +``data`` behavior. +""" + +from __future__ import annotations + +import contextlib +import sys +import warnings +from pathlib import Path + +from pyulog import ULog + +from sift_client.sift_types.channel import ChannelDataType +from sift_client.sift_types.data_import import UlogDataColumn, UlogImportConfig + +# 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, + "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, +} + +# Base channel for ULog logged strings; tagged logs use ``log_messages_``. +LOG_MESSAGES_CHANNEL = "log_messages" + + +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]]: + """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. + """ + 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(ulog: ULog) -> list[UlogDataColumn]: + """Return importable channels as ``UlogDataColumn``s with default names + and data types. + + Decoded topics become ``_.``. The timestamp axis + and padding fields are excluded; logged strings become ``log_messages`` + or ``log_messages_``. + """ + 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 + 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.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.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 + + +def _parse_ulog(path: Path) -> ULog: + """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: + 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. + + 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 imports all channels with the same + defaults. + """ + path = Path(file_path) + 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, + ) + 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 new file mode 100644 index 000000000..897b5d8aa --- /dev/null +++ b/python/lib/sift_client/_tests/_internal/test_ulog.py @@ -0,0 +1,320 @@ +"""Tests for ULog channel detection.""" + +from __future__ import annotations + +import struct + +import pytest + +from sift_client._internal.util.ulog import ( + 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: + """Stand-in for pyulog's MessageFormat: `fields` is a list of (type, array_size, name).""" + + 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; the TestDetectUlogConfig cases parse these with pyulog. + + +def _header() -> bytes: + # 7 magic bytes + 1 version byte + uint64 start timestamp. + return b"\x55\x4c\x6f\x67\x01\x12\x35" + b"\x01" + 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: + return _message("A", 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(" 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 = { + "sensor_accel": _FakeFormat( + [("uint64_t", 0, "timestamp"), ("float", 0, "x"), ("float", 0, "y")] + ) + } + assert expand_message_fields(formats, "sensor_accel") == [ + ("timestamp", "uint64_t"), + ("x", "float"), + ("y", "float"), + ] + + def test_array_expands_per_element(self): + formats = {"gyro": _FakeFormat([("float", 3, "gyro_rad")])} + assert expand_message_fields(formats, "gyro") == [ + ("gyro_rad[0]", "float"), + ("gyro_rad[1]", "float"), + ("gyro_rad[2]", "float"), + ] + + def test_char_scalar_and_array_collapse(self): + formats = {"m": _FakeFormat([("char", 0, "a"), ("char", 16, "b")])} + assert expand_message_fields(formats, "m") == [("a", "char"), ("b", "char")] + + def test_nested_message_recurses(self): + formats = { + "outer": _FakeFormat([("inner", 0, "current")]), + "inner": _FakeFormat([("double", 0, "lat"), ("double", 0, "lon")]), + } + assert expand_message_fields(formats, "outer") == [ + ("current.lat", "double"), + ("current.lon", "double"), + ] + + def test_nested_message_array(self): + formats = { + "report": _FakeFormat([("esc", 2, "esc")]), + "esc": _FakeFormat([("int32_t", 0, "rpm")]), + } + assert expand_message_fields(formats, "report") == [ + ("esc[0].rpm", "int32_t"), + ("esc[1].rpm", "int32_t"), + ] + + +class TestDetectUlogFields: + def test_skips_timestamp_and_padding(self): + ulog = _FakeUlog( + message_formats={ + "sensor_accel": _FakeFormat( + [ + ("uint64_t", 0, "timestamp"), + ("float", 0, "x"), + ("uint8_t", 0, "_padding0"), + ] + ) + }, + data_list=[_FakeDataset("sensor_accel", 0)], + ) + 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) == [] + + def test_log_message_channels_sorted_by_tag(self): + ulog = _FakeUlog(logged_messages=["boot"], logged_messages_tagged={2: [], 9: [], 5: []}) + 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), + ] + + +class TestDetectUlogConfig: + def test_detects_channel_per_field(self, tmp_path): + path = tmp_path / "log.ulg" + path.write_bytes(_accel_log()) + + config = detect_ulog_config(path, asset_name="drone") + assert config.asset_name == "drone" + 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), + } + + def test_multi_instance_topics_stay_separate(self, tmp_path): + path = tmp_path / "log.ulg" + path.write_bytes( + _header() + + _flag_bits() + + ACCEL_FORMAT + + _add_logged(0, 0, "sensor_accel") + + _add_logged(1, 1, "sensor_accel") + + _data_record(0, ACCEL_RECORD) + + _data_record(1, ACCEL_RECORD) + ) + + config = detect_ulog_config(path) + 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_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.message_name, d.instance, d.field_name, d.name, d.data_type) for d in config.data + } + assert channels == { + ("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): + # 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( + " 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 @@ -236,6 +244,11 @@ async def detect_config( in the metadata row; they are applied server-side during import but are not included in the returned config. + For ULog files, ``data`` lists the channels pyulog decodes from the + file. When imported, a non-empty ``data`` list restricts the import + 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. @@ -244,11 +257,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. @@ -267,7 +280,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 +314,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,14 +360,15 @@ 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." ) 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/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index d977109be..c799e8a85 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 @@ -723,6 +723,11 @@ class DataImportAPI: in the metadata row; they are applied server-side during import but are not included in the returned config. + For ULog files, ``data`` lists the channels pyulog decodes from the + file. When imported, a non-empty ``data`` list restricts the import + 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. @@ -731,11 +736,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 +793,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 @@ -826,19 +831,20 @@ class DataImportAPI: when ``config`` already has ``asset_name`` set. config: Import configuration describing the file format and column mapping. When provided, ``data_type`` is ignored. If omitted, - the config is auto-detected via ``detect_config``. You can - call ``detect_config`` yourself to inspect and modify the - config before passing it here. + the config is auto-detected via ``detect_config`` (for ULog + the detected channel list is dropped so every channel in the + file is imported). You can call ``detect_config`` yourself to + inspect and modify the config before passing it here. data_type: Explicit data type key. Required for formats with 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 diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 88d0d0625..870596b33 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,170 @@ def _to_proto(self) -> Hdf5ConfigProto: return proto +class UlogParseErrorPolicy(Enum): + """Controls how ULog import handles recoverable parse errors. + + Recoverable errors include a truncated final record or bytes skipped while + 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 + """Fail the import on the first recoverable parse error.""" + + IGNORE_ERROR = ULOG_PARSE_ERROR_POLICY_IGNORE_ERROR + """Skip bad records and import the rest of the file.""" + + +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, 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 ``default_channel_name``, + e.g. ``"sensor_accel_0.x"``. + """ + + message_name: str + instance: int = 0 + field_name: str = "" + name: str = "" + + @property + 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 _apply_default_name(self) -> UlogDataColumn: + if not self.name: + self.name = self.default_channel_name + return self + + +class UlogImportConfig(ImportConfigBase): + """Configuration for importing a PX4 ULog (``.ulg``) file. + + 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: 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 + 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 + 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 configured ULog channel 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( + 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, + 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( + 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, + 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 dfe94c043..ed45037b0 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 7a0c68645..0fafe266d 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"