From 3534540dd115c877be204d160e371a6c6a835e6c Mon Sep 17 00:00:00 2001 From: Wei Qi Lu Date: Sun, 23 Aug 2026 14:48:47 -0700 Subject: [PATCH 1/4] python(feat): add MCAP imports --- python/CHANGELOG.md | 12 + python/docs/guides/pytest_plugin/index.md | 10 +- python/examples/data_import/mcap/main.py | 71 +++ .../data_import/mcap/requirements.txt | 2 + .../data_import/mcap/sample_data.mcap | Bin 0 -> 6621 bytes .../low_level_wrappers/data_imports.py | 3 + python/lib/sift_client/_internal/util/mcap.py | 430 ++++++++++++++++++ .../low_level_wrappers/test_data_imports.py | 34 ++ .../sift_client/_tests/_internal/test_mcap.py | 376 +++++++++++++++ .../_tests/resources/test_data_imports.py | 180 ++++++++ .../lib/sift_client/resources/data_imports.py | 50 +- .../resources/sync_stubs/__init__.pyi | 31 +- .../lib/sift_client/sift_types/data_import.py | 184 ++++++++ python/pyproject.toml | 22 +- python/uv.lock | 55 ++- 15 files changed, 1428 insertions(+), 32 deletions(-) create mode 100644 python/examples/data_import/mcap/main.py create mode 100644 python/examples/data_import/mcap/requirements.txt create mode 100644 python/examples/data_import/mcap/sample_data.mcap create mode 100644 python/lib/sift_client/_internal/util/mcap.py create mode 100644 python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data_imports.py create mode 100644 python/lib/sift_client/_tests/_internal/test_mcap.py diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index b568ef513..5d63a1247 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -5,6 +5,18 @@ This project adheres to [Semantic Versioning](http://semver.org/). ## [Unreleased] +### What's New + +#### MCAP imports + +`client.data_import` now supports MCAP (`.mcap`) files with ROS 2 (`ros2msg`/`cdr`) topics. + +```python +job = client.data_import.import_from_path("recording.mcap", asset=my_asset) +``` + +Importing without a config ingests every supported channel; a file containing unsupported topics fails under the default parse error policy unless `McapParseErrorPolicy.IGNORE_ERROR` is set. Call `detect_config` to enumerate the file's channels locally and edit the returned `McapImportConfig` (channel selection, names, types, metadata records, parse error policy, complex types mode) before importing. Detection requires the new `mcap` extra: `pip install sift-stack-py[mcap]`. + ## [v0.20.0] - August 25, 2026 ### What's New diff --git a/python/docs/guides/pytest_plugin/index.md b/python/docs/guides/pytest_plugin/index.md index 215dc9048..6d33f1ff3 100644 --- a/python/docs/guides/pytest_plugin/index.md +++ b/python/docs/guides/pytest_plugin/index.md @@ -100,11 +100,11 @@ The plugin runs in one of three modes, picked at invocation. | **Offline** | `--sift-offline` | No; records to a log file for later replay | Environments without Sift access. | | **Disabled** | `--sift-disabled` | No | Local dev. Bounds still evaluate and return a real pass/fail. | -Online mode pings Sift once at session start and aborts if Sift is unreachable or the credentials are invalid, -so a misconfigured job fails immediately instead of silently producing no report. -During the run, every create and update is appended to a JSONL log file. -A background worker uploads new entries to Sift incrementally. -If the connection drops mid-test, the test keeps running and the log keeps writing locally. +Online mode pings Sift once at session start and aborts if Sift is unreachable or the credentials are invalid, +so a misconfigured job fails immediately instead of silently producing no report. +During the run, every create and update is appended to a JSONL log file. +A background worker uploads new entries to Sift incrementally. +If the connection drops mid-test, the test keeps running and the log keeps writing locally. The remaining entries can be uploaded afterward by running import-test-result-log, which the plugin prints on exit. That command resumes into the report the interrupted run created rather than starting a second one. See [Running Modes](running_modes.md) for the log-file and replay pipeline, diff --git a/python/examples/data_import/mcap/main.py b/python/examples/data_import/mcap/main.py new file mode 100644 index 000000000..25ecfbe4c --- /dev/null +++ b/python/examples/data_import/mcap/main.py @@ -0,0 +1,71 @@ +"""Import an MCAP (.mcap) file into Sift. + +MCAP files are self-describing, so the import needs no column mapping: every +channel of every supported topic (ros2msg schemas with cdr messages) is +imported. Channel names combine the topic and flattened field path, e.g. +"/imu/data.orientation.x". + +Swap sample_data.mcap for your own .mcap recording. +""" + +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.mcap", + 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, skip undecodable topics, set a start time for logs on a + # non-Unix epoch, or pick metadata records to import. + # + # from datetime import datetime, timezone + # + # from sift_client.sift_types.data_import import McapParseErrorPolicy + # + # config = client.data_import.detect_config("sample_data.mcap") + # print(config) # inspect every detected channel + # + # # Example: import only the IMU topic + # config.data = [d for d in config.data if d.topic == "/imu/data"] + # + # # Example: skip undecodable topics and records instead of failing + # config.parse_error_policy = McapParseErrorPolicy.IGNORE_ERROR + # + # # Example: reinterpret log_time as elapsed nanoseconds from an explicit + # # start; only for recorders whose clock did not track Unix time + # config.relative_start_time = datetime(2026, 1, 1, tzinfo=timezone.utc) + # + # # Example: import every key of a named metadata record as run metadata + # config.metadata_records = ["calibration"] + # + # import_job = client.data_import.import_from_path( + # "sample_data.mcap", + # asset=asset_name, + # config=config, + # ) + # import_job.wait_until_complete() diff --git a/python/examples/data_import/mcap/requirements.txt b/python/examples/data_import/mcap/requirements.txt new file mode 100644 index 000000000..a0739c91b --- /dev/null +++ b/python/examples/data_import/mcap/requirements.txt @@ -0,0 +1,2 @@ +python-dotenv +sift-stack-py[mcap] diff --git a/python/examples/data_import/mcap/sample_data.mcap b/python/examples/data_import/mcap/sample_data.mcap new file mode 100644 index 0000000000000000000000000000000000000000..0e72f96e324739c15cb56e11064f1ae61ad27864 GIT binary patch literal 6621 zcmb`MX*^W#-+;&1_pR*vUiLL>*_X&4*$IQO#Smjn8ha8EQ}(TB$Sx70#WLBlPN?jp zq9i0@9%r2Seg4ny^Spe{i#g}IukZPud(OF+v!qxg$ zfCnr{8UX2#(4e5eV5ESYjIzu*4FNz1$jd0n$Wiz)f;&R&UdH&aaiA0(4uJI_kl=Xv zEY}l{OFp^ z7(MuBU`A(j{i0@%?iNj#&FASJ)w)!tsix~Q8clDrxre^YX|sJlH41$X#X+;Qe#Jhy z%@4uurI7?Dvrt(bZJ6c@1ncStjpW|TKv-+nW_~(O^@Q&|XBVdh*-(F?G-E``cT$l% z;u>aAg=cd;C)4iFHE4+SncTyNt+Xb&q|&J{G`VnAt{5XVZH!%hE}hG}V{Gx@ZolZy z_jS}JtPPDuZ~GbzDXkR$nhz+6>LC8KoT1hsD`I6vc@#PydM3D-(~9ADA2C1}TCh15%UrR}f}b#Qn-bJ) zFtHl{h_+-Ano>!|Oa1OB@MF>PijfXME}9yry(&*fl%KgvYo@U-YedtVnEOpsEbL80*!|YPcX7D;?F}iB-z< z%#DdI-blmi-6B(V&p1%NRN8WKgSMMvI<|qV&@-9lR-H*qadL5ydlMcOPaa0uM)=HN zs|5#oy9$?kXeo4Dl2n%*man!J4*Syye^&CQfO$^n?&`-r?ukSX;PswkGtziR-jv3_|C*YbWOjFS zjFm*eNWsO`(dCKEgVgJxaGS~I#VFrvZ|_7l*LJqHs#dkxI#YN?I_&tgyUt1?ChXr2 zZ0s+39pZg#?b$7-?0(}}IY!_@`KNJc`T?46RbOKgBd@(VntSyju)UScMMv*N5l(PS zJl0ChOIv2-JKNk>5}Lj=5&(FqR&Qe@1FmjJ?%V^^7-qWb}MgoL8B_jgM_; zzOm^TcQw9Vp6`)|kW0eMu~<%}NfQTuBV*=sn}OHey#%cD<7|H|Wf|Jqbwyp_3#r=o z=Ah!eqbL^P7aL6quYO(__O3bmzA>k%OIsE@ny#^Ae!oTI3(uM!i*Vn>+S`zKH^~1C zC1|jcI7*%N9Z`4kSr**yM)8HtUS=r}drlXwGf~D^t*ejUWtA2;NU}a_52cw2l>A(d z>7@CHP2u7Trb|-o;pf7)m^yASk)Ms$3n=nA(AsMxb6#D}pwmPqYQg?~DB#hhO0!ZJ? zi&c4Kx0%)p3iU?|V|_-hR2HRi{AsMRo_}SD|a5pJOJU#iM@IK z*LZg>h2m)qX7e_OA3GEWa!VLCDt3QIHUys_342;^2B-2t6_4;@KF8odsC{kbq7eRB z1v9mqSbiR3eXiTNs9bXL0(u7pSJBh7&#^~7W2i1~IqNyw=Q)QHzw*b(UbRkJOSVBU zaW^#-*Ug{kz=fHTMLD0}mF4NgPCEo|7qS_J!{m*ws2&C8Mz`9>|4gV{x7e04)(T|~ z3WeAVyLYqTRlHapaYH^CNU~*{OGSCZvP5{Yp?~tIy>-(mD+kDxqGR1ot7S$BT)?1} z@hVZYJE((k0K2cG0PY8qbLoO~?D`R(ia-^a_?3%42`3EQqzv$|~PUk{4lG$iqS zFqr!)aZ@#|*BaCRPEBRFjoxDOGT$}zQnhrih89_O)fb`{6Yebg4D-BJva&}+{f_@W z2-%(UlJ0Sjc+_4Y^WeEUbzQzaZlP?ItCM8R-NDqEI(-~r9Fv<|cl8m+i~{rGe;z5p zhvcdMes~g>>*2>_!t!W~J0-r+#4G=9|cEc}k0* z4fkNutBFzR_%J{76z}Wg4P2c%E{lGzY?$qA-PTI{)T%i})iD}WU4Aiz2F{1(k?I$z zdh5Q8KUC^BdzM?2CtL!*idExPTlw5=CrHbO9eEpWAchtDX>PZ<7k8MHCo|>w)6Nm9 zK(%cha4Mb2uF-w}m1cIj4$9L;rD<*Z@q!B_sMXX(y(|PR zT)eMJ$?Rcb>M3L6L%zPz(OY{r^RsU!x3ecY^pce`wK<|kZ=H7H*)J@ws*o6p+cube zPAMb2NOnF#&ec* zB*y2H{qucA!+n3?Tu|uohjN+NZ`5W(Ue0_cX>A&PM@K z>7h6Ep;`QK9KZL+MkLo4KK9sq$0tQv*rVM3Mr72}6*$jn&LZP*3$vby*-$FjudW+C z^k#K6Lvhcq-e`*IE%nXWU$8XinIl(_w9)Shg~dO~G{Cixsu?mpSDOz};J_RE{|6r+ zsbJW_^L`YW%(AnAS@cc1BL3r%(0w0@nh3NeR3FhFylQlV7G)Uq@^`QN=uwzv3gqZR zo$fh=U@-FH7mK=Lu1rH#c!cEe^$l~b+f7Mkp%bhK+IGoYDpKD`vSk}L_V+!$Zi07VARCN`pB1JV9G6Y+3mt5lZ z_0ptwuug*Lp>Jw@`jmOAuV-k$f%p32j(wA?TALp07;hbmwmSg#btSM442!}1t@&FXGTestovu~ShV-5=vYM^Y!g0^2j4 zrtZ^4CdC}748Ki(Co5_4joQ=j`W>InB3nH90r@F_m^xya(e${Xjc__MyF54G1n3BAsH8m{0>^FI9Qy?UGnK9(dc z{~SU!=sdY5%+|6!{g=iTTe)M#f;9czzn1KZ4UNMH?v7_7I;2iUS7a2N8JycXa*RBm z`U#dr3TqgtOxaSpBPS`R(OR_ITj{1wLG{&et)u48+8hrf5^0j&$S@#eE&i+telug5 z&1{`oCbPT4%tjvIk^M9wO_6Pw-GReqj)58XzBPFP#u4mksyL8-zY6|i$A~p*AfUqH z;%cH5Hr9ERED6h=DV)*29T2KEu#X8UTHDC+tya^kQWg+f=~y2UV4O6eEteI=<5pEh zm$BsVnP5#t%x?7s%T76?XG%@$8Li2erK?khn4W4snT@=^G15kscwK^5rXJP5+Z*d-Piy>?eJ);>4k;x$8@OytjnN4Vm3#j(5Y^Pxbkf6 zNG%kfdG0{-%8MY)y*~#rDpYisxZ7j~MRzFbMF+PhT7#v&6mFK{3C1T;)&>Pq8g5suA(MG)PB(>VDPWzGE;rTRE zQ45c24Jol^{>74X>W=aUr>433{aDDCeqKsJ zxL3px; zkU+EvTPF|R=_DtmsQW(%58`sr6NX6NdIda?$AgS|Bk&qv9ML;IC-50kaQ{uqH-L`@ z8It^!!0(8jH@+M2Lcmi!usr4N0XP(7)WxR+#uEL4SufzZ$xp&j;VT5r1sTE+(+Bu< zqJPAO1^5E^FCiLr1il9{YSid0;8%#=XP&^Slwkg&hDEkithEjw6N5~N1Z9qmP(hUQQA$p#20v`~4qt-jXo6~^%o8Sms3NmUA zF#`B)q7S1S1=xrdtUp}|f%8CyJX9S6{07mhjS~1O9hiT_cO3AYAfvpX696;QgZ?m& zz@8wZvJ@u)-$L|f2MJ8W03QFPCl2seK!(^I61bk|H_-0^PfC38lOQt%ur0_au08@! z5PgH|G~mUV!2Cvg1dak3GMh02_#vW4p7{W<05e#B+SddQ0U1^5_!02EM6ddbz?>{# z{^8V5fcFO(0vDMDxQplyItZ-A3g*9OGY9w#kWmuHx0}bjMD$%r^8l-y0`uGOF94hZ zGGx7lz#oY|&SDYpvZulPoL>kW2QsQIb_wv4L~qEo46pwA1oMkC zeFZoZWYh}`ff>0#Z>{qU@a`a^zD^UkndoDK*8xw(4IZDJb^~B1kRi1X30zC``kLPX zPr?J{|1d#dYmiab0)7DgG0{^|Yy!Mb^hHGkj^G9NS3dU>@B<*D2Hp{vj}P?jK3jml z1~O_7@(bW@qQ9F*J}0WklU0|I-1jJlb<3-|@1 zXOsR7u#(_OJgT~fz{wy(^j!7;KST7>y9AaNI*Eq_-Tec26v!wFiG6^_iN5e9fyKZV zvcwm(kswo6fO(*h62jKWbV3dVi^1U=K^(5Z;qpCk_ymV1ZQ^hQ4hNOQ?hkf97Gie- zoBMTQ^9GyOGO@XVm-7UygfTX10q(=e)(ucqiA%VfJ0U_QYvd2HN zCIO+qpbxykKE@Bno8Iui05~!@@+8I+g zfFDP}JUrn3@L(9yHxNOi_5ZxkT)J$eF7SUIpT^5S5QbDy5{UR`EApSMC>m1hKv1<8>}({* zcOAe)C+iy!D(eYD!hpL156|G^UkDkoM?;V605lI-_mhMER}8H#5P7^HPFBZ}#D)eG zoLtWT{UreyS4kkA|HKpLO)7h^2=cE_587aP2-pNq5;TEyd_b`29f5ZcK#ThC&lAK? z7YxK1hO`sh!xwCB{}?iHaMUybkNJT-I=HW literal 0 HcmV?d00001 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 cd430a773..83fc1e088 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 @@ -17,6 +17,7 @@ CsvImportConfig, Hdf5ImportConfig, ImportConfig, + McapImportConfig, ParquetFlatDatasetImportConfig, ParquetSingleChannelPerRowImportConfig, TdmsImportConfig, @@ -47,6 +48,8 @@ def _set_config_on_request( request.hdf5_config.CopyFrom(config._to_proto()) elif isinstance(config, UlogImportConfig): request.ulog_config.CopyFrom(config._to_proto()) + elif isinstance(config, McapImportConfig): + request.mcap_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/mcap.py b/python/lib/sift_client/_internal/util/mcap.py new file mode 100644 index 000000000..56326545f --- /dev/null +++ b/python/lib/sift_client/_internal/util/mcap.py @@ -0,0 +1,430 @@ +"""Detect channels in MCAP (``.mcap``) files. + +Detection reads the file's schema and channel records without decoding +message payloads, then flattens each topic's ros2msg schema into importable +leaf fields, mirroring the server importer's rules. + +See ``detect_mcap_config`` for caveats on unsupported topics and empty +``data`` behavior. +""" + +from __future__ import annotations + +import warnings +from collections import defaultdict +from pathlib import Path +from typing import NamedTuple + +from mcap import records as mcap_records +from mcap.reader import make_reader +from mcap.stream_reader import StreamReader +from mcap_ros2 import _dynamic as ros2_dynamic + +from sift_client.sift_types.channel import ChannelDataType +from sift_client.sift_types.data_import import McapDataColumn, McapImportConfig + +MCAP_MAGIC = b"\x89MCAP0\r\n" + +# The importer rejects chunk compressions outside this set. +SUPPORTED_COMPRESSIONS = frozenset(("", "zstd", "lz4")) + +# ROS 2 scalar types to Sift channel types. Narrow integers widen to 32-bit +# like the other import types. byte and char both map to UINT_32: ROS 2 +# defines them as unsigned 8-bit (octet and uint8). +ROS2_TO_SIFT_TYPE: dict[str, ChannelDataType] = { + "bool": ChannelDataType.BOOL, + "int8": ChannelDataType.INT_32, + "int16": ChannelDataType.INT_32, + "int32": ChannelDataType.INT_32, + "int64": ChannelDataType.INT_64, + "uint8": ChannelDataType.UINT_32, + "uint16": ChannelDataType.UINT_32, + "uint32": ChannelDataType.UINT_32, + "uint64": ChannelDataType.UINT_64, + "byte": ChannelDataType.UINT_32, + "char": ChannelDataType.UINT_32, + "float32": ChannelDataType.FLOAT, + "float64": ChannelDataType.DOUBLE, + "string": ChannelDataType.STRING, +} + +# builtin_interfaces Time and Duration import as one INT_64 nanosecond channel. +TIME_MESSAGE_TYPES = frozenset(("builtin_interfaces/Time", "builtin_interfaces/Duration")) +ROS2_TIME_TYPE = "__time__" + +# Suffix of the JSON expansion of a variable-cardinality field. +JSON_CHANNEL_SUFFIX = ".json" + +# Guards malformed schemas with self-referential fixed nesting. +MAX_FIELD_DEPTH = 32 + + +class UnsupportedTopicError(Exception): + """The topic's schema cannot be decoded by the importer.""" + + +class LeafField(NamedTuple): + """A leaf field of a topic's message type. Scalar leaves carry one value + per message; complex leaves are variable-cardinality and expand per the + complex types import mode. + """ + + field_path: str + kind: str # "scalar" or "complex" + # The ROS 2 base type for scalar leaves (ROS2_TIME_TYPE for + # builtin_interfaces Time/Duration). None for complex leaves. + ros_type: str | None + + def sift_type(self) -> ChannelDataType: + if self.ros_type == ROS2_TIME_TYPE: + return ChannelDataType.INT_64 + return ROS2_TO_SIFT_TYPE[self.ros_type] # type: ignore[index] + + +class TopicInfo(NamedTuple): + """A supported topic and its importable leaves.""" + + topic: str + leaves: list[LeafField] + + +def parse_schema_defs(schema: mcap_records.Schema): + """Parse a ros2msg concatenated schema into (root msgdef, msgdefs by + name), using the same vendored parser the server importer uses.""" + msgdefs: dict = { + "builtin_interfaces/Time": ros2_dynamic.TimeDefinition, + "builtin_interfaces/Duration": ros2_dynamic.TimeDefinition, + } + + def handle(cur_schema_name: str, short_name: str, msgdef) -> None: + msgdefs[cur_schema_name] = msgdef + msgdefs[short_name] = msgdef + + try: + text = schema.data.decode("utf-8") + ros2_dynamic._for_each_msgdef(schema.name, text, handle) + except Exception as e: + raise UnsupportedTopicError(f"its schema failed to parse ({e})") from e + + root = msgdefs.get(schema.name) or msgdefs.get( + "/".join((schema.name.split("/")[0], schema.name.split("/")[-1])) + ) + if root is None: + raise UnsupportedTopicError("its schema does not define the root message") + return root, msgdefs + + +def _is_variable_array(ftype) -> bool: + """Unbounded and bounded ([<=N]) arrays decode with a dynamic length.""" + return ftype.is_array and (ftype.array_size is None or ftype.is_upper_bound) + + +def _check_primitive_supported(type_name: str, label: str) -> None: + if type_name == "wstring": + raise UnsupportedTopicError( + f"field '{label}' uses wstring, which the decoder does not implement" + ) + if type_name not in ROS2_TO_SIFT_TYPE: + raise UnsupportedTopicError(f"field '{label}' has unsupported type '{type_name}'") + + +def _resolve_or_raise(msgdefs: dict, ftype, label: str): + nested = msgdefs.get(f"{ftype.pkg_name}/{ftype.type}") + if nested is None: + raise UnsupportedTopicError( + f"field '{label}' has unknown type '{ftype.pkg_name}/{ftype.type}'" + ) + return nested + + +def _check_ftype_decodable( + msgdefs: dict, ftype, label: str, visited: frozenset = frozenset(), depth: int = 0 +) -> None: + """Raise UnsupportedTopicError if the field type cannot be decoded. + + Variable-cardinality fields do not expand into leaves, but the importer + still decodes every element, so wstring or unknown types anywhere in the + subtree make the whole topic unsupported. + """ + if ftype.is_primitive_type(): + _check_primitive_supported(ftype.type, label) + return + if f"{ftype.pkg_name}/{ftype.type}" in TIME_MESSAGE_TYPES: + return + _check_subtree_decodable(msgdefs, _resolve_or_raise(msgdefs, ftype, label), visited, depth) + + +def _check_subtree_decodable(msgdefs: dict, msgdef, visited: frozenset, depth: int) -> None: + if depth > MAX_FIELD_DEPTH: + raise UnsupportedTopicError(f"its schema nests deeper than {MAX_FIELD_DEPTH} levels") + name = f"{msgdef.base_type.pkg_name}/{msgdef.msg_name}" + if name in visited: + return + visited = visited | {name} + for field in msgdef.fields: + _check_ftype_decodable(msgdefs, field.type, field.name, visited, depth + 1) + + +def expand_message_fields(root_msgdef, msgdefs: dict) -> list[LeafField]: + """Flatten the root message into importable leaves. + + Scalar leaves append their dot-delimited field path. Fixed-size arrays + expand with bracketed indexes. Variable-cardinality fields become one + complex leaf. Named constants are not message fields and never appear. + """ + leaves: list[LeafField] = [] + + def walk(prefix: str, msgdef, depth: int) -> None: + if depth > MAX_FIELD_DEPTH: + raise UnsupportedTopicError(f"its schema nests deeper than {MAX_FIELD_DEPTH} levels") + for field in msgdef.fields: + ftype = field.type + path = f"{prefix}{field.name}" + if _is_variable_array(ftype): + # The importer decodes every element even though the leaf is + # imported whole, so verify the element type decodes. + _check_ftype_decodable(msgdefs, ftype, path) + leaves.append(LeafField(path, "complex", None)) + continue + + indexes = [f"[{i}]" for i in range(ftype.array_size)] if ftype.is_array else [""] + if ftype.is_primitive_type(): + _check_primitive_supported(ftype.type, path) + leaves.extend( + LeafField(f"{path}{index}", "scalar", ftype.type) for index in indexes + ) + elif f"{ftype.pkg_name}/{ftype.type}" in TIME_MESSAGE_TYPES: + leaves.extend( + LeafField(f"{path}{index}", "scalar", ROS2_TIME_TYPE) for index in indexes + ) + else: + nested = _resolve_or_raise(msgdefs, ftype, path) + for index in indexes: + walk(f"{path}{index}.", nested, depth + 1) + + walk("", root_msgdef, 0) + return leaves + + +def _read_schemas_and_channels( + path: Path, +) -> tuple[dict[int, mcap_records.Schema], list[mcap_records.Channel], list[str]]: + """Read schema and channel records without decoding message payloads. + + Mirrors the server importer's scan: a top-level pass validates chunk + compression and collects the records of unchunked files, the summary + section supplies the records of chunked files, and a file without a + readable summary (e.g. truncated) is scanned through its decompressed + chunks instead, keeping what parsed with a warning. + """ + parse_warnings: list[str] = [] + schemas: dict[int, mcap_records.Schema] = {} + channels: list[mcap_records.Channel] = [] + seen_channel_ids: set[int] = set() + + def add_channel(channel: mcap_records.Channel) -> None: + if channel.id not in seen_channel_ids: + seen_channel_ids.add(channel.id) + channels.append(channel) + + def scan(stream: StreamReader) -> None: + records = iter(stream.records) + while True: + try: + record = next(records) + except StopIteration: + return + except Exception as e: + # Truncation errors often stringify empty, so always say something. + detail = str(e) or type(e).__name__ + message = ( + "stopped reading at an unparseable record; the detected " + f"channels may be incomplete: {detail}" + ) + # Both passes hit the same broken spot. + if message not in parse_warnings: + parse_warnings.append(message) + return + if isinstance(record, mcap_records.Chunk): + # The importer rejects unsupported compression regardless of + # parse_error_policy; the mcap reader would silently treat it + # as uncompressed. + if record.compression not in SUPPORTED_COMPRESSIONS: + raise ValueError( + f"unsupported chunk compression '{record.compression}'; " + "supported compressions are none, zstd, and lz4" + ) + elif isinstance(record, mcap_records.Schema): + schemas[record.id] = record + elif isinstance(record, mcap_records.Channel): + add_channel(record) + + with open(path, "rb") as file: + if file.read(len(MCAP_MAGIC)) != MCAP_MAGIC: + raise ValueError(f"'{path.name}' is not an MCAP file (bad magic bytes)") + file.seek(0) + + # Top-level pass: chunks stay unopened, so this validates compression + # cheaply and picks up the records of unchunked files. + scan(StreamReader(file, emit_chunks=True)) + + # Chunked files carry their records inside chunks; the summary section + # repeats them. Without a readable summary (e.g. a truncated file), + # scan through the decompressed chunks instead. + file.seek(0) + try: + summary = make_reader(file).get_summary() + except Exception: + summary = None + if summary is not None: + schemas.update(summary.schemas) + for channel in sorted(summary.channels.values(), key=lambda c: c.id): + add_channel(channel) + else: + file.seek(0) + scan(StreamReader(file, emit_chunks=False)) + + return schemas, channels, parse_warnings + + +def detect_mcap_topics( + schemas: dict[int, mcap_records.Schema], + channels: list[mcap_records.Channel], + parse_warnings: list[str], +) -> list[TopicInfo]: + """Derive the supported topics and their importable leaves. + + Same-topic channels merge only when their schemas and message encodings + match. Distinct topics colliding case-insensitively keep the first. + Unsupported topics are skipped with a warning; the import itself gates + them on ``parse_error_policy``. + """ + channels_by_topic: defaultdict[str, list[mcap_records.Channel]] = defaultdict(list) + for channel in channels: + channels_by_topic[channel.topic].append(channel) + + # Sift channel names compare case-insensitively, so distinct topics + # colliding only by case conflict; the first wins. + kept_by_lower: dict[str, str] = {} + for topic in channels_by_topic: + first = kept_by_lower.setdefault(topic.lower(), topic) + if first != topic: + parse_warnings.append( + f"topic '{topic}' collides with topic '{first}' by case only; kept the first" + ) + + topics: list[TopicInfo] = [] + unsupported: dict[str, str] = {} + for topic, topic_channels in channels_by_topic.items(): + if kept_by_lower[topic.lower()] != topic: + continue + # Same-topic channels merge only when they agree. + encodings = {c.message_encoding for c in topic_channels} + topic_schemas = [schemas.get(c.schema_id) for c in topic_channels] + signatures = {None if s is None else (s.name, s.encoding, s.data) for s in topic_schemas} + if len(encodings) > 1 or len(signatures) > 1: + unsupported[topic] = ( + "it has multiple channels with mismatched schemas or message encodings" + ) + continue + channel = topic_channels[0] + schema = topic_schemas[0] + if schema is None: + unsupported[topic] = "it has no schema" + continue + if channel.message_encoding != "cdr": + unsupported[topic] = ( + f"its message encoding is '{channel.message_encoding}' (only cdr is supported)" + ) + continue + if schema.encoding != "ros2msg": + unsupported[topic] = ( + f"its schema encoding is '{schema.encoding}' (only ros2msg is supported)" + ) + continue + try: + root, msgdefs = parse_schema_defs(schema) + leaves = expand_message_fields(root, msgdefs) + except UnsupportedTopicError as e: + unsupported[topic] = str(e) + continue + topics.append(TopicInfo(topic=topic, leaves=leaves)) + + if unsupported: + details = "; ".join(f"'{t}': {reason}" for t, reason in sorted(unsupported.items())) + parse_warnings.append(f"skipped unsupported topics: {details}") + return topics + + +def detect_mcap_fields(topics: list[TopicInfo]) -> list[McapDataColumn]: + """Return importable channels as ``McapDataColumn``s with default names + and data types. + + Scalar leaves become one channel named ``.``. Complex + leaves expand like the default complex types import mode (``BOTH``): Arrow + IPC bytes under the base name and a JSON string under ``.json``. + """ + channels: list[McapDataColumn] = [] + # Sift channel names are unique per asset and compare case-insensitively. + taken_names: dict[str, str] = {} + for topic in topics: + for leaf in topic.leaves: + base_name = f"{topic.topic}.{leaf.field_path}" + if leaf.kind == "scalar": + expansions = [(base_name, leaf.sift_type())] + else: + expansions = [ + (base_name, ChannelDataType.BYTES), + (base_name + JSON_CHANNEL_SUFFIX, ChannelDataType.STRING), + ] + for name, data_type in expansions: + existing = taken_names.get(name.lower()) + if existing is not None: + raise ValueError( + f"the generated channel name '{name}' conflicts with channel '{existing}'" + ) + taken_names[name.lower()] = name + channels.append( + McapDataColumn( + topic=topic.topic, + field_path=leaf.field_path, + name=name, + data_type=data_type, + ) + ) + return channels + + +def detect_mcap_config(file_path: str | Path, asset_name: str = "") -> McapImportConfig: + """Detect an MCAP import config by enumerating the file's channels. + + Channels come from the file's schema and channel records; message payloads + are not read, so a topic is listed even when it logged no messages. Topics + the importer does not support (non-cdr message encodings, non-ros2msg + schemas, undecodable schemas) are skipped with a warning; importing such a + file fails under the default parse error policy, so set + ``McapParseErrorPolicy.IGNORE_ERROR`` to import the rest. + + Args: + file_path: Path to the ``.mcap`` 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. + + Raises: + ValueError: If the file is not MCAP, uses an unsupported chunk + compression, or two detected channels generate the same Sift + channel name. The importer rejects all three regardless of + ``parse_error_policy``. + """ + path = Path(file_path) + schemas, channels, parse_warnings = _read_schemas_and_channels(path) + topics = detect_mcap_topics(schemas, channels, parse_warnings) + data = detect_mcap_fields(topics) + for message in parse_warnings: + warnings.warn(f"'{path.name}': {message}", stacklevel=2) + return McapImportConfig(asset_name=asset_name, data=data) diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data_imports.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data_imports.py new file mode 100644 index 000000000..ba2f0939c --- /dev/null +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_data_imports.py @@ -0,0 +1,34 @@ +"""Tests for the data imports low-level wrapper.""" + +from __future__ import annotations + +import pytest +from sift.data_imports.v2.data_imports_pb2 import CreateDataImportFromUploadRequest + +from sift_client._internal.low_level_wrappers.data_imports import _set_config_on_request +from sift_client.sift_types.channel import ChannelDataType +from sift_client.sift_types.data_import import McapDataColumn, McapImportConfig + + +class TestSetConfigOnRequest: + def test_mcap_config_sets_mcap_field(self): + request = CreateDataImportFromUploadRequest() + config = McapImportConfig( + asset_name="my_asset", + data=[ + McapDataColumn( + topic="/imu", field_path="orientation.x", data_type=ChannelDataType.DOUBLE + ) + ], + ) + + _set_config_on_request(request, config) + + assert request.HasField("mcap_config") + assert request.mcap_config.asset_name == "my_asset" + assert request.mcap_config.data[0].ros2.field_path == "orientation.x" + + def test_unknown_config_type_raises(self): + request = CreateDataImportFromUploadRequest() + with pytest.raises(TypeError, match="Unsupported import config type"): + _set_config_on_request(request, object()) # type: ignore[arg-type] diff --git a/python/lib/sift_client/_tests/_internal/test_mcap.py b/python/lib/sift_client/_tests/_internal/test_mcap.py new file mode 100644 index 000000000..47b8306c3 --- /dev/null +++ b/python/lib/sift_client/_tests/_internal/test_mcap.py @@ -0,0 +1,376 @@ +"""Tests for MCAP channel detection.""" + +from __future__ import annotations + +import pytest +from mcap.records import Channel, Schema +from mcap.writer import Writer + +from sift_client._internal.util.mcap import ( + UnsupportedTopicError, + detect_mcap_config, + detect_mcap_topics, + expand_message_fields, + parse_schema_defs, +) +from sift_client.sift_types.channel import ChannelDataType + + +def _schema(data: str, schema_id: int = 1, name: str = "test_msgs/msg/Test") -> Schema: + return Schema(id=schema_id, data=data.encode(), encoding="ros2msg", name=name) + + +def _channel(topic: str, schema_id: int = 1, channel_id: int = 1, encoding: str = "cdr") -> Channel: + return Channel( + id=channel_id, topic=topic, message_encoding=encoding, metadata={}, schema_id=schema_id + ) + + +def _leaves(data: str, name: str = "test_msgs/msg/Test"): + root, msgdefs = parse_schema_defs(_schema(data, name=name)) + return expand_message_fields(root, msgdefs) + + +IMU_SCHEMA = """geometry_msgs/Vector3 gyro +float64 temp +================================================================================ +MSG: geometry_msgs/Vector3 +float64 x +float64 y +float64 z +""" + + +def _write_mcap(path, schemas_and_topics: list[tuple[str, str, str]]) -> None: + """Write an MCAP file with one channel per (schema_name, schema_text, topic).""" + with open(path, "wb") as f: + writer = Writer(f) + writer.start() + for schema_name, schema_text, topic in schemas_and_topics: + schema_id = writer.register_schema( + name=schema_name, encoding="ros2msg", data=schema_text.encode() + ) + writer.register_channel(topic=topic, message_encoding="cdr", schema_id=schema_id) + writer.finish() + + +class TestExpandMessageFields: + def test_scalars(self): + assert _leaves("float64 x\nuint32 seq\nstring status\n") == [ + ("x", "scalar", "float64"), + ("seq", "scalar", "uint32"), + ("status", "scalar", "string"), + ] + + def test_nested_message_uses_dotted_paths(self): + leaves = _leaves(IMU_SCHEMA) + assert [leaf.field_path for leaf in leaves] == ["gyro.x", "gyro.y", "gyro.z", "temp"] + + def test_fixed_array_expands_per_element(self): + assert [leaf.field_path for leaf in _leaves("float32[3] accel\n")] == [ + "accel[0]", + "accel[1]", + "accel[2]", + ] + + def test_variable_array_is_one_complex_leaf(self): + assert _leaves("int32[] samples\n") == [("samples", "complex", None)] + + def test_bounded_array_is_one_complex_leaf(self): + assert _leaves("int32[<=4] samples\n") == [("samples", "complex", None)] + + def test_variable_array_of_messages_is_one_complex_leaf(self): + schema = ( + "geometry_msgs/Vector3[] path\n" + + "=" * 80 + + "\nMSG: geometry_msgs/Vector3\nfloat64 x\nfloat64 y\nfloat64 z\n" + ) + assert _leaves(schema) == [("path", "complex", None)] + + def test_fixed_array_of_messages_expands_per_element(self): + schema = ( + "geometry_msgs/Vector3[2] corners\n" + + "=" * 80 + + "\nMSG: geometry_msgs/Vector3\nfloat64 x\nfloat64 y\nfloat64 z\n" + ) + assert [leaf.field_path for leaf in _leaves(schema)] == [ + "corners[0].x", + "corners[0].y", + "corners[0].z", + "corners[1].x", + "corners[1].y", + "corners[1].z", + ] + + def test_time_and_duration_collapse_to_int64(self): + leaves = _leaves("builtin_interfaces/Time stamp\nbuiltin_interfaces/Duration elapsed\n") + assert [leaf.field_path for leaf in leaves] == ["stamp", "elapsed"] + assert all(leaf.sift_type() == ChannelDataType.INT_64 for leaf in leaves) + + def test_constants_are_not_fields(self): + assert _leaves("int32 STATUS_OK=0\nint32 status\n") == [ + ("status", "scalar", "int32"), + ] + + def test_maps_every_ros2_scalar_type(self): + # Narrow ints widen to 32-bit; byte and char are unsigned 8-bit in ROS 2. + definition = "\n".join( + f"{ros_type} f_{ros_type}" + for ros_type in ( + "bool", + "int8", + "int16", + "int32", + "int64", + "uint8", + "uint16", + "uint32", + "uint64", + "byte", + "char", + "float32", + "float64", + "string", + ) + ) + assert [leaf.sift_type() for leaf in _leaves(definition)] == [ + ChannelDataType.BOOL, + ChannelDataType.INT_32, + ChannelDataType.INT_32, + ChannelDataType.INT_32, + ChannelDataType.INT_64, + ChannelDataType.UINT_32, + ChannelDataType.UINT_32, + ChannelDataType.UINT_32, + ChannelDataType.UINT_64, + ChannelDataType.UINT_32, + ChannelDataType.UINT_32, + ChannelDataType.FLOAT, + ChannelDataType.DOUBLE, + ChannelDataType.STRING, + ] + + def test_wstring_raises_unsupported(self): + with pytest.raises(UnsupportedTopicError, match="wstring"): + _leaves("wstring label\n") + + def test_variable_array_of_wstring_raises(self): + # The importer decodes every element of a complex leaf, so an + # undecodable element type makes the whole topic unsupported. + with pytest.raises(UnsupportedTopicError, match="wstring"): + _leaves("wstring[] labels\n") + + def test_variable_array_of_messages_with_wstring_raises(self): + schema = "pkg/Bad[] items\n" + "=" * 80 + "\nMSG: pkg/Bad\nwstring label\n" + with pytest.raises(UnsupportedTopicError, match="wstring"): + _leaves(schema) + + def test_nesting_beyond_max_depth_raises(self): + # A chain of 40 nested message types exceeds MAX_FIELD_DEPTH (32). + parts = ["pkg/M0 child\n"] + parts.extend("=" * 80 + f"\nMSG: pkg/M{i}\npkg/M{i + 1} child\n" for i in range(40)) + parts.append("=" * 80 + "\nMSG: pkg/M40\nfloat64 x\n") + with pytest.raises(UnsupportedTopicError, match="nests deeper"): + _leaves("".join(parts)) + + def test_unknown_nested_type_raises(self): + with pytest.raises(UnsupportedTopicError, match="unknown type"): + _leaves("other_msgs/Missing part\n") + + def test_root_message_missing_raises(self): + with pytest.raises(UnsupportedTopicError, match="root message"): + parse_schema_defs( + Schema( + id=1, + data=b"MSG: other_msgs/Other\nfloat64 x\n", + encoding="ros2msg", + name="test_msgs/msg/Test", + ) + ) + + +class TestDetectMcapTopics: + def test_supported_topic_yields_leaves(self): + warnings: list[str] = [] + topics = detect_mcap_topics({1: _schema(IMU_SCHEMA)}, [_channel("/imu")], warnings) + assert [t.topic for t in topics] == ["/imu"] + assert [leaf.field_path for leaf in topics[0].leaves] == [ + "gyro.x", + "gyro.y", + "gyro.z", + "temp", + ] + assert warnings == [] + + def test_non_cdr_encoding_skipped_with_warning(self): + warnings: list[str] = [] + topics = detect_mcap_topics( + {1: _schema(IMU_SCHEMA)}, [_channel("/imu", encoding="json")], warnings + ) + assert topics == [] + assert any("only cdr is supported" in w for w in warnings) + + def test_non_ros2msg_schema_skipped_with_warning(self): + schema = Schema(id=1, data=b"{}", encoding="jsonschema", name="Test") + warnings: list[str] = [] + assert detect_mcap_topics({1: schema}, [_channel("/diag")], warnings) == [] + assert any("only ros2msg is supported" in w for w in warnings) + + def test_missing_schema_skipped_with_warning(self): + warnings: list[str] = [] + assert detect_mcap_topics({}, [_channel("/imu")], warnings) == [] + assert any("no schema" in w for w in warnings) + + def test_case_colliding_topics_keep_first(self): + schemas = {1: _schema(IMU_SCHEMA)} + channels = [ + _channel("/imu", channel_id=1), + _channel("/IMU", channel_id=2), + ] + warnings: list[str] = [] + topics = detect_mcap_topics(schemas, channels, warnings) + assert [t.topic for t in topics] == ["/imu"] + assert any("by case only" in w for w in warnings) + + def test_same_topic_channels_merge_when_identical(self): + schemas = {1: _schema(IMU_SCHEMA)} + channels = [_channel("/imu", channel_id=1), _channel("/imu", channel_id=2)] + warnings: list[str] = [] + topics = detect_mcap_topics(schemas, channels, warnings) + assert [t.topic for t in topics] == ["/imu"] + assert warnings == [] + + def test_same_topic_mismatched_schemas_skipped(self): + schemas = { + 1: _schema(IMU_SCHEMA, schema_id=1), + 2: _schema("float64 other\n", schema_id=2), + } + channels = [ + _channel("/imu", schema_id=1, channel_id=1), + _channel("/imu", schema_id=2, channel_id=2), + ] + warnings: list[str] = [] + assert detect_mcap_topics(schemas, channels, warnings) == [] + assert any("mismatched schemas" in w for w in warnings) + + +class TestDetectMcapConfig: + def test_detects_channel_per_field(self, tmp_path): + path = tmp_path / "log.mcap" + _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu/data")]) + + config = detect_mcap_config(path, asset_name="robot") + assert config.asset_name == "robot" + assert len(config.data) == 4 + channels = {(d.topic, d.field_path, d.name, d.data_type) for d in config.data} + assert channels == { + ("/imu/data", "gyro.x", "/imu/data.gyro.x", ChannelDataType.DOUBLE), + ("/imu/data", "gyro.y", "/imu/data.gyro.y", ChannelDataType.DOUBLE), + ("/imu/data", "gyro.z", "/imu/data.gyro.z", ChannelDataType.DOUBLE), + ("/imu/data", "temp", "/imu/data.temp", ChannelDataType.DOUBLE), + } + + def test_complex_field_expands_to_bytes_and_json(self, tmp_path): + path = tmp_path / "log.mcap" + _write_mcap(path, [("test_msgs/msg/Samples", "int32[] samples\n", "/samples")]) + + config = detect_mcap_config(path) + assert [(d.field_path, d.name, d.data_type) for d in config.data] == [ + ("samples", "/samples.samples", ChannelDataType.BYTES), + ("samples", "/samples.samples.json", ChannelDataType.STRING), + ] + + def test_unsupported_topic_warns_and_keeps_supported(self, tmp_path): + path = tmp_path / "log.mcap" + with open(path, "wb") as f: + writer = Writer(f) + writer.start() + imu = writer.register_schema( + name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() + ) + writer.register_channel(topic="/imu", message_encoding="cdr", schema_id=imu) + diag = writer.register_schema(name="diag", encoding="jsonschema", data=b"{}") + writer.register_channel(topic="/diag", message_encoding="json", schema_id=diag) + writer.finish() + + with pytest.warns(UserWarning, match="skipped unsupported topics"): + config = detect_mcap_config(path) + assert {d.topic for d in config.data} == {"/imu"} + + def test_clean_file_does_not_warn(self, tmp_path, recwarn): + path = tmp_path / "log.mcap" + _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")]) + + detect_mcap_config(path) + assert not [w for w in recwarn.list if issubclass(w.category, UserWarning)] + + def test_truncated_file_scans_linearly(self, tmp_path): + path = tmp_path / "log.mcap" + _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")]) + # Cutting the trailing magic invalidates the footer, so detection + # falls back to a linear scan of the intact data section. + data = path.read_bytes() + path.write_bytes(data[:-8]) + + with pytest.warns(UserWarning, match="stopped reading"): + config = detect_mcap_config(path) + assert {d.topic for d in config.data} == {"/imu"} + + def test_rejects_non_mcap_file(self, tmp_path): + path = tmp_path / "log.mcap" + path.write_bytes(b"NOTMCAP!" + b"\x00" * 64) + with pytest.raises(ValueError, match="not an MCAP file"): + detect_mcap_config(path) + + def test_records_only_in_data_section_are_detected(self, tmp_path): + # An unchunked file whose summary omits the schema/channel repeats is + # spec-legal; the top-level pass must pick the records up. + path = tmp_path / "log.mcap" + with open(path, "wb") as f: + writer = Writer(f, use_chunking=False, repeat_channels=False, repeat_schemas=False) + writer.start() + schema_id = writer.register_schema( + name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() + ) + writer.register_channel(topic="/imu", message_encoding="cdr", schema_id=schema_id) + writer.finish() + + config = detect_mcap_config(path) + assert {d.topic for d in config.data} == {"/imu"} + + def test_rejects_unsupported_chunk_compression(self, tmp_path): + # The importer rejects unsupported compression regardless of + # parse_error_policy, so detection must too rather than listing + # channels for a file that can never import. + path = tmp_path / "log.mcap" + with open(path, "wb") as f: + writer = Writer(f) # defaults to zstd-compressed chunks + writer.start() + schema_id = writer.register_schema( + name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() + ) + channel_id = writer.register_channel( + topic="/imu", message_encoding="cdr", schema_id=schema_id + ) + writer.add_message(channel_id=channel_id, log_time=0, data=b"\x00", publish_time=0) + writer.finish() + # Rewrite the chunk's compression string to an unsupported one. + path.write_bytes(path.read_bytes().replace(b"zstd", b"lzma")) + + with pytest.raises(ValueError, match="unsupported chunk compression"): + detect_mcap_config(path) + + def test_duplicate_generated_names_raise(self, tmp_path): + # Topic '/a' with variable array field 'b' expands to '/a.b.json', + # colliding with topic '/a.b' scalar field 'json'. The importer + # rejects the file the same way. + path = tmp_path / "log.mcap" + _write_mcap( + path, + [ + ("pkg/msg/A", "int32[] b\n", "/a"), + ("pkg/msg/B", "int32 json\n", "/a.b"), + ], + ) + with pytest.raises(ValueError, match="conflicts with channel"): + detect_mcap_config(path) 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 c960437a4..94b9ca44f 100644 --- a/python/lib/sift_client/_tests/resources/test_data_imports.py +++ b/python/lib/sift_client/_tests/resources/test_data_imports.py @@ -4,6 +4,7 @@ from datetime import datetime, timezone from typing import TYPE_CHECKING, cast +from unittest.mock import AsyncMock, MagicMock import pytest from sift.common.type.v1.channel_config_pb2 import ChannelConfig as ChannelConfigProto @@ -34,6 +35,10 @@ DataTypeKey, Hdf5DataColumn, Hdf5ImportConfig, + McapComplexTypesImportMode, + McapDataColumn, + McapImportConfig, + McapParseErrorPolicy, ParquetDataColumn, ParquetFlatDatasetImportConfig, ParquetSingleChannelPerRowImportConfig, @@ -458,6 +463,178 @@ def test_getitem_not_found(self): self._config()["nonexistent"] +class TestMcapConfig: + def _config(self): + return McapImportConfig( + asset_name="my_asset", + run_name="run1", + data=[ + McapDataColumn( + topic="/imu/data", + field_path="orientation.x", + data_type=ChannelDataType.DOUBLE, + ), + McapDataColumn( + topic="/battery", + field_path="voltage", + name="battery_voltage", + data_type=ChannelDataType.FLOAT, + units="V", + description="pack voltage", + ), + ], + metadata_records=["calibration"], + parse_error_policy=McapParseErrorPolicy.IGNORE_ERROR, + complex_types_import_mode=McapComplexTypesImportMode.STRING, + ) + + def test_to_proto(self): + proto = self._config()._to_proto() + assert proto.asset_name == "my_asset" + assert proto.run_name == "run1" + assert len(proto.data) == 2 + assert proto.data[0].topic == "/imu/data" + assert proto.data[0].ros2.field_path == "orientation.x" + assert proto.data[0].channel_config.name == "/imu/data.orientation.x" + assert proto.data[1].topic == "/battery" + assert proto.data[1].channel_config.name == "battery_voltage" + assert proto.data[1].channel_config.units == "V" + assert list(proto.metadata_records) == ["calibration"] + + def test_to_proto_defaults(self): + """An empty config imports all channels; the default policy fails on + error and imports complex fields as both bytes and JSON strings. + """ + from sift.data_imports.v2.data_imports_pb2 import ( + MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH, + MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR, + ) + + proto = McapImportConfig(asset_name="a")._to_proto() + assert len(proto.data) == 0 + assert proto.run_id == "" + assert proto.parse_error_policy == MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR + assert proto.complex_types_import_mode == MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH + assert not proto.HasField("relative_start_time") + + def test_relative_start_time_round_trips(self): + config = McapImportConfig( + asset_name="a", + relative_start_time=datetime(2026, 1, 1, tzinfo=timezone.utc), + ) + proto = config._to_proto() + assert proto.HasField("relative_start_time") + restored = McapImportConfig._from_proto(proto) + assert restored.relative_start_time == config.relative_start_time + + def test_from_proto_round_trip(self): + config = self._config() + restored = McapImportConfig._from_proto(config._to_proto()) + assert restored.asset_name == config.asset_name + assert restored.run_name == config.run_name + assert restored.metadata_records == config.metadata_records + assert restored.parse_error_policy == McapParseErrorPolicy.IGNORE_ERROR + assert restored.complex_types_import_mode == McapComplexTypesImportMode.STRING + assert len(restored.data) == 2 + assert restored.data[0].topic == "/imu/data" + assert restored.data[0].field_path == "orientation.x" + assert restored.data[0].name == "/imu/data.orientation.x" + assert restored.data[1].name == "battery_voltage" + assert restored.data[1].data_type == ChannelDataType.FLOAT + assert restored.data[1].units == "V" + assert restored.data[1].description == "pack voltage" + + def test_from_proto_unspecified_enums_fall_back_to_defaults(self): + """UNSPECIFIED proto values mean FAIL_ON_ERROR and BOTH on the server.""" + from sift.data_imports.v2.data_imports_pb2 import McapConfig as McapConfigProto + + restored = McapImportConfig._from_proto(McapConfigProto(asset_name="a")) + assert restored.parse_error_policy == McapParseErrorPolicy.FAIL_ON_ERROR + assert restored.complex_types_import_mode == McapComplexTypesImportMode.BOTH + + def test_run_id_takes_precedence(self): + proto = McapImportConfig(asset_name="a", run_name="ignored", run_id="run_123")._to_proto() + assert proto.run_id == "run_123" + + def test_name_defaults_to_channel(self): + col = McapDataColumn( + topic="/imu/data", field_path="orientation.x", data_type=ChannelDataType.DOUBLE + ) + assert col.default_channel_name == "/imu/data.orientation.x" + assert col.name == "/imu/data.orientation.x" + + def test_explicit_name_overrides_channel(self): + col = McapDataColumn( + topic="/battery", + field_path="voltage", + name="battery_voltage", + data_type=ChannelDataType.FLOAT, + ) + assert col.name == "battery_voltage" + + def test_getitem(self): + col = self._config()["battery_voltage"] + assert col.field_path == "voltage" + + def test_getitem_not_found(self): + with pytest.raises(KeyError, match="nonexistent"): + self._config()["nonexistent"] + + +class TestImportFromPathClearsDetectedChannels: + """Auto-detected ULog and MCAP configs import with an empty channel list + so the server imports every channel instead of strictly filtering on a + list that client detection may have misread. + """ + + async def _import(self, tmp_path, filename, detected): + path = tmp_path / filename + path.write_bytes(b"") + + api = DataImportAPIAsync(MagicMock()) + api.detect_config = AsyncMock(return_value=detected) + api._low_level_client = MagicMock() + api._low_level_client.create_from_upload = AsyncMock(return_value=("import_1", "url")) + api.client.async_.jobs.get = AsyncMock(return_value="job") + + return await api.import_from_path(path, asset="my_asset", show_progress=False) + + @pytest.mark.asyncio + async def test_mcap_data_cleared(self, tmp_path, monkeypatch): + monkeypatch.setattr( + "sift_client.resources.data_imports.upload_file", lambda *a, **k: {"jobId": "j1"} + ) + detected = McapImportConfig( + asset_name="", + data=[McapDataColumn(topic="/imu", field_path="x", data_type=ChannelDataType.DOUBLE)], + ) + + job = await self._import(tmp_path, "log.mcap", detected) + + assert job == "job" + assert detected.data == [] + assert detected.asset_name == "my_asset" + + @pytest.mark.asyncio + async def test_ulog_data_cleared(self, tmp_path, monkeypatch): + monkeypatch.setattr( + "sift_client.resources.data_imports.upload_file", lambda *a, **k: {"jobId": "j1"} + ) + detected = UlogImportConfig( + asset_name="", + data=[ + UlogDataColumn( + message_name="sensor_accel", field_name="x", data_type=ChannelDataType.FLOAT + ) + ], + ) + + job = await self._import(tmp_path, "log.ulg", detected) + + assert job == "job" + assert detected.data == [] + + class TestCsvToProto: def test_to_proto(self, csv_config): proto = csv_config._to_proto() @@ -551,6 +728,9 @@ def test_known_extension_uses_map(self): def test_ulog_extension_uses_map(self): assert _resolve_data_type_key(".ulg", None) == DataTypeKey.ULOG + def test_mcap_extension_uses_map(self): + assert _resolve_data_type_key(".mcap", None) == DataTypeKey.MCAP + def test_explicit_data_type_overrides_extension(self): result = _resolve_data_type_key(".csv", DataTypeKey.TDMS) assert result == DataTypeKey.TDMS diff --git a/python/lib/sift_client/resources/data_imports.py b/python/lib/sift_client/resources/data_imports.py index c2a19c097..97dfe6d40 100644 --- a/python/lib/sift_client/resources/data_imports.py +++ b/python/lib/sift_client/resources/data_imports.py @@ -16,6 +16,7 @@ DataTypeKey, Hdf5ImportConfig, ImportConfig, + McapImportConfig, ParquetFlatDatasetImportConfig, ParquetSingleChannelPerRowImportConfig, ParquetTimeColumn, @@ -67,7 +68,7 @@ async def import_from_path( completion before proceeding. When ``config`` is omitted the file format is auto-detected via - ``detect_config`` (CSV, Parquet, HDF5, TDMS, and ULog). + ``detect_config`` (CSV, Parquet, HDF5, TDMS, ULog, and MCAP). 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 @@ -106,16 +107,17 @@ async def import_from_path( 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`` (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. + and MCAP 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 for CSV, Parquet, HDF5, and TDMS. - Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the - detected format if available, otherwise + Ignored for ULog and MCAP. 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. @@ -142,7 +144,7 @@ async def import_from_path( data_type=data_type, time_format=time_format, ) - if isinstance(config, UlogImportConfig): + if isinstance(config, (UlogImportConfig, McapImportConfig)): # 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 @@ -220,9 +222,10 @@ async def detect_config( 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. + endpoint; TDMS, HDF5, ULog, and MCAP are detected locally on the + client. - CSV, Parquet, HDF5, TDMS, and ULog files are supported for + CSV, Parquet, HDF5, TDMS, ULog, and MCAP files are supported for auto-detection. For CSV files, the server scans the first two rows for an optional @@ -249,6 +252,15 @@ async def detect_config( to exactly those channels; the import fails if a listed channel is not in the file. Clear ``data`` to import every channel. + For MCAP files, ``data`` lists the channels of each supported topic's + flattened fields, without decoding messages. A variable-cardinality + field expands to two channels: Arrow IPC bytes under the base name and + a JSON string under ``.json``. The same non-empty ``data`` + semantics as ULog apply. Topics the importer does not support are + skipped with a warning; importing such a file fails under the default + parse error policy, so set ``McapParseErrorPolicy.IGNORE_ERROR`` to + import the rest. + For file types with multiple supported layouts (Parquet, HDF5), ``data_type`` must be specified explicitly. @@ -258,8 +270,8 @@ async def detect_config( multiple supported layouts (Parquet, HDF5) where the file extension alone is ambiguous. 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 + Ignored for ULog and MCAP. When omitted, CSV, Parquet, and + HDF5 use the detected format if available, otherwise ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its detected/default time handling. @@ -281,7 +293,7 @@ async def detect_config( if time_format is not None: _apply_time_format(config, time_format) elif ( - not isinstance(config, (TdmsImportConfig, UlogImportConfig)) + not isinstance(config, (TdmsImportConfig, UlogImportConfig, McapImportConfig)) and _get_time_format(config) is None ): _apply_time_format(config, TimeFormat.ABSOLUTE_UNIX_NANOSECONDS) @@ -323,6 +335,15 @@ async def _detect_config_for_type( "Install it via `pip install sift-stack-py[ulog]`." ) from e return await run_sync_function(lambda: detect_ulog_config(path)) + if data_type_key == DataTypeKey.MCAP: + try: + from sift_client._internal.util.mcap import detect_mcap_config + except ImportError as e: + raise RuntimeError( + "mcap and mcap-ros2-support are required for MCAP import. " + "Install them via `pip install sift-stack-py[mcap]`." + ) from e + return await run_sync_function(lambda: detect_mcap_config(path)) is_parquet = data_type_key in ( DataTypeKey.PARQUET_FLATDATASET, @@ -360,7 +381,7 @@ def _read_sample() -> bytes: raise ValueError( f"No supported configuration detected for '{path.name}'. " - "Only CSV, Parquet, HDF5, TDMS, and ULog are supported by auto-detection." + "Only CSV, Parquet, HDF5, TDMS, ULog, and MCAP are supported by auto-detection." ) @@ -368,7 +389,8 @@ 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``. ULog has no configurable time format. + HDF5 store it on ``time_format``. ULog and MCAP have 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 e92fbd35d..ed02c7feb 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -715,9 +715,10 @@ class DataImportAPI: 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. + endpoint; TDMS, HDF5, ULog, and MCAP are detected locally on the + client. - CSV, Parquet, HDF5, TDMS, and ULog files are supported for + CSV, Parquet, HDF5, TDMS, ULog, and MCAP files are supported for auto-detection. For CSV files, the server scans the first two rows for an optional @@ -744,6 +745,15 @@ class DataImportAPI: to exactly those channels; the import fails if a listed channel is not in the file. Clear ``data`` to import every channel. + For MCAP files, ``data`` lists the channels of each supported topic's + flattened fields, without decoding messages. A variable-cardinality + field expands to two channels: Arrow IPC bytes under the base name and + a JSON string under ``.json``. The same non-empty ``data`` + semantics as ULog apply. Topics the importer does not support are + skipped with a warning; importing such a file fails under the default + parse error policy, so set ``McapParseErrorPolicy.IGNORE_ERROR`` to + import the rest. + For file types with multiple supported layouts (Parquet, HDF5), ``data_type`` must be specified explicitly. @@ -753,8 +763,8 @@ class DataImportAPI: multiple supported layouts (Parquet, HDF5) where the file extension alone is ambiguous. 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 + Ignored for ULog and MCAP. When omitted, CSV, Parquet, and + HDF5 use the detected format if available, otherwise ``TimeFormat.ABSOLUTE_UNIX_NANOSECONDS``. TDMS keeps its detected/default time handling. @@ -809,7 +819,7 @@ class DataImportAPI: completion before proceeding. When ``config`` is omitted the file format is auto-detected via - ``detect_config`` (CSV, Parquet, HDF5, TDMS, and ULog). + ``detect_config`` (CSV, Parquet, HDF5, TDMS, ULog, and MCAP). 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 @@ -848,16 +858,17 @@ class DataImportAPI: 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`` (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. + and MCAP 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 for CSV, Parquet, HDF5, and TDMS. - Ignored for ULog. When omitted, CSV, Parquet, and HDF5 use the - detected format if available, otherwise + Ignored for ULog and MCAP. 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. diff --git a/python/lib/sift_client/sift_types/data_import.py b/python/lib/sift_client/sift_types/data_import.py index 870596b33..dac2a2ac9 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -11,10 +11,17 @@ from sift.data_imports.v2.data_imports_pb2 import ( DATA_TYPE_KEY_CSV, DATA_TYPE_KEY_HDF5, + DATA_TYPE_KEY_MCAP, DATA_TYPE_KEY_PARQUET_FLATDATASET, DATA_TYPE_KEY_PARQUET_SINGLE_CHANNEL_PER_ROW, DATA_TYPE_KEY_TDMS, DATA_TYPE_KEY_ULOG, + MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH, + MCAP_COMPLEX_TYPES_IMPORT_MODE_BYTES, + MCAP_COMPLEX_TYPES_IMPORT_MODE_IGNORE, + MCAP_COMPLEX_TYPES_IMPORT_MODE_STRING, + MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR, + MCAP_PARSE_ERROR_POLICY_IGNORE_ERROR, PARQUET_COMPLEX_TYPES_IMPORT_MODE_BOTH, PARQUET_COMPLEX_TYPES_IMPORT_MODE_BYTES, PARQUET_COMPLEX_TYPES_IMPORT_MODE_IGNORE, @@ -32,6 +39,9 @@ from sift.data_imports.v2.data_imports_pb2 import CsvTimeColumn as CsvTimeColumnProto from sift.data_imports.v2.data_imports_pb2 import Hdf5Config as Hdf5ConfigProto from sift.data_imports.v2.data_imports_pb2 import Hdf5DataConfig as Hdf5DataConfigProto +from sift.data_imports.v2.data_imports_pb2 import McapConfig as McapConfigProto +from sift.data_imports.v2.data_imports_pb2 import McapDataConfig as McapDataConfigProto +from sift.data_imports.v2.data_imports_pb2 import McapRos2Selector as McapRos2SelectorProto from sift.data_imports.v2.data_imports_pb2 import ParquetConfig as ParquetConfigProto from sift.data_imports.v2.data_imports_pb2 import ParquetDataColumn as ParquetDataColumnProto from sift.data_imports.v2.data_imports_pb2 import ( @@ -85,6 +95,7 @@ class DataTypeKey(Enum): HDF5_TWO_D = "hdf5_two_d" HDF5_COMPOUND = "hdf5_compound" ULOG = "ulog" + MCAP = "mcap" DATA_TYPE_KEY_TO_PROTO = { @@ -96,6 +107,7 @@ class DataTypeKey(Enum): DataTypeKey.HDF5_TWO_D: DATA_TYPE_KEY_HDF5, DataTypeKey.HDF5_COMPOUND: DATA_TYPE_KEY_HDF5, DataTypeKey.ULOG: DATA_TYPE_KEY_ULOG, + DataTypeKey.MCAP: DATA_TYPE_KEY_MCAP, } @@ -103,6 +115,7 @@ class DataTypeKey(Enum): ".csv": DataTypeKey.CSV, ".tdms": DataTypeKey.TDMS, ".ulg": DataTypeKey.ULOG, + ".mcap": DataTypeKey.MCAP, } @@ -1008,6 +1021,176 @@ def _from_proto(cls, proto: UlogConfigProto) -> UlogImportConfig: ) +class McapParseErrorPolicy(Enum): + """Controls how MCAP import handles recoverable parse errors. + + Recoverable errors include truncated or undecodable records and + unsupported topics. The policy applies when the file is imported, not + during ``detect_config``. + """ + + FAIL_ON_ERROR = MCAP_PARSE_ERROR_POLICY_FAIL_ON_ERROR + """Fail the import on any recoverable parse error.""" + + IGNORE_ERROR = MCAP_PARSE_ERROR_POLICY_IGNORE_ERROR + """Import what decoded. Skipped topics and records surface as warnings.""" + + +class McapComplexTypesImportMode(Enum): + """Controls how variable-cardinality MCAP fields (dynamic and bounded + arrays) are imported. + + Under ``BOTH``, each such field imports as two channels: Arrow IPC bytes + under the field's base name and a JSON string under ``.json``. + """ + + IGNORE = MCAP_COMPLEX_TYPES_IMPORT_MODE_IGNORE + BOTH = MCAP_COMPLEX_TYPES_IMPORT_MODE_BOTH + STRING = MCAP_COMPLEX_TYPES_IMPORT_MODE_STRING + BYTES = MCAP_COMPLEX_TYPES_IMPORT_MODE_BYTES + + +class McapDataColumn(DataColumnBase): + """A single MCAP channel selection. + + Channels are selected by topic and flattened field path, as returned by + ``detect_config``. A variable-cardinality field is selected whole by its + base path and imports per its ``data_type``: ``BYTES`` (Arrow IPC) or + ``STRING`` (JSON), which ``complex_types_import_mode`` must allow. + + Attributes: + topic: The topic the channel comes from (e.g. ``"/imu/data"``). + field_path: The dot-delimited field path within the decoded message + (e.g. ``"orientation.x"``, ``"orientation_covariance[0]"``). + name: Sift channel name to create. Defaults to ``default_channel_name``, + e.g. ``"/imu/data.orientation.x"``. + """ + + topic: str + field_path: str + name: str = "" + + @property + def default_channel_name(self) -> str: + """The default Sift channel name for this selection, + ``.`` (e.g. ``"/imu/data.orientation.x"``). + """ + return f"{self.topic}.{self.field_path}" + + @model_validator(mode="after") + def _apply_default_name(self) -> McapDataColumn: + if not self.name: + self.name = self.default_channel_name + return self + + +class McapImportConfig(ImportConfigBase): + """Configuration for importing an MCAP (``.mcap``) file. + + MCAP 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: Log-start UTC, only for logs on a non-Unix epoch. + When set, ``log_time`` is reinterpreted as elapsed nanoseconds from + this start. + metadata_records: Metadata records to import as run metadata. Every key + of each named record is stored as ``.``. Empty + imports none. + parse_error_policy: How to handle recoverable parse errors. Defaults to + failing the import. + complex_types_import_mode: How to import variable-cardinality fields. + Defaults to importing them as both Arrow IPC bytes and JSON strings. + """ + + data: list[McapDataColumn] = [] + relative_start_time: datetime | None = None + metadata_records: list[str] = [] + parse_error_policy: McapParseErrorPolicy = McapParseErrorPolicy.FAIL_ON_ERROR + complex_types_import_mode: McapComplexTypesImportMode = McapComplexTypesImportMode.BOTH + + def __getitem__(self, name: str) -> McapDataColumn: + """Look up a configured MCAP channel by Sift channel name. + + Example:: + + config["/imu/data.orientation.x"].data_type = ChannelDataType.FLOAT + """ + for dc in self.data: + if dc.name == name: + return dc + raise KeyError(f"No data column named '{name}'") + + def _to_proto(self) -> McapConfigProto: + proto = McapConfigProto( + asset_name=self.asset_name, + run_name=self.run_name or "", + run_id=self.run_id or "", + metadata_records=self.metadata_records, + parse_error_policy=self.parse_error_policy.value, + complex_types_import_mode=self.complex_types_import_mode.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( + McapDataConfigProto( + topic=dc.topic, + ros2=McapRos2SelectorProto(field_path=dc.field_path), + 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: McapConfigProto) -> McapImportConfig: + """Create from a proto McapConfig (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 = McapParseErrorPolicy.FAIL_ON_ERROR + if proto.parse_error_policy == MCAP_PARSE_ERROR_POLICY_IGNORE_ERROR: + parse_error_policy = McapParseErrorPolicy.IGNORE_ERROR + + mode = proto.complex_types_import_mode + data = [ + McapDataColumn( + topic=d.topic, + field_path=d.ros2.field_path, + 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, + metadata_records=list(proto.metadata_records), + parse_error_policy=parse_error_policy, + complex_types_import_mode=McapComplexTypesImportMode(mode) + if mode + else McapComplexTypesImportMode.BOTH, + ) + + ImportConfig = Union[ CsvImportConfig, ParquetFlatDatasetImportConfig, @@ -1015,4 +1198,5 @@ def _from_proto(cls, proto: UlogConfigProto) -> UlogImportConfig: TdmsImportConfig, Hdf5ImportConfig, UlogImportConfig, + McapImportConfig, ] diff --git a/python/pyproject.toml b/python/pyproject.toml index 4aa459cd1..81dc5a893 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -63,6 +63,8 @@ import-test-result-log = "sift_client.scripts.import_test_result_log:main" all = [ 'cffi~=1.14', 'h5py~=3.11', + 'mcap-ros2-support~=0.5.7', + 'mcap~=1.4', 'npTDMS~=1.9', 'polars~=1.8', 'pyOpenSSL<24.0.0', @@ -98,6 +100,8 @@ dev-all = [ 'cffi~=1.14', 'grpcio-testing~=1.13', 'h5py~=3.11', + 'mcap-ros2-support~=0.5.7', + 'mcap~=1.4', 'mypy==1.10.0', 'npTDMS~=1.9', 'pdoc==14.5.0', @@ -149,6 +153,8 @@ docs-build = [ "griffe-pydantic==1.3.1 ; python_version >= '3.10'", 'grpcio-testing~=1.13', 'h5py~=3.11', + 'mcap-ros2-support~=0.5.7', + 'mcap~=1.4', "mike==2.1.3 ; python_version >= '3.10'", "mkdocs-api-autonav==0.4.0 ; python_version >= '3.10'", "mkdocs-include-markdown-plugin==7.1.6 ; python_version >= '3.10'", @@ -179,6 +185,8 @@ docs-build = [ ] file-imports = [ 'h5py~=3.11', + 'mcap-ros2-support~=0.5.7', + 'mcap~=1.4', 'npTDMS~=1.9', 'polars~=1.8', 'pyulog~=1.2.2', @@ -188,6 +196,10 @@ hdf5 = [ 'h5py~=3.11', 'polars~=1.8', ] +mcap = [ + 'mcap-ros2-support~=0.5.7', + 'mcap~=1.4', +] openssl = [ 'cffi~=1.14', 'pyOpenSSL<24.0.0', @@ -251,6 +263,7 @@ rosbags = ["rosbags~=0.0 ; python_full_version >= '3.8.2'"] sift-stream = ["sift-stream-bindings==0.5.0"] 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"] +mcap = ["mcap~=1.4", "mcap-ros2-support~=0.5.7"] data-review = ["pyarrow>=17.0.0"] [tool.sift.extras.combine] @@ -261,7 +274,7 @@ dev = ["development"] sift-stream-bindings = ["sift-stream"] # combinations -file-imports = ["tdms", "rosbags", "hdf5", "ulog"] +file-imports = ["tdms", "rosbags", "hdf5", "ulog", "mcap"] all = ["openssl", "sift-stream", "file-imports", "data-review"] dev-all = ["development", "all", "build"] @@ -394,6 +407,13 @@ module = "alive_progress" follow_imports = "skip" ignore_errors = true +# mcap's writer does `from .__init__ import __version__`, which makes mypy +# discover mcap/__init__.py under two module names and abort. Only the MCAP +# detection tests import the writer. +[[tool.mypy.overrides]] +module = "mcap.writer" +follow_imports = "skip" + [tool.setuptools.packages.find] where = ["lib"] exclude = ["sift_client._tests", "sift_client._tests.*"] diff --git a/python/uv.lock b/python/uv.lock index 2a91f3bb5..b8a7b1bdc 100644 --- a/python/uv.lock +++ b/python/uv.lock @@ -1480,6 +1480,7 @@ version = "4.3.3" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version >= '3.8.2' and python_full_version < '3.9'", + "python_full_version < '3.8.2'", ] sdist = { url = "https://files.pythonhosted.org/packages/a4/31/ec1259ca8ad11568abaf090a7da719616ca96b60d097ccc5799cd0ff599c/lz4-4.3.3.tar.gz", hash = "sha256:01fe674ef2889dbb9899d8a67361e0c4a2c833af5aeb37dd505727cf5d2a131e", size = 171509, upload-time = "2024-01-01T23:03:13.535Z" } wheels = [ @@ -1795,6 +1796,33 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/41/09/5b161152e2d90f7b87f781c2e1267494aef9c32498df793f73ad0a0a494a/matplotlib_inline-0.2.2-py3-none-any.whl", hash = "sha256:3c821cf1c209f59fb2d2d64abbf5b23b67bcb2210d663f9918dd851c6da1fcf6", size = 9534, upload-time = "2026-05-08T17:33:32.055Z" }, ] +[[package]] +name = "mcap" +version = "1.4.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "lz4", version = "4.3.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "lz4", version = "4.4.5", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, + { name = "zstandard", version = "0.23.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, + { name = "zstandard", version = "0.25.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.9'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/f8/d7/0f17e59733a71bd4d5b38afc8484531c2cd7648b08c79863ee95fc42b002/mcap-1.4.0.tar.gz", hash = "sha256:0528e2f86a61bfec73779e0628e6cf27af83d01d89e20b27d5ec9f0b556a63ac", size = 22155, upload-time = "2026-06-18T21:50:07.965Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/48/14/7e0b2a74b67e16e5f40ab78cc3e5aa4e7bdd55e0aec963d573edc07a15cd/mcap-1.4.0-py3-none-any.whl", hash = "sha256:0b48b1cc951b8d5aabd2599e60d410bae4f1be1819094f54117b7cbf6b3ee2e9", size = 20826, upload-time = "2026-06-18T21:50:06.704Z" }, +] + +[[package]] +name = "mcap-ros2-support" +version = "0.5.7" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mcap" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/84/0d/f6da01d8e4b73861dba17fd2ebd9a78d2fd4fa582ad348cf18d16a79a7e1/mcap_ros2_support-0.5.7.tar.gz", hash = "sha256:8ddb67e452a6e2963664e29bc8868e61d7011b72feb2b62832ed444e27ca3ab8", size = 23295, upload-time = "2025-12-24T21:25:37.614Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/92/c5/8e8636099c180031436a011908a22216c25e6cb7790ba748212e203317c4/mcap_ros2_support-0.5.7-py3-none-any.whl", hash = "sha256:349f0e0f7af8ebeb516003e801ea5c5de5b4b65359ce4a17a8d98b25a9cea260", size = 22086, upload-time = "2025-12-24T21:25:36.425Z" }, +] + [[package]] name = "mdit-py-plugins" version = "0.6.1" @@ -4486,6 +4514,8 @@ all = [ { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "mcap" }, + { name = "mcap-ros2-support" }, { name = "nptdms" }, { 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.*'" }, @@ -4532,6 +4562,8 @@ dev-all = [ { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "mcap" }, + { name = "mcap-ros2-support" }, { name = "mypy" }, { name = "nptdms" }, { name = "pdoc" }, @@ -4592,6 +4624,8 @@ docs-build = [ { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "mcap" }, + { name = "mcap-ros2-support" }, { name = "mike", marker = "python_full_version >= '3.10'" }, { name = "mkdocs", marker = "python_full_version >= '3.10'" }, { name = "mkdocs-api-autonav", marker = "python_full_version >= '3.10'" }, @@ -4629,6 +4663,8 @@ file-imports = [ { name = "h5py", version = "3.11.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.9'" }, { name = "h5py", version = "3.14.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.9.*'" }, { name = "h5py", version = "3.16.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.10'" }, + { name = "mcap" }, + { name = "mcap-ros2-support" }, { name = "nptdms" }, { 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.*'" }, @@ -4645,6 +4681,10 @@ hdf5 = [ { 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'" }, ] +mcap = [ + { name = "mcap" }, + { name = "mcap-ros2-support" }, +] openssl = [ { name = "cffi" }, { name = "pyopenssl" }, @@ -4693,6 +4733,16 @@ requires-dist = [ { name = "h5py", marker = "extra == 'docs-build'", specifier = "~=3.11" }, { name = "h5py", marker = "extra == 'file-imports'", specifier = "~=3.11" }, { name = "h5py", marker = "extra == 'hdf5'", specifier = "~=3.11" }, + { name = "mcap", marker = "extra == 'all'", specifier = "~=1.4" }, + { name = "mcap", marker = "extra == 'dev-all'", specifier = "~=1.4" }, + { name = "mcap", marker = "extra == 'docs-build'", specifier = "~=1.4" }, + { name = "mcap", marker = "extra == 'file-imports'", specifier = "~=1.4" }, + { name = "mcap", marker = "extra == 'mcap'", specifier = "~=1.4" }, + { name = "mcap-ros2-support", marker = "extra == 'all'", specifier = "~=0.5.7" }, + { name = "mcap-ros2-support", marker = "extra == 'dev-all'", specifier = "~=0.5.7" }, + { name = "mcap-ros2-support", marker = "extra == 'docs-build'", specifier = "~=0.5.7" }, + { name = "mcap-ros2-support", marker = "extra == 'file-imports'", specifier = "~=0.5.7" }, + { name = "mcap-ros2-support", marker = "extra == 'mcap'", specifier = "~=0.5.7" }, { name = "mike", marker = "python_full_version >= '3.10' and extra == 'docs'", specifier = "==2.1.3" }, { name = "mike", marker = "python_full_version >= '3.10' and extra == 'docs-build'", specifier = "==2.1.3" }, { name = "mkdocs", marker = "python_full_version >= '3.10' and extra == 'docs'", specifier = "==1.6.1" }, @@ -4807,7 +4857,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", "ulog"] +provides-extras = ["all", "build", "data-review", "dev", "dev-all", "development", "docs", "docs-build", "file-imports", "hdf5", "mcap", "openssl", "rosbags", "sift-stream", "sift-stream-bindings", "tdms", "ulog"] [[package]] name = "sift-stream-bindings" @@ -5433,9 +5483,10 @@ version = "0.23.0" source = { registry = "https://pypi.org/simple" } resolution-markers = [ "python_full_version >= '3.8.2' and python_full_version < '3.9'", + "python_full_version < '3.8.2'", ] dependencies = [ - { name = "cffi", marker = "python_full_version >= '3.8.2' and python_full_version < '3.9' and platform_python_implementation == 'PyPy'" }, + { name = "cffi", marker = "python_full_version < '3.9' and platform_python_implementation == 'PyPy'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ed/f6/2ac0287b442160a89d726b17a9184a4c615bb5237db763791a7fd16d9df1/zstandard-0.23.0.tar.gz", hash = "sha256:b2d8c62d08e7255f68f7a740bae85b3c9b8e5466baa9cbf7f57f1cde0ac6bc09", size = 681701, upload-time = "2024-07-15T00:18:06.141Z" } wheels = [ From 6a063335006eada6c427f081c19373a4a16b4881 Mon Sep 17 00:00:00 2001 From: Wei Lu Date: Thu, 27 Aug 2026 13:18:35 -0700 Subject: [PATCH 2/4] update detection --- python/CHANGELOG.md | 13 +- python/examples/data_import/mcap/main.py | 14 +- python/lib/sift_client/_internal/util/mcap.py | 171 ++++++++------- .../sift_client/_tests/_internal/test_mcap.py | 200 +++++++++++++++++- .../lib/sift_client/resources/data_imports.py | 21 +- .../resources/sync_stubs/__init__.pyi | 16 +- .../lib/sift_client/sift_types/data_import.py | 46 +++- 7 files changed, 367 insertions(+), 114 deletions(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 5d63a1247..b547ad863 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -15,7 +15,18 @@ This project adheres to [Semantic Versioning](http://semver.org/). job = client.data_import.import_from_path("recording.mcap", asset=my_asset) ``` -Importing without a config ingests every supported channel; a file containing unsupported topics fails under the default parse error policy unless `McapParseErrorPolicy.IGNORE_ERROR` is set. Call `detect_config` to enumerate the file's channels locally and edit the returned `McapImportConfig` (channel selection, names, types, metadata records, parse error policy, complex types mode) before importing. Detection requires the new `mcap` extra: `pip install sift-stack-py[mcap]`. +Importing without a config ingests every supported channel; a file with topics that cannot be decoded fails unless `McapParseErrorPolicy.IGNORE_ERROR` is set. Call `detect_config` to enumerate the file's channels locally, then edit the returned `McapImportConfig` (channel selection, names, types, metadata records, parse error policy) before importing. + +`data` lists one entry per field. Variable-cardinality fields (dynamic and bounded arrays) are typed `BYTES`, and `complex_types_import_mode` decides what each becomes: Arrow IPC bytes, a JSON string under `.json`, both (the default), or neither. As with Parquet, set it on the config: + +```python +from sift_client.sift_types.data_import import McapComplexTypesImportMode + +config = client.data_import.detect_config("recording.mcap") +config.complex_types_import_mode = McapComplexTypesImportMode.STRING +``` + +MCAP files are read locally to detect their channels, so both `detect_config` and importing without a config require the new `mcap` extra: `pip install sift-stack-py[mcap]`. Passing an `McapImportConfig` explicitly does not. ## [v0.20.0] - August 25, 2026 diff --git a/python/examples/data_import/mcap/main.py b/python/examples/data_import/mcap/main.py index 25ecfbe4c..05b3932c5 100644 --- a/python/examples/data_import/mcap/main.py +++ b/python/examples/data_import/mcap/main.py @@ -2,8 +2,9 @@ MCAP files are self-describing, so the import needs no column mapping: every channel of every supported topic (ros2msg schemas with cdr messages) is -imported. Channel names combine the topic and flattened field path, e.g. -"/imu/data.orientation.x". +imported. Channel names combine the topic and flattened field path; the +bundled sample_data.mcap produces "/imu/data.angular_velocity.x", +"/imu/data.linear_acceleration.x", "/battery.voltage", and their siblings. Swap sample_data.mcap for your own .mcap recording. """ @@ -45,11 +46,18 @@ # # from datetime import datetime, timezone # - # from sift_client.sift_types.data_import import McapParseErrorPolicy + # from sift_client.sift_types.data_import import ( + # McapComplexTypesImportMode, + # McapParseErrorPolicy, + # ) # # config = client.data_import.detect_config("sample_data.mcap") # print(config) # inspect every detected channel # + # # Example: import array fields only as JSON strings, instead of the + # # default of both JSON and Arrow IPC bytes + # config.complex_types_import_mode = McapComplexTypesImportMode.STRING + # # # Example: import only the IMU topic # config.data = [d for d in config.data if d.topic == "/imu/data"] # diff --git a/python/lib/sift_client/_internal/util/mcap.py b/python/lib/sift_client/_internal/util/mcap.py index 56326545f..95c1f26ae 100644 --- a/python/lib/sift_client/_internal/util/mcap.py +++ b/python/lib/sift_client/_internal/util/mcap.py @@ -1,11 +1,8 @@ """Detect channels in MCAP (``.mcap``) files. -Detection reads the file's schema and channel records without decoding -message payloads, then flattens each topic's ros2msg schema into importable -leaf fields, mirroring the server importer's rules. - -See ``detect_mcap_config`` for caveats on unsupported topics and empty -``data`` behavior. +Reads the file's schema and channel records without decoding message +payloads, then flattens each topic's ros2msg schema into leaf fields, one +channel per leaf. """ from __future__ import annotations @@ -25,7 +22,7 @@ MCAP_MAGIC = b"\x89MCAP0\r\n" -# The importer rejects chunk compressions outside this set. +# Chunk compressions Sift can read. SUPPORTED_COMPRESSIONS = frozenset(("", "zstd", "lz4")) # ROS 2 scalar types to Sift channel types. Narrow integers widen to 32-bit @@ -52,21 +49,17 @@ TIME_MESSAGE_TYPES = frozenset(("builtin_interfaces/Time", "builtin_interfaces/Duration")) ROS2_TIME_TYPE = "__time__" -# Suffix of the JSON expansion of a variable-cardinality field. -JSON_CHANNEL_SUFFIX = ".json" - # Guards malformed schemas with self-referential fixed nesting. MAX_FIELD_DEPTH = 32 class UnsupportedTopicError(Exception): - """The topic's schema cannot be decoded by the importer.""" + """The topic's schema cannot be decoded.""" class LeafField(NamedTuple): """A leaf field of a topic's message type. Scalar leaves carry one value - per message; complex leaves are variable-cardinality and expand per the - complex types import mode. + per message; complex leaves are variable-cardinality. """ field_path: str @@ -90,7 +83,7 @@ class TopicInfo(NamedTuple): def parse_schema_defs(schema: mcap_records.Schema): """Parse a ros2msg concatenated schema into (root msgdef, msgdefs by - name), using the same vendored parser the server importer uses.""" + name).""" msgdefs: dict = { "builtin_interfaces/Time": ros2_dynamic.TimeDefinition, "builtin_interfaces/Duration": ros2_dynamic.TimeDefinition, @@ -142,9 +135,9 @@ def _check_ftype_decodable( ) -> None: """Raise UnsupportedTopicError if the field type cannot be decoded. - Variable-cardinality fields do not expand into leaves, but the importer - still decodes every element, so wstring or unknown types anywhere in the - subtree make the whole topic unsupported. + Variable-cardinality fields do not expand into leaves, but every element + is still decoded on import, so a wstring or unknown type anywhere in the + subtree makes the whole topic unsupported. """ if ftype.is_primitive_type(): _check_primitive_supported(ftype.type, label) @@ -181,8 +174,8 @@ def walk(prefix: str, msgdef, depth: int) -> None: ftype = field.type path = f"{prefix}{field.name}" if _is_variable_array(ftype): - # The importer decodes every element even though the leaf is - # imported whole, so verify the element type decodes. + # The leaf imports whole, but its elements are still + # decoded, so check the element type. _check_ftype_decodable(msgdefs, ftype, path) leaves.append(LeafField(path, "complex", None)) continue @@ -211,23 +204,31 @@ def _read_schemas_and_channels( ) -> tuple[dict[int, mcap_records.Schema], list[mcap_records.Channel], list[str]]: """Read schema and channel records without decoding message payloads. - Mirrors the server importer's scan: a top-level pass validates chunk - compression and collects the records of unchunked files, the summary - section supplies the records of chunked files, and a file without a - readable summary (e.g. truncated) is scanned through its decompressed - chunks instead, keeping what parsed with a warning. + A file can hold these records in three places, so this takes up to three + passes: + + 1. At the top level, where unchunked files keep them. + 2. In the summary section, which usually repeats what the chunks hold. + 3. Inside the chunks themselves, when the summary is missing or partial. + + A file that stops parsing partway keeps what was read, with a warning. """ parse_warnings: list[str] = [] schemas: dict[int, mcap_records.Schema] = {} channels: list[mcap_records.Channel] = [] seen_channel_ids: set[int] = set() + saw_chunks = False + saw_message = False + attachment_count = 0 + statistics: mcap_records.Statistics | None = None def add_channel(channel: mcap_records.Channel) -> None: if channel.id not in seen_channel_ids: seen_channel_ids.add(channel.id) channels.append(channel) - def scan(stream: StreamReader) -> None: + def scan(stream: StreamReader, top_level: bool) -> None: + nonlocal saw_chunks, saw_message, attachment_count, statistics records = iter(stream.records) while True: try: @@ -246,9 +247,9 @@ def scan(stream: StreamReader) -> None: parse_warnings.append(message) return if isinstance(record, mcap_records.Chunk): - # The importer rejects unsupported compression regardless of - # parse_error_policy; the mcap reader would silently treat it - # as uncompressed. + saw_chunks = True + # An unknown compression would be read as uncompressed + # garbage, so stop here instead. if record.compression not in SUPPORTED_COMPRESSIONS: raise ValueError( f"unsupported chunk compression '{record.compression}'; " @@ -258,19 +259,25 @@ def scan(stream: StreamReader) -> None: schemas[record.id] = record elif isinstance(record, mcap_records.Channel): add_channel(record) + elif isinstance(record, mcap_records.Message): + saw_message = True + elif top_level and isinstance(record, mcap_records.Attachment): + # Counted in this pass only, so a later pass cannot + # double the count. + attachment_count += 1 + elif top_level and isinstance(record, mcap_records.Statistics): + statistics = record with open(path, "rb") as file: if file.read(len(MCAP_MAGIC)) != MCAP_MAGIC: raise ValueError(f"'{path.name}' is not an MCAP file (bad magic bytes)") file.seek(0) - # Top-level pass: chunks stay unopened, so this validates compression - # cheaply and picks up the records of unchunked files. - scan(StreamReader(file, emit_chunks=True)) + # Chunks stay unopened here, so this pass is cheap: it checks + # compression and picks up the records of unchunked files. + scan(StreamReader(file, emit_chunks=True), top_level=True) - # Chunked files carry their records inside chunks; the summary section - # repeats them. Without a readable summary (e.g. a truncated file), - # scan through the decompressed chunks instead. + # The summary section usually repeats what the chunks hold. file.seek(0) try: summary = make_reader(file).get_summary() @@ -280,9 +287,19 @@ def scan(stream: StreamReader) -> None: schemas.update(summary.schemas) for channel in sorted(summary.channels.values(), key=lambda c: c.id): add_channel(channel) - else: + + # Those repeats are optional, so what we have may be incomplete. A + # Statistics record covering at least one message means the summary is + # trustworthy; without one, read the chunks. + have_range = statistics is not None and statistics.message_count > 0 + if summary is None or (not have_range and (saw_chunks or not saw_message)): file.seek(0) - scan(StreamReader(file, emit_chunks=False)) + scan(StreamReader(file, emit_chunks=False), top_level=False) + + if attachment_count: + parse_warnings.append( + f"the file has {attachment_count} attachment(s); attachments are not imported" + ) return schemas, channels, parse_warnings @@ -296,8 +313,7 @@ def detect_mcap_topics( Same-topic channels merge only when their schemas and message encodings match. Distinct topics colliding case-insensitively keep the first. - Unsupported topics are skipped with a warning; the import itself gates - them on ``parse_error_policy``. + Topics that cannot be decoded are skipped with a warning. """ channels_by_topic: defaultdict[str, list[mcap_records.Channel]] = defaultdict(list) for channel in channels: @@ -310,7 +326,9 @@ def detect_mcap_topics( first = kept_by_lower.setdefault(topic.lower(), topic) if first != topic: parse_warnings.append( - f"topic '{topic}' collides with topic '{first}' by case only; kept the first" + f"topic '{topic}' collides with topic '{first}' by case only; kept the " + "first. Set McapParseErrorPolicy.IGNORE_ERROR to import it and skip the " + "rest, otherwise the import fails" ) topics: list[TopicInfo] = [] @@ -357,41 +375,36 @@ def detect_mcap_topics( def detect_mcap_fields(topics: list[TopicInfo]) -> list[McapDataColumn]: - """Return importable channels as ``McapDataColumn``s with default names - and data types. + """Return one ``McapDataColumn`` per leaf field, named ``.``. - Scalar leaves become one channel named ``.``. Complex - leaves expand like the default complex types import mode (``BOTH``): Arrow - IPC bytes under the base name and a JSON string under ``.json``. + Variable-cardinality fields get ``BYTES``. Whether one of those imports as + bytes, as a JSON string, as both, or not at all is decided by the config's + ``complex_types_import_mode`` when the config is sent. """ channels: list[McapDataColumn] = [] # Sift channel names are unique per asset and compare case-insensitively. - taken_names: dict[str, str] = {} + # Values are the (name, topic, field path) that first claimed the key. + taken_names: dict[str, tuple[str, str, str]] = {} for topic in topics: for leaf in topic.leaves: - base_name = f"{topic.topic}.{leaf.field_path}" - if leaf.kind == "scalar": - expansions = [(base_name, leaf.sift_type())] - else: - expansions = [ - (base_name, ChannelDataType.BYTES), - (base_name + JSON_CHANNEL_SUFFIX, ChannelDataType.STRING), - ] - for name, data_type in expansions: - existing = taken_names.get(name.lower()) - if existing is not None: - raise ValueError( - f"the generated channel name '{name}' conflicts with channel '{existing}'" - ) - taken_names[name.lower()] = name - channels.append( - McapDataColumn( - topic=topic.topic, - field_path=leaf.field_path, - name=name, - data_type=data_type, - ) + name = f"{topic.topic}.{leaf.field_path}" + data_type = ChannelDataType.BYTES if leaf.kind == "complex" else leaf.sift_type() + existing = taken_names.get(name.lower()) + if existing is not None: + raise ValueError( + f"two channels are both named '{name}': topic '{topic.topic}' field " + f"'{leaf.field_path}' and topic '{existing[1]}' field '{existing[2]}'. " + "Build an McapImportConfig by hand to give them distinct names." + ) + taken_names[name.lower()] = (name, topic.topic, leaf.field_path) + channels.append( + McapDataColumn( + topic=topic.topic, + field_path=leaf.field_path, + name=name, + data_type=data_type, ) + ) return channels @@ -400,31 +413,29 @@ def detect_mcap_config(file_path: str | Path, asset_name: str = "") -> McapImpor Channels come from the file's schema and channel records; message payloads are not read, so a topic is listed even when it logged no messages. Topics - the importer does not support (non-cdr message encodings, non-ros2msg - schemas, undecodable schemas) are skipped with a warning; importing such a - file fails under the default parse error policy, so set - ``McapParseErrorPolicy.IGNORE_ERROR`` to import the rest. + that cannot be decoded (non-cdr message encodings, non-ros2msg schemas, + undecodable schemas) are skipped with a warning. Importing such a file + fails unless ``parse_error_policy`` is ``McapParseErrorPolicy.IGNORE_ERROR``. Args: file_path: Path to the ``.mcap`` 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. + A config whose ``data`` lists one channel per leaf field, with default + Sift names and data types. Remove entries to skip channels, or edit + them before importing. Leaving ``data`` empty imports all channels + with the same defaults. Raises: ValueError: If the file is not MCAP, uses an unsupported chunk - compression, or two detected channels generate the same Sift - channel name. The importer rejects all three regardless of - ``parse_error_policy``. + compression, or two channels share a name. """ path = Path(file_path) schemas, channels, parse_warnings = _read_schemas_and_channels(path) topics = detect_mcap_topics(schemas, channels, parse_warnings) - data = detect_mcap_fields(topics) + # Emitted before the names are built so a name clash does not discard + # what the scan found. for message in parse_warnings: warnings.warn(f"'{path.name}': {message}", stacklevel=2) - return McapImportConfig(asset_name=asset_name, data=data) + return McapImportConfig(asset_name=asset_name, data=detect_mcap_fields(topics)) diff --git a/python/lib/sift_client/_tests/_internal/test_mcap.py b/python/lib/sift_client/_tests/_internal/test_mcap.py index 47b8306c3..a2c0d36d0 100644 --- a/python/lib/sift_client/_tests/_internal/test_mcap.py +++ b/python/lib/sift_client/_tests/_internal/test_mcap.py @@ -14,6 +14,7 @@ parse_schema_defs, ) from sift_client.sift_types.channel import ChannelDataType +from sift_client.sift_types.data_import import McapComplexTypesImportMode def _schema(data: str, schema_id: int = 1, name: str = "test_msgs/msg/Test") -> Schema: @@ -231,6 +232,9 @@ def test_case_colliding_topics_keep_first(self): topics = detect_mcap_topics(schemas, channels, warnings) assert [t.topic for t in topics] == ["/imu"] assert any("by case only" in w for w in warnings) + # The importer rejects the collision unless the policy is IGNORE_ERROR, + # so the warning has to say so. + assert any("IGNORE_ERROR" in w for w in warnings) def test_same_topic_channels_merge_when_identical(self): schemas = {1: _schema(IMU_SCHEMA)} @@ -270,14 +274,14 @@ def test_detects_channel_per_field(self, tmp_path): ("/imu/data", "temp", "/imu/data.temp", ChannelDataType.DOUBLE), } - def test_complex_field_expands_to_bytes_and_json(self, tmp_path): + def test_complex_field_is_one_bytes_channel(self, tmp_path): + # The mode turns this single entry into the channels that get imported. path = tmp_path / "log.mcap" _write_mcap(path, [("test_msgs/msg/Samples", "int32[] samples\n", "/samples")]) config = detect_mcap_config(path) assert [(d.field_path, d.name, d.data_type) for d in config.data] == [ ("samples", "/samples.samples", ChannelDataType.BYTES), - ("samples", "/samples.samples.json", ChannelDataType.STRING), ] def test_unsupported_topic_warns_and_keeps_supported(self, tmp_path): @@ -360,10 +364,94 @@ def test_rejects_unsupported_chunk_compression(self, tmp_path): with pytest.raises(ValueError, match="unsupported chunk compression"): detect_mcap_config(path) - def test_duplicate_generated_names_raise(self, tmp_path): - # Topic '/a' with variable array field 'b' expands to '/a.b.json', - # colliding with topic '/a.b' scalar field 'json'. The importer - # rejects the file the same way. + def test_duplicate_channel_names_raise(self, tmp_path): + # Topic '/a' field 'b.c' and topic '/a.b' field 'c' both name a + # channel '/a.b.c'. + path = tmp_path / "log.mcap" + _write_mcap( + path, + [ + ("pkg/msg/A", "pkg/B b\n" + "=" * 80 + "\nMSG: pkg/B\nint32 c\n", "/a"), + ("pkg/msg/C", "int32 c\n", "/a.b"), + ], + ) + with pytest.raises(ValueError, match="both named '/a.b.c'"): + detect_mcap_config(path) + + +class TestComplexTypesImportMode: + """A variable-cardinality field is one entry in ``data``; the mode decides + which channels it becomes when the config is sent. + """ + + SCHEMA = "int32 count\nint32[] samples\n" + + def _config(self, tmp_path, mode): + path = tmp_path / "log.mcap" + _write_mcap(path, [("test_msgs/msg/Samples", self.SCHEMA, "/samples")]) + config = detect_mcap_config(path) + config.complex_types_import_mode = mode + return config + + @pytest.mark.parametrize( + ("mode", "expected"), + [ + ( + McapComplexTypesImportMode.BOTH, + [ + ("/samples.count", ChannelDataType.INT_32), + ("/samples.samples", ChannelDataType.BYTES), + ("/samples.samples.json", ChannelDataType.STRING), + ], + ), + ( + McapComplexTypesImportMode.BYTES, + [ + ("/samples.count", ChannelDataType.INT_32), + ("/samples.samples", ChannelDataType.BYTES), + ], + ), + ( + McapComplexTypesImportMode.STRING, + [ + ("/samples.count", ChannelDataType.INT_32), + ("/samples.samples.json", ChannelDataType.STRING), + ], + ), + ( + McapComplexTypesImportMode.IGNORE, + [("/samples.count", ChannelDataType.INT_32)], + ), + ], + ) + def test_mode_decides_the_channels_sent(self, tmp_path, mode, expected): + proto = self._config(tmp_path, mode)._to_proto() + assert [ + (d.channel_config.name, ChannelDataType(d.channel_config.data_type)) for d in proto.data + ] == expected + + def test_both_channels_share_the_field_selector(self, tmp_path): + proto = self._config(tmp_path, McapComplexTypesImportMode.BOTH)._to_proto() + selectors = [(d.topic, d.ros2.field_path) for d in proto.data] + assert selectors[1] == selectors[2] == ("/samples", "samples") + + def test_default_is_both(self, tmp_path): + path = tmp_path / "log.mcap" + _write_mcap(path, [("test_msgs/msg/Samples", self.SCHEMA, "/samples")]) + config = detect_mcap_config(path) + assert config.complex_types_import_mode is McapComplexTypesImportMode.BOTH + assert len(config._to_proto().data) == 3 + + def test_ignoring_every_configured_channel_raises(self, tmp_path): + # An empty list would mean "import the whole file", so refuse instead. + config = self._config(tmp_path, McapComplexTypesImportMode.IGNORE) + config.data = [d for d in config.data if d.data_type == ChannelDataType.BYTES] + with pytest.raises(ValueError, match="nothing would be imported"): + config._to_proto() + + def test_generated_json_name_clash_raises(self, tmp_path): + # '/a' field 'b' generates '/a.b.json', which topic '/a.b' field + # 'json' already claims. path = tmp_path / "log.mcap" _write_mcap( path, @@ -372,5 +460,103 @@ def test_duplicate_generated_names_raise(self, tmp_path): ("pkg/msg/B", "int32 json\n", "/a.b"), ], ) - with pytest.raises(ValueError, match="conflicts with channel"): + config = detect_mcap_config(path) + assert {d.name for d in config.data} == {"/a.b", "/a.b.json"} + with pytest.raises(ValueError, match="would both be imported as '/a.b.json'"): + config._to_proto() + + def test_clash_disappears_under_bytes_mode(self, tmp_path): + path = tmp_path / "log.mcap" + _write_mcap( + path, + [ + ("pkg/msg/A", "int32[] b\n", "/a"), + ("pkg/msg/B", "int32 json\n", "/a.b"), + ], + ) + config = detect_mcap_config(path) + config.complex_types_import_mode = McapComplexTypesImportMode.BYTES + assert {d.channel_config.name for d in config._to_proto().data} == {"/a.b", "/a.b.json"} + + +class TestScanCompleteness: + def test_summary_without_statistics_falls_back_to_chunk_scan(self, tmp_path): + # A chunked file whose summary repeats neither schemas nor channels and + # carries no Statistics record: the summary yields nothing, so + # detection must read the records inside the chunks like the importer. + path = tmp_path / "log.mcap" + with open(path, "wb") as f: + writer = Writer(f, repeat_channels=False, repeat_schemas=False, use_statistics=False) + writer.start() + schema_id = writer.register_schema( + name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() + ) + channel_id = writer.register_channel( + topic="/imu", message_encoding="cdr", schema_id=schema_id + ) + for i in range(10): + writer.add_message( + channel_id=channel_id, + log_time=1_700_000_000_000_000_000 + i, + publish_time=1_700_000_000_000_000_000 + i, + data=b"\x00" * 32, + ) + writer.finish() + + config = detect_mcap_config(path) + assert {d.topic for d in config.data} == {"/imu"} + + def test_attachments_warn(self, tmp_path): + path = tmp_path / "log.mcap" + with open(path, "wb") as f: + writer = Writer(f) + writer.start() + schema_id = writer.register_schema( + name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() + ) + writer.register_channel(topic="/imu", message_encoding="cdr", schema_id=schema_id) + writer.add_attachment( + create_time=1, log_time=1, name="notes.txt", media_type="text/plain", data=b"hi" + ) + writer.finish() + + with pytest.warns(UserWarning, match="1 attachment"): + config = detect_mcap_config(path) + assert {d.topic for d in config.data} == {"/imu"} + + def test_scan_warnings_survive_a_name_clash(self, tmp_path): + # The clash raises, but what the scan already found must still reach + # the caller instead of being discarded with the exception. + path = tmp_path / "log.mcap" + nested = "pkg/B b\n" + "=" * 80 + "\nMSG: pkg/B\nint32 c\n" + with open(path, "wb") as f: + writer = Writer(f) + writer.start() + a = writer.register_schema(name="pkg/msg/A", encoding="ros2msg", data=nested.encode()) + writer.register_channel(topic="/a", message_encoding="cdr", schema_id=a) + b = writer.register_schema(name="pkg/msg/C", encoding="ros2msg", data=b"int32 c\n") + writer.register_channel(topic="/a.b", message_encoding="cdr", schema_id=b) + writer.add_attachment( + create_time=1, log_time=1, name="notes.txt", media_type="text/plain", data=b"hi" + ) + writer.finish() + + with pytest.warns(UserWarning, match="1 attachment"), pytest.raises( + ValueError, match="both named" + ): + detect_mcap_config(path) + + def test_clash_error_names_both_origins(self, tmp_path): + path = tmp_path / "log.mcap" + _write_mcap( + path, + [ + ("pkg/msg/A", "pkg/B b\n" + "=" * 80 + "\nMSG: pkg/B\nint32 c\n", "/a"), + ("pkg/msg/C", "int32 c\n", "/a.b"), + ], + ) + with pytest.raises(ValueError, match="both named") as excinfo: detect_mcap_config(path) + message = str(excinfo.value) + assert "topic '/a' field 'b.c'" in message + assert "topic '/a.b' field 'c'" in message diff --git a/python/lib/sift_client/resources/data_imports.py b/python/lib/sift_client/resources/data_imports.py index 97dfe6d40..216e71b1a 100644 --- a/python/lib/sift_client/resources/data_imports.py +++ b/python/lib/sift_client/resources/data_imports.py @@ -145,10 +145,7 @@ async def import_from_path( time_format=time_format, ) if isinstance(config, (UlogImportConfig, McapImportConfig)): - # 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. + # An empty channel list imports every channel config.data = [] if asset is not None: @@ -252,14 +249,14 @@ async def detect_config( to exactly those channels; the import fails if a listed channel is not in the file. Clear ``data`` to import every channel. - For MCAP files, ``data`` lists the channels of each supported topic's - flattened fields, without decoding messages. A variable-cardinality - field expands to two channels: Arrow IPC bytes under the base name and - a JSON string under ``.json``. The same non-empty ``data`` - semantics as ULog apply. Topics the importer does not support are - skipped with a warning; importing such a file fails under the default - parse error policy, so set ``McapParseErrorPolicy.IGNORE_ERROR`` to - import the rest. + For MCAP files, ``data`` lists one channel per flattened field of each + supported topic, without decoding messages. A variable-cardinality + field is one entry; ``complex_types_import_mode`` on the config decides + whether it imports as Arrow IPC bytes, a JSON string under + ``.json``, both, or neither. The same non-empty ``data`` + semantics as ULog apply. Topics that cannot be decoded are skipped with + a warning; importing such a file fails unless + ``McapParseErrorPolicy.IGNORE_ERROR`` is set. For file types with multiple supported layouts (Parquet, HDF5), ``data_type`` must be specified explicitly. diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index ed02c7feb..753217a62 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -745,14 +745,14 @@ class DataImportAPI: to exactly those channels; the import fails if a listed channel is not in the file. Clear ``data`` to import every channel. - For MCAP files, ``data`` lists the channels of each supported topic's - flattened fields, without decoding messages. A variable-cardinality - field expands to two channels: Arrow IPC bytes under the base name and - a JSON string under ``.json``. The same non-empty ``data`` - semantics as ULog apply. Topics the importer does not support are - skipped with a warning; importing such a file fails under the default - parse error policy, so set ``McapParseErrorPolicy.IGNORE_ERROR`` to - import the rest. + For MCAP files, ``data`` lists one channel per flattened field of each + supported topic, without decoding messages. A variable-cardinality + field is one entry; ``complex_types_import_mode`` on the config decides + whether it imports as Arrow IPC bytes, a JSON string under + ``.json``, both, or neither. The same non-empty ``data`` + semantics as ULog apply. Topics that cannot be decoded are skipped with + a warning; importing such a file fails unless + ``McapParseErrorPolicy.IGNORE_ERROR`` is set. 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 dac2a2ac9..fec25f2a8 100644 --- a/python/lib/sift_client/sift_types/data_import.py +++ b/python/lib/sift_client/sift_types/data_import.py @@ -1036,6 +1036,10 @@ class McapParseErrorPolicy(Enum): """Import what decoded. Skipped topics and records surface as warnings.""" +# Suffix given to the JSON channel of a variable-cardinality field. +MCAP_JSON_CHANNEL_SUFFIX = ".json" + + class McapComplexTypesImportMode(Enum): """Controls how variable-cardinality MCAP fields (dynamic and bounded arrays) are imported. @@ -1106,6 +1110,8 @@ class McapImportConfig(ImportConfigBase): failing the import. complex_types_import_mode: How to import variable-cardinality fields. Defaults to importing them as both Arrow IPC bytes and JSON strings. + ``data`` lists one entry per field; the mode decides which channels + that entry becomes, so it can be changed on a detected config. """ data: list[McapDataColumn] = [] @@ -1137,19 +1143,53 @@ def _to_proto(self) -> McapConfigProto: ) 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: + + mode = self.complex_types_import_mode + # Channel names are unique per asset and compare case-insensitively. + taken_names: dict[str, str] = {} + + def add(dc: McapDataColumn, name: str, data_type: ChannelDataType) -> None: + source = taken_names.get(name.lower()) + if source is not None: + raise ValueError( + f"channels '{source}' and '{dc.name}' would both be imported as " + f"'{name}'. Rename or remove one before importing." + ) + taken_names[name.lower()] = dc.name proto.data.append( McapDataConfigProto( topic=dc.topic, ros2=McapRos2SelectorProto(field_path=dc.field_path), channel_config=ChannelConfigProto( - name=dc.name, - data_type=dc.data_type.value, + name=name, + data_type=data_type.value, units=dc.units, description=dc.description, ), ) ) + + for dc in self.data: + # Only variable-cardinality fields can be BYTES, and the mode + # decides which channels they become. + if dc.data_type != ChannelDataType.BYTES: + add(dc, dc.name, dc.data_type) + continue + if mode is McapComplexTypesImportMode.IGNORE: + continue + if mode in (McapComplexTypesImportMode.BYTES, McapComplexTypesImportMode.BOTH): + add(dc, dc.name, ChannelDataType.BYTES) + if mode in (McapComplexTypesImportMode.STRING, McapComplexTypesImportMode.BOTH): + add(dc, dc.name + MCAP_JSON_CHANNEL_SUFFIX, ChannelDataType.STRING) + + if self.data and not proto.data: + # An empty list means "import everything", which is not what + # selecting channels and then dropping them all should do. + raise ValueError( + "complex_types_import_mode is IGNORE and every configured channel is " + "variable-cardinality, so nothing would be imported. Choose another mode " + "or clear 'data' to import the whole file." + ) return proto @classmethod From 613e6e49d98f874ab8f12947acf07c87aaa20c85 Mon Sep 17 00:00:00 2001 From: Wei Lu Date: Thu, 27 Aug 2026 13:28:30 -0700 Subject: [PATCH 3/4] update --- python/CHANGELOG.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index b547ad863..7c18008c6 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -15,18 +15,18 @@ This project adheres to [Semantic Versioning](http://semver.org/). job = client.data_import.import_from_path("recording.mcap", asset=my_asset) ``` -Importing without a config ingests every supported channel; a file with topics that cannot be decoded fails unless `McapParseErrorPolicy.IGNORE_ERROR` is set. Call `detect_config` to enumerate the file's channels locally, then edit the returned `McapImportConfig` (channel selection, names, types, metadata records, parse error policy) before importing. +Importing without a config ingests every supported channel. Topics that cannot be decoded fail the import unless `McapParseErrorPolicy.IGNORE_ERROR` is set. -`data` lists one entry per field. Variable-cardinality fields (dynamic and bounded arrays) are typed `BYTES`, and `complex_types_import_mode` decides what each becomes: Arrow IPC bytes, a JSON string under `.json`, both (the default), or neither. As with Parquet, set it on the config: +`detect_config` reads the file's channels locally without decoding messages, returning an `McapImportConfig` whose `data` holds one entry per field. Edit it to select, rename, or retype channels before importing. ```python -from sift_client.sift_types.data_import import McapComplexTypesImportMode - config = client.data_import.detect_config("recording.mcap") config.complex_types_import_mode = McapComplexTypesImportMode.STRING ``` -MCAP files are read locally to detect their channels, so both `detect_config` and importing without a config require the new `mcap` extra: `pip install sift-stack-py[mcap]`. Passing an `McapImportConfig` explicitly does not. +Variable-cardinality fields (dynamic and bounded arrays) are typed `BYTES`. As with Parquet, `complex_types_import_mode` on the config decides what each becomes: Arrow IPC bytes, a JSON string under `.json`, both (the default), or neither. + +Reading a file locally needs the new `mcap` extra (`pip install sift-stack-py[mcap]`), so both `detect_config` and importing without a config require it. ## [v0.20.0] - August 25, 2026 From e53838d641c30e39067284ea010a6fb2df432486 Mon Sep 17 00:00:00 2001 From: Wei Lu Date: Fri, 28 Aug 2026 12:13:27 -0700 Subject: [PATCH 4/4] check summary first --- python/lib/sift_client/_internal/util/mcap.py | 66 +++-- .../sift_client/_tests/_internal/test_mcap.py | 245 ++++++++---------- 2 files changed, 145 insertions(+), 166 deletions(-) diff --git a/python/lib/sift_client/_internal/util/mcap.py b/python/lib/sift_client/_internal/util/mcap.py index 95c1f26ae..ce20550d4 100644 --- a/python/lib/sift_client/_internal/util/mcap.py +++ b/python/lib/sift_client/_internal/util/mcap.py @@ -22,7 +22,9 @@ MCAP_MAGIC = b"\x89MCAP0\r\n" -# Chunk compressions Sift can read. +# Chunk compressions Sift can read. The reader skips a chunk it cannot +# decompress without complaining, hiding every channel inside it, so anything +# else has to be rejected explicitly. SUPPORTED_COMPRESSIONS = frozenset(("", "zstd", "lz4")) # ROS 2 scalar types to Sift channel types. Narrow integers widen to 32-bit @@ -204,12 +206,10 @@ def _read_schemas_and_channels( ) -> tuple[dict[int, mcap_records.Schema], list[mcap_records.Channel], list[str]]: """Read schema and channel records without decoding message payloads. - A file can hold these records in three places, so this takes up to three - passes: - - 1. At the top level, where unchunked files keep them. - 2. In the summary section, which usually repeats what the chunks hold. - 3. Inside the chunks themselves, when the summary is missing or partial. + A well-formed file repeats them in its summary section, a few kilobytes at + the end, and that is all we need. Otherwise they only exist in the data + section and the whole file has to be read: unchunked files keep them at the + top level, chunked files keep them inside the chunks. A file that stops parsing partway keeps what was read, with a warning. """ @@ -217,18 +217,15 @@ def _read_schemas_and_channels( schemas: dict[int, mcap_records.Schema] = {} channels: list[mcap_records.Channel] = [] seen_channel_ids: set[int] = set() - saw_chunks = False - saw_message = False attachment_count = 0 - statistics: mcap_records.Statistics | None = None def add_channel(channel: mcap_records.Channel) -> None: if channel.id not in seen_channel_ids: seen_channel_ids.add(channel.id) channels.append(channel) - def scan(stream: StreamReader, top_level: bool) -> None: - nonlocal saw_chunks, saw_message, attachment_count, statistics + def scan(stream: StreamReader) -> None: + nonlocal attachment_count records = iter(stream.records) while True: try: @@ -247,9 +244,6 @@ def scan(stream: StreamReader, top_level: bool) -> None: parse_warnings.append(message) return if isinstance(record, mcap_records.Chunk): - saw_chunks = True - # An unknown compression would be read as uncompressed - # garbage, so stop here instead. if record.compression not in SUPPORTED_COMPRESSIONS: raise ValueError( f"unsupported chunk compression '{record.compression}'; " @@ -259,42 +253,44 @@ def scan(stream: StreamReader, top_level: bool) -> None: schemas[record.id] = record elif isinstance(record, mcap_records.Channel): add_channel(record) - elif isinstance(record, mcap_records.Message): - saw_message = True - elif top_level and isinstance(record, mcap_records.Attachment): - # Counted in this pass only, so a later pass cannot - # double the count. + elif isinstance(record, mcap_records.Attachment): attachment_count += 1 - elif top_level and isinstance(record, mcap_records.Statistics): - statistics = record with open(path, "rb") as file: if file.read(len(MCAP_MAGIC)) != MCAP_MAGIC: raise ValueError(f"'{path.name}' is not an MCAP file (bad magic bytes)") - file.seek(0) - # Chunks stay unopened here, so this pass is cheap: it checks - # compression and picks up the records of unchunked files. - scan(StreamReader(file, emit_chunks=True), top_level=True) - - # The summary section usually repeats what the chunks hold. file.seek(0) try: summary = make_reader(file).get_summary() except Exception: summary = None + if summary is not None: + for chunk_index in summary.chunk_indexes: + if chunk_index.compression not in SUPPORTED_COMPRESSIONS: + raise ValueError( + f"unsupported chunk compression '{chunk_index.compression}'; " + "supported compressions are none, zstd, and lz4" + ) schemas.update(summary.schemas) for channel in sorted(summary.channels.values(), key=lambda c: c.id): add_channel(channel) - - # Those repeats are optional, so what we have may be incomplete. A - # Statistics record covering at least one message means the summary is - # trustworthy; without one, read the chunks. - have_range = statistics is not None and statistics.message_count > 0 - if summary is None or (not have_range and (saw_chunks or not saw_message)): + attachment_count = len(summary.attachment_indexes) + + if not channels: + # Repeating the records in the summary is optional, so fall back to + # reading the data section. Anything counted above gets counted + # again there, so start over. + attachment_count = 0 + if summary is None: + # Chunk records carry the only copy of the compression strings + # when there are no chunk indexes to read them from, and they + # are visible only while the chunks stay unopened. + file.seek(0) + scan(StreamReader(file, emit_chunks=True)) file.seek(0) - scan(StreamReader(file, emit_chunks=False), top_level=False) + scan(StreamReader(file, emit_chunks=False)) if attachment_count: parse_warnings.append( diff --git a/python/lib/sift_client/_tests/_internal/test_mcap.py b/python/lib/sift_client/_tests/_internal/test_mcap.py index a2c0d36d0..be248831b 100644 --- a/python/lib/sift_client/_tests/_internal/test_mcap.py +++ b/python/lib/sift_client/_tests/_internal/test_mcap.py @@ -42,16 +42,36 @@ def _leaves(data: str, name: str = "test_msgs/msg/Test"): """ -def _write_mcap(path, schemas_and_topics: list[tuple[str, str, str]]) -> None: - """Write an MCAP file with one channel per (schema_name, schema_text, topic).""" +def _write_mcap( + path, + schemas_and_topics: list[tuple[str, str, str]], + messages: int = 0, + attachment: bool = False, + **writer_kwargs, +) -> None: + """Write an MCAP file with one channel per (schema_name, schema_text, topic). + + Each channel logs ``messages`` cdr messages, and ``writer_kwargs`` go to the + mcap ``Writer`` (chunking, schema and channel repeats, statistics). + """ with open(path, "wb") as f: - writer = Writer(f) + writer = Writer(f, **writer_kwargs) writer.start() for schema_name, schema_text, topic in schemas_and_topics: schema_id = writer.register_schema( name=schema_name, encoding="ros2msg", data=schema_text.encode() ) - writer.register_channel(topic=topic, message_encoding="cdr", schema_id=schema_id) + channel_id = writer.register_channel( + topic=topic, message_encoding="cdr", schema_id=schema_id + ) + for i in range(messages): + writer.add_message( + channel_id=channel_id, log_time=i, publish_time=i, data=b"\x00" * 32 + ) + if attachment: + writer.add_attachment( + create_time=1, log_time=1, name="notes.txt", media_type="text/plain", data=b"hi" + ) writer.finish() @@ -74,19 +94,19 @@ def test_fixed_array_expands_per_element(self): "accel[2]", ] - def test_variable_array_is_one_complex_leaf(self): - assert _leaves("int32[] samples\n") == [("samples", "complex", None)] - - def test_bounded_array_is_one_complex_leaf(self): - assert _leaves("int32[<=4] samples\n") == [("samples", "complex", None)] - - def test_variable_array_of_messages_is_one_complex_leaf(self): - schema = ( - "geometry_msgs/Vector3[] path\n" + @pytest.mark.parametrize( + "definition", + [ + "int32[] samples\n", + "int32[<=4] samples\n", + "geometry_msgs/Vector3[] samples\n" + "=" * 80 - + "\nMSG: geometry_msgs/Vector3\nfloat64 x\nfloat64 y\nfloat64 z\n" - ) - assert _leaves(schema) == [("path", "complex", None)] + + "\nMSG: geometry_msgs/Vector3\nfloat64 x\n", + ], + ids=["unbounded", "bounded", "of_messages"], + ) + def test_variable_array_is_one_complex_leaf(self, definition): + assert _leaves(definition) == [("samples", "complex", None)] def test_fixed_array_of_messages_expands_per_element(self): schema = ( @@ -151,20 +171,20 @@ def test_maps_every_ros2_scalar_type(self): ChannelDataType.STRING, ] - def test_wstring_raises_unsupported(self): - with pytest.raises(UnsupportedTopicError, match="wstring"): - _leaves("wstring label\n") - - def test_variable_array_of_wstring_raises(self): - # The importer decodes every element of a complex leaf, so an - # undecodable element type makes the whole topic unsupported. - with pytest.raises(UnsupportedTopicError, match="wstring"): - _leaves("wstring[] labels\n") - - def test_variable_array_of_messages_with_wstring_raises(self): - schema = "pkg/Bad[] items\n" + "=" * 80 + "\nMSG: pkg/Bad\nwstring label\n" + @pytest.mark.parametrize( + "definition", + [ + "wstring label\n", + "wstring[] labels\n", + "pkg/Bad[] items\n" + "=" * 80 + "\nMSG: pkg/Bad\nwstring label\n", + ], + ids=["scalar", "array", "nested_in_array"], + ) + def test_wstring_raises_unsupported(self, definition): + # Complex leaves still decode every element, so an undecodable element + # type anywhere makes the whole topic unsupported. with pytest.raises(UnsupportedTopicError, match="wstring"): - _leaves(schema) + _leaves(definition) def test_nesting_beyond_max_depth_raises(self): # A chain of 40 nested message types exceeds MAX_FIELD_DEPTH (32). @@ -327,46 +347,32 @@ def test_rejects_non_mcap_file(self, tmp_path): detect_mcap_config(path) def test_records_only_in_data_section_are_detected(self, tmp_path): - # An unchunked file whose summary omits the schema/channel repeats is - # spec-legal; the top-level pass must pick the records up. + # An unchunked file whose summary omits the schema and channel repeats + # is spec-legal; the records are only in the data section. path = tmp_path / "log.mcap" - with open(path, "wb") as f: - writer = Writer(f, use_chunking=False, repeat_channels=False, repeat_schemas=False) - writer.start() - schema_id = writer.register_schema( - name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() - ) - writer.register_channel(topic="/imu", message_encoding="cdr", schema_id=schema_id) - writer.finish() + _write_mcap( + path, + [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], + use_chunking=False, + repeat_channels=False, + repeat_schemas=False, + ) config = detect_mcap_config(path) assert {d.topic for d in config.data} == {"/imu"} def test_rejects_unsupported_chunk_compression(self, tmp_path): - # The importer rejects unsupported compression regardless of - # parse_error_policy, so detection must too rather than listing - # channels for a file that can never import. + # Channels inside a chunk we cannot decompress are invisible, so the + # file can never import; refuse it rather than listing nothing. path = tmp_path / "log.mcap" - with open(path, "wb") as f: - writer = Writer(f) # defaults to zstd-compressed chunks - writer.start() - schema_id = writer.register_schema( - name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() - ) - channel_id = writer.register_channel( - topic="/imu", message_encoding="cdr", schema_id=schema_id - ) - writer.add_message(channel_id=channel_id, log_time=0, data=b"\x00", publish_time=0) - writer.finish() - # Rewrite the chunk's compression string to an unsupported one. + _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], messages=1) path.write_bytes(path.read_bytes().replace(b"zstd", b"lzma")) with pytest.raises(ValueError, match="unsupported chunk compression"): detect_mcap_config(path) def test_duplicate_channel_names_raise(self, tmp_path): - # Topic '/a' field 'b.c' and topic '/a.b' field 'c' both name a - # channel '/a.b.c'. + # '/a' field 'b.c' and '/a.b' field 'c' both name a channel '/a.b.c'. path = tmp_path / "log.mcap" _write_mcap( path, @@ -375,8 +381,11 @@ def test_duplicate_channel_names_raise(self, tmp_path): ("pkg/msg/C", "int32 c\n", "/a.b"), ], ) - with pytest.raises(ValueError, match="both named '/a.b.c'"): + + with pytest.raises(ValueError, match="both named '/a.b.c'") as excinfo: detect_mcap_config(path) + assert "topic '/a' field 'b.c'" in str(excinfo.value) + assert "topic '/a.b' field 'c'" in str(excinfo.value) class TestComplexTypesImportMode: @@ -438,9 +447,9 @@ def test_both_channels_share_the_field_selector(self, tmp_path): def test_default_is_both(self, tmp_path): path = tmp_path / "log.mcap" _write_mcap(path, [("test_msgs/msg/Samples", self.SCHEMA, "/samples")]) - config = detect_mcap_config(path) - assert config.complex_types_import_mode is McapComplexTypesImportMode.BOTH - assert len(config._to_proto().data) == 3 + assert detect_mcap_config(path).complex_types_import_mode is ( + McapComplexTypesImportMode.BOTH + ) def test_ignoring_every_configured_channel_raises(self, tmp_path): # An empty list would mean "import the whole file", so refuse instead. @@ -449,9 +458,7 @@ def test_ignoring_every_configured_channel_raises(self, tmp_path): with pytest.raises(ValueError, match="nothing would be imported"): config._to_proto() - def test_generated_json_name_clash_raises(self, tmp_path): - # '/a' field 'b' generates '/a.b.json', which topic '/a.b' field - # 'json' already claims. + def test_generated_json_name_clash_depends_on_the_mode(self, tmp_path): path = tmp_path / "log.mcap" _write_mcap( path, @@ -462,91 +469,66 @@ def test_generated_json_name_clash_raises(self, tmp_path): ) config = detect_mcap_config(path) assert {d.name for d in config.data} == {"/a.b", "/a.b.json"} + + # BOTH generates a second '/a.b.json' from '/a.b'; BYTES does not. with pytest.raises(ValueError, match="would both be imported as '/a.b.json'"): config._to_proto() + config.complex_types_import_mode = McapComplexTypesImportMode.BYTES + assert {d.channel_config.name for d in config._to_proto().data} == {"/a.b", "/a.b.json"} + + +class TestFileScanning: + """A well-formed file is read from its summary alone. The data section is + only read when the summary comes up short. + """ + + def test_well_formed_file_never_reads_the_data_section(self, tmp_path, monkeypatch): + path = tmp_path / "log.mcap" + _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], messages=200) + + def fail(*args, **kwargs): + raise AssertionError("read the data section for a file with a usable summary") - def test_clash_disappears_under_bytes_mode(self, tmp_path): + monkeypatch.setattr("sift_client._internal.util.mcap.StreamReader", fail) + config = detect_mcap_config(path) + assert {d.topic for d in config.data} == {"/imu"} + + def test_summary_without_repeats_falls_back_to_the_chunks(self, tmp_path): + # Statistics alone does not make the summary usable: without the + # channel repeats the channels exist only inside the chunks. path = tmp_path / "log.mcap" _write_mcap( path, - [ - ("pkg/msg/A", "int32[] b\n", "/a"), - ("pkg/msg/B", "int32 json\n", "/a.b"), - ], + [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], + messages=10, + repeat_channels=False, + repeat_schemas=False, ) - config = detect_mcap_config(path) - config.complex_types_import_mode = McapComplexTypesImportMode.BYTES - assert {d.channel_config.name for d in config._to_proto().data} == {"/a.b", "/a.b.json"} + config = detect_mcap_config(path) + assert {d.topic for d in config.data} == {"/imu"} -class TestScanCompleteness: - def test_summary_without_statistics_falls_back_to_chunk_scan(self, tmp_path): - # A chunked file whose summary repeats neither schemas nor channels and - # carries no Statistics record: the summary yields nothing, so - # detection must read the records inside the chunks like the importer. + def test_unsupported_compression_rejected_without_a_summary(self, tmp_path): + # No summary means no chunk indexes, so the compression string is only + # readable from the chunk records themselves. path = tmp_path / "log.mcap" - with open(path, "wb") as f: - writer = Writer(f, repeat_channels=False, repeat_schemas=False, use_statistics=False) - writer.start() - schema_id = writer.register_schema( - name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() - ) - channel_id = writer.register_channel( - topic="/imu", message_encoding="cdr", schema_id=schema_id - ) - for i in range(10): - writer.add_message( - channel_id=channel_id, - log_time=1_700_000_000_000_000_000 + i, - publish_time=1_700_000_000_000_000_000 + i, - data=b"\x00" * 32, - ) - writer.finish() + _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], messages=1) + path.write_bytes(path.read_bytes().replace(b"zstd", b"lzma")[:-8]) - config = detect_mcap_config(path) - assert {d.topic for d in config.data} == {"/imu"} + with pytest.raises(ValueError, match="unsupported chunk compression"): + detect_mcap_config(path) def test_attachments_warn(self, tmp_path): path = tmp_path / "log.mcap" - with open(path, "wb") as f: - writer = Writer(f) - writer.start() - schema_id = writer.register_schema( - name="sensors/msg/Imu", encoding="ros2msg", data=IMU_SCHEMA.encode() - ) - writer.register_channel(topic="/imu", message_encoding="cdr", schema_id=schema_id) - writer.add_attachment( - create_time=1, log_time=1, name="notes.txt", media_type="text/plain", data=b"hi" - ) - writer.finish() + _write_mcap(path, [("sensors/msg/Imu", IMU_SCHEMA, "/imu")], attachment=True) with pytest.warns(UserWarning, match="1 attachment"): config = detect_mcap_config(path) assert {d.topic for d in config.data} == {"/imu"} - def test_scan_warnings_survive_a_name_clash(self, tmp_path): + def test_warnings_survive_a_name_clash(self, tmp_path): # The clash raises, but what the scan already found must still reach # the caller instead of being discarded with the exception. - path = tmp_path / "log.mcap" - nested = "pkg/B b\n" + "=" * 80 + "\nMSG: pkg/B\nint32 c\n" - with open(path, "wb") as f: - writer = Writer(f) - writer.start() - a = writer.register_schema(name="pkg/msg/A", encoding="ros2msg", data=nested.encode()) - writer.register_channel(topic="/a", message_encoding="cdr", schema_id=a) - b = writer.register_schema(name="pkg/msg/C", encoding="ros2msg", data=b"int32 c\n") - writer.register_channel(topic="/a.b", message_encoding="cdr", schema_id=b) - writer.add_attachment( - create_time=1, log_time=1, name="notes.txt", media_type="text/plain", data=b"hi" - ) - writer.finish() - - with pytest.warns(UserWarning, match="1 attachment"), pytest.raises( - ValueError, match="both named" - ): - detect_mcap_config(path) - - def test_clash_error_names_both_origins(self, tmp_path): path = tmp_path / "log.mcap" _write_mcap( path, @@ -554,9 +536,10 @@ def test_clash_error_names_both_origins(self, tmp_path): ("pkg/msg/A", "pkg/B b\n" + "=" * 80 + "\nMSG: pkg/B\nint32 c\n", "/a"), ("pkg/msg/C", "int32 c\n", "/a.b"), ], + attachment=True, ) - with pytest.raises(ValueError, match="both named") as excinfo: + + with pytest.warns(UserWarning, match="1 attachment"), pytest.raises( + ValueError, match="both named" + ): detect_mcap_config(path) - message = str(excinfo.value) - assert "topic '/a' field 'b.c'" in message - assert "topic '/a.b' field 'c'" in message