&`i& dZddlZddlZddlmZddlmZmZmZm Z m Z m Z m Z m Z ddlmZddlmZddlmZddlmZdd lmZerddlZdd lmZmZmZejeZeGd d Z eGd deZ!dS)zMCAP (Message Capture) datasource for Ray Data. MCAP is a standardized format for storing timestamped messages from robotics and autonomous systems, commonly used for sensor data, control commands, and other time-series data. N) dataclass) TYPE_CHECKINGAnyDictIteratorListOptionalSetUnion)DelegatingBlockBuilder) _check_import)Block)FileBasedDatasource) DeveloperAPI)ChannelMessageSchemac.eZdZUdZeed<eed<dZdS) TimeRangezTime range for filtering MCAP messages. Attributes: start_time: Start time in nanoseconds (inclusive). end_time: End time in nanoseconds (exclusive). start_timeend_timec|j|jkr td|jd|jd|jdks |jdkrtd|jd|jdS)z)Validate time range after initialization.z start_time (z) must be less than end_time ()rz1time values must be non-negative, got start_time=z , end_time=N)rr ValueErrorselfs /home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/datasource/mcap_datasource.py __post_init__zTimeRange.__post_init__&s ?dm + +.t..!]...  ?Q  $-!"3"3,DO,, M,, #4"3N)__name__ __module__ __qualname____doc__int__annotations__rrrrrsCOOOMMM     rrc deZdZdZdgZ ddeeeefdeeeee efdee deeeee efd e f fd Z d d d ede efdZddddddde fdZddddddd edeeeff dZdefdZede fdZxZS)MCAPDatasourcea MCAP (Message Capture) datasource for Ray Data. This datasource provides reading of MCAP files with predicate pushdown optimization for filtering by topics, time ranges, and message types. MCAP is a standardized format for storing timestamped messages from robotics and autonomous systems, commonly used for sensor data, control commands, and other time-series data. Examples: Basic usage: >>> import ray # doctest: +SKIP >>> ds = ray.data.read_mcap("/path/to/data.mcap") # doctest: +SKIP With topic filtering and time range: >>> from ray.data.datasource import TimeRange # doctest: +SKIP >>> ds = ray.data.read_mcap( # doctest: +SKIP ... "/path/to/data.mcap", ... topics={"/camera/image_raw", "/lidar/points"}, ... time_range=TimeRange(start_time=1000000000, end_time=2000000000) ... ) # doctest: +SKIP With multiple files and metadata: >>> ds = ray.data.read_mcap( # doctest: +SKIP ... ["file1.mcap", "file2.mcap"], ... topics={"/camera/image_raw", "/lidar/points"}, ... message_types={"sensor_msgs/Image", "sensor_msgs/PointCloud2"}, ... include_metadata=True ... ) # doctest: +SKIP mcapNTpathstopics time_range message_typesinclude_metadatac tj|fi|t|dd|rt|nd|_|rt|nd|_||_||_dS)aInitialize MCAP datasource. Args: paths: Path or list of paths to MCAP files. topics: Optional list/set of topic names to include. If specified, only messages from these topics will be read. time_range: Optional TimeRange for filtering messages by timestamp. TimeRange contains start_time and end_time in nanoseconds, where both values must be non-negative and start_time < end_time. message_types: Optional list/set of message type names (schema names) to include. Only messages with matching schema names will be read. include_metadata: Whether to include MCAP metadata fields in the output. Defaults to True. When True, includes schema, channel, and message metadata. **file_based_datasource_kwargs: Additional arguments for FileBasedDatasource. r))modulepackageN)super__init__r set_topics_message_types _time_range_include_metadata)rr*r+r,r-r.file_based_datasource_kwargs __class__s rr3zMCAPDatasource.__init__Zs2 ??">???d66::::'-6s6{{{$ 4AKc-000t%!1rfzpyarrow.NativeFilepathreturnc#Kddlm}||}||jrt |jnd|jr |jjnd|jr |jjnddd}t}|D]K\}}} | ||| s| ||| |} | | L| dkr| VdSdS)aXRead MCAP file and yield blocks of message data. This method implements efficient MCAP reading with predicate pushdown. It uses MCAP's built-in filtering capabilities for optimal performance and applies additional filters when needed. Args: f: File-like object to read from. Must be seekable for MCAP reading. path: Path to the MCAP file being processed. Yields: Block: Blocks of MCAP message data as pyarrow Tables. Raises: ValueError: If the MCAP file cannot be read or has invalid format. r) make_readerNTF)r+rrlog_time_orderreverse) mcap.readerr? iter_messagesr5listr7rrr _should_include_message_message_to_dictaddnum_rowsbuild) rr;r<r?readermessagesbuilderschemachannelmessage message_datas r _read_streamzMCAPDatasource._read_stream}s9" ,+++++Q '')-?4 %%%46:6FPt'22D262BLT%.. (  )**(0 & & $FGW//II  00'4PPL KK % % % %      ! !--// ! ! ! ! ! " !rrMrrNrrOrc8|jr|r|j|jvrdSdS)a_Check if a message should be included based on filters. This method applies Python-level filtering that cannot be pushed down to the MCAP library level. Topic filters are already handled by the MCAP reader, so only message_types filtering is needed here. Args: schema: MCAP schema object containing message type information. channel: MCAP channel object containing topic and metadata. message: MCAP message object containing the actual data. Returns: True if the message should be included, False otherwise. FT)r6name)rrMrNrOs rrEz&MCAPDatasource._should_include_messages/$   6 fkAT.T.T5trc|j}|jdkrkt|jtrQ t j|jd}n##tjtf$r |j}YnwxYw||j |j |j |j d}|j r@||j|j|r|jnd|r|jnd|r|jnddt%|ddr||d<|S) aVConvert MCAP message to dictionary format. This method converts MCAP message objects into a standardized dictionary format suitable for Ray Data processing. Args: schema: MCAP schema object containing message type and encoding info. channel: MCAP channel object containing topic and channel metadata. message: MCAP message object containing the actual message data. path: Path to the source file (for include_paths functionality). Returns: Dictionary containing message data in Ray Data format. jsonzutf-8)datatopiclog_time publish_timesequenceN) channel_idmessage_encoding schema_nameschema_encoding schema_data include_pathsFr<)rVr\ isinstancebytesrUloadsdecodeJSONDecodeErrorUnicodeDecodeErrorrWrXrYrZr8updater[rSencodinggetattr)rrMrNrOr< decoded_datarPs rrFzMCAPDatasource._message_to_dicts2$|  #v - -*W\52Q2Q - ,#z',*=*=g*F*FGG (*<= , , ,&|  , !](#0(     !    ")"4(/(@28#B6;;d:@'Jvd28#B6;;d     4% 0 0 (#'L s,AA;:A;cdS)z1Return a human-readable name for this datasource.MCAPr&rs rget_namezMCAPDatasource.get_namesvrcdS)zWhether this datasource supports distributed reads. MCAP files can be read in parallel across multiple files. Tr&rs rsupports_distributed_readsz)MCAPDatasource.supports_distributed_readss tr)NNNT)r r!r"r#_FILE_EXTENSIONSr strrr r rboolr3rrrQrErrrFrmpropertyro __classcell__)r:s@rr(r(4s  Dx 8<*.>B!% !2!2S$s)^$!2tCy#c(234!2Y' !2  d3iS&9 :; !2  !2!2!2!2!2!2F-"2-"#-"(5/-"-"-"-"^)2=F .33)23=F3NQ3 c3h3333j#DXrr()"r#rUlogging dataclassesrtypingrrrrrr r r +ray.data._internal.delegating_block_builderr ray.data._internal.utilr ray.data.blockr)ray.data.datasource.file_based_datasourcerray.util.annotationsrpyarrowrBrrr getLoggerr loggerrr(r&rrrs !!!!!!QQQQQQQQQQQQQQQQQQQQNNNNNN111111 IIIIII------5NNN4444444444  8 $ $  2MMMMM(MMMMMr