ZTh/>dZddlZddlZddlmZddlmZddlmZmZm Z m Z m Z m Z m Z ddlmZ ddlmZddlmZdd lmZmZmZdd lmZdd lmZmZdd l m!Z!ddl"m#Z#ddl$m%Z%ddl&m'Z'ddl(m)Z)ddl*m+Z+ddl,m-Z-GddZ.y#e$rZed edZ[wwxYw)a  Service to upload ROS2 bag files. ROS organizes data exchange through topics, messages, and fields: * topics: Named messages that facilitate communication between ROS nodes. * messages: Data structures published and subscribed to on topics. Each message type defines a specific schema. * fields: Individual data elements within a message, such as integers, floats, strings, or nested structures. This class extracts messages from a ROS bag, flattens their fields, and prepares them for uploading to Sift. N)glob)Path)CallableDictListOptionalSetTextIOUnion)alive_it) Typesdict)Reader)Storesget_types_from_msg get_typestore) TypestorezThe rosbags package is required to use the rosbag upload service. Please include this dependency in your project by specifying `sift-stack-py[rosbags]`.) DataColumn TimeColumn) RosChannel) CsvConfig)CsvUploadService)DataImportService)NamedTemporaryFile)TimeFormatType)SiftRestConfigcVeZdZUdZeed<defdZ ddee e fde ee e fde d e d e d ee d ee d eee efde defdZ ddee e fdede ee e fde d e d eee efde de efdZdede ee e fdee fdZ dde ed e d ee d ee def dZy)RosbagsUploadServicez+ Service to upload ROS2 bag files. _csv_upload_service rest_confc$t||_y)N)rr)selfrs `/home/www/backend.miabetepe.com/venv/lib/python3.12/site-packages/sift_py/data_import/rosbags.py__init__zRosbagsUploadService.__init__1s#3I#> Npathmsg_dirsstore asset_name ignore_errorsrun_namerun_idhandlers show_progressreturnc .t|tr t|n|} | jst d|d| j st d|dt dd5} |j|| ||||| } | st d| |j| |||}td |jj| j|| cd d d S#tj$r} t d| d } ~ wwxYw#1swYy xYw) a Uploads the ROS2 bag file pointed to by `path` to the specified asset. Arguments: path: Path to the ROS2 bag file. msg_dirs: List of directories containing message definitions. Each entry should be a path the root directory of the msg definitions (e.g, `/path/to/std_msgs`). Inspect your topics and verify that the 'type' matches the directory structure of your message definitions. For example if the type is `custom_msgs/msg/MyCustomMessage` your directory structure should match that and you should include `/path/to/custom_msgs` in the `msg_dirs` list passed into this function. store: The Store type to use for the message definitions. asset_name: Name of the asset to upload the data to. ignore_errors: If True, will skip messages without definitions. run_name: Name of the run to create for this data. run_id: ID of the run to add this data to. handlers: Dictionary of messages to callbacks for custom processing or sequence data (e.g, images or videos). Keys should be the ROS topic, value is a callable with the following signature: def callback(topic: str, timestamp: int, msg: object) ... show_progress: Whether to show the status bar or not. zProvided path, 'z', does not existsz!', does not point to a directory.wtz.csv.gz)modesuffixzNo valid channels remaining in zXFailed to parse the rosbag. Ensure you're using the correct type store. Original error: NzUploading file) isinstancestrrexists Exceptionis_dirr_convert_to_csvstructerror_create_csv_configprintruploadname)r!r%r&r'r(r)r*r+r,r- posix_path temp_filevalid_channelse csv_configs r"r=zRosbagsUploadService.upload4s<F$.dC#8T$Zd   ".tf4FGH H  ".tf4UVW W T) < ^  !%!5!5!!"&#&EdV$LMM&00XW]^J " #++229>>:}]+ ^ ^<< nopnqr  ^ ^s1)D +'C"AD "D5DDD  Dsrc_pathdst_filec|xsi}t|}|j||} i} d} d} t|5} | jD]}|j| vr5|rt d|jd-t d|jd|j|j}|jD]H}| |||}|| vrt d|tj| |j||| |<Jdg| jDcgc]}|D]}|jc}}z}tj||}|j!t d t#| j%| j&d | D] \}}}|j| vr|rt d|jdi}|j)||j}|j|j}||d<|j|vr!||j|j|||jD]L}| |||}|| vr|rt d |d| |}|D] }|j+|||j<"N|j-|  d d d |j/| jD cgc] } | D]}| c}} Scc}}w#1swYFxYwcc}} w)aConverts the ROS2 bag file to a temporary CSV on disk that we will upload. Args: src_path: The path of the rosbag. dst_file: The path to save the CSV. msg_dirs: The list of directories containing rosbag message definitions. store: The rosbag type store to use. ignore_errors: Whether to ignore errors (e.g, unknown message definitions). handlers: Dictionary of messages to callbacks for custom processing or sequence data (e.g, images or videos). Keys should be the ROS topic, value is a callable with the following signature: def callback(topic: str, timestamp: int, msg: object) ... show_progress: Whether to show the status bar or not. Returns: The list valid channels after parsing the ROS2 bag file. crdj|jd}|jdr|dd}|S)N_/)joinsplit startswith)r>results r"sanitizez6RosbagsUploadService._convert_to_csv..sanitizes6XXdjjo.F  %Mr$c>|jd|jd|S)N:)topicr>) connectionmsg_deffields r"get_keyz5RosbagsUploadService._convert_to_csv..get_keys$ &&'qaw? ?r$zWARNING: Skipping z. msg file not found.z Message type z not found in custom types.zDuplicate key: timezProcessing rosbag messagesz messages)totalunitdisablezMessage field N)r_register_typesr connectionsmsgtyper<r6 get_msgdeffieldsrget_underlying_fieldsrRvalues channel_namecsv DictWriter writeheaderr messages message_countdeserialize_cdr extract_valuewriterowclose)r!rDrEr&r'r)r,r- typestoreregistered_msg_types ros_channelsrOrVreaderrSrTrUkeychannelscheadersw timestampraw_datarowmsgs r"r8z$RosbagsUploadService._convert_to_csvvsA8>r!%( #33IxH57    @H > $00  %%-AA$ 2:3E3E2FF[\] '+J,>,>+??Z[ $..z/A/AB$^^E!*gu=Cl*'/#(?@@(2(H(H !1!12E9)L%  (h/;/B/B/D"#+h"IJ""Gx1A MMO . /3;!** )) 4! / Ix %%-AA$ '+J,>,>+??Z[@B//*:L:LM#..z/A/AB'F ##x/.HZ--.z/?/?CP$^^ CE!*gu=C,.($"+nSEA\,]"^^+C0H%C./ooc.BANN+C C 3C! ;> D &2&9&9&;Rl\RRRR]"-> > HSs%CK K -E-K K,K  K)rlc i}|D]}t|}tt|dz dz D]}t|j|}|j|z }|j t t|jt|jddjdd|j|t|jS)a$Register custom message types with the typestore. Args: typestore: The type store to register messages against. msg_dirs: The list of directories containing message definitions. Returns: Set of all registered message definitions. z**z*.msg\rIz.msg) rrr4 relative_tor>updater read_textreplaceregistersetkeys) r!rlr& msg_types dir_pathnamedir_path msg_pathnamerelative_msg_pathmsg_path_from_roots r"r[z$RosbagsUploadService._register_typess " $ LL)H $SD7)B%C D  $($6$B$B<$P!%-]]5F%F"  &\*446./77cBJJ6SUV   9%9>>#$$r$rqci}d}t|D]2\}}|j} t|j| dd} | |||z<4||t t j d|d} ||| d<||| d<t| S) zAConstruct a CsvConfig based on metadata within the ROS2 bag file.r{)r> data_type descriptionunitsrJ)format column_number)r(first_data_row time_column data_columnsr*r+) enumeraterrrbrrABSOLUTE_UNIX_NANOSECONDSr) r!rqr(r*r+ data_configfirst_data_columnichannelrchannel_config config_infos r"r;z'RosbagsUploadService._create_csv_configs.0 #H- @JAw))I'))# N2@K)A- . @%/%%??(   &.K #  $*K !%%r$)FNNNT)NT)NN)__name__ __module__ __qualname____doc__r__annotations__rr#r r4rrrboolrrrrr=r rr8rr r[rr;r$r"rr*s*)?.?$"& $26"@^CI@^uS$Y'(@^ @^  @^  @^3-@^ @^4X ./@^@^ @^R37"pST "pSpSuS$Y'( pS  pS  pS4X ./pSpS j pSd%%d5dCS>T%Y\]`Ya%>#' $ '&z"'&'&3- '&  '&  '&r$r)/rrcr9rpathlibrtypingrrrrr r r alive_progressr rosbags.interfaces.typingr rosbags.rosbag2.readerrrosbags.typesysrrrrosbags.typesys.storer ImportErrorrB RuntimeErrorsift_py.data_import._configrr sift_py.data_import._ros_channelrsift_py.data_import.configrsift_py.data_import.csvrsift_py.data_import.statusrsift_py.data_import.tempfilersift_py.data_import.time_formatr sift_py.restrrrr$r"rs  EEE#  3-II/?7048;:'@&@&!  a    sB B BB