Fixes for new mypy (Ubuntu 24.04)

.github/workflows/test.yml

@@ -91,7 +91,7 @@ jobs:
     - name: Is regeneration of Python stubs required?
       run: |
         rosbag2_path=$(colcon list -p --packages-select rosbag2)/..
-        sudo apt update && sudo apt -y install mypy=0.942-1ubuntu1
+        sudo apt update && sudo apt -y install mypy
         source install/setup.sh
         stubgen -p rosbag2_py -o ${rosbag2_path}/rosbag2_py
         cd ${rosbag2_path}

rosbag2_py/README.md

@@ -14,7 +14,12 @@ source install/setup.sh
 # Make sure rosbag2_py can be imported
 python3 -c 'import rosbag2_py'
 
-sudo apt update && sudo apt install mypy=0.942-1ubuntu1
+# Ubuntu 24.04
+sudo apt update && sudo apt install mypy
+
+# Older Ubuntu
+# sudo pip3 install -U --break-system-packages mypy==1.9
+
 cd <rosbag2 git repo>
 stubgen -p rosbag2_py -o rosbag2_py
 ```

rosbag2_py/rosbag2_py/__init__.pyi

@@ -6,3 +6,5 @@ from rosbag2_py._reindexer import Reindexer as Reindexer
 from rosbag2_py._storage import BagMetadata as BagMetadata, ConverterOptions as ConverterOptions, FileInformation as FileInformation, MessageDefinition as MessageDefinition, MetadataIo as MetadataIo, ReadOrder as ReadOrder, ReadOrderSortBy as ReadOrderSortBy, StorageFilter as StorageFilter, StorageOptions as StorageOptions, TopicInformation as TopicInformation, TopicMetadata as TopicMetadata, get_default_storage_id as get_default_storage_id, to_rclcpp_qos_vector as to_rclcpp_qos_vector
 from rosbag2_py._transport import PlayOptions as PlayOptions, Player as Player, RecordOptions as RecordOptions, Recorder as Recorder, ServiceRequestsSource as ServiceRequestsSource, bag_rewrite as bag_rewrite
 from rosbag2_py._writer import SequentialCompressionWriter as SequentialCompressionWriter, SequentialWriter as SequentialWriter, get_registered_compressors as get_registered_compressors, get_registered_serializers as get_registered_serializers, get_registered_writers as get_registered_writers
+
+__all__ = ['bag_rewrite', 'CompressionMode', 'CompressionOptions', 'compression_mode_from_string', 'compression_mode_to_string', 'ConverterOptions', 'FileInformation', 'get_default_storage_id', 'get_registered_readers', 'get_registered_writers', 'get_registered_compressors', 'get_registered_serializers', 'to_rclcpp_qos_vector', 'ReadOrder', 'ReadOrderSortBy', 'Reindexer', 'SequentialCompressionReader', 'SequentialCompressionWriter', 'SequentialReader', 'SequentialWriter', 'StorageFilter', 'StorageOptions', 'TopicMetadata', 'TopicInformation', 'BagMetadata', 'MessageDefinition', 'MetadataIo', 'Info', 'Player', 'PlayOptions', 'ServiceRequestsSource', 'Recorder', 'RecordOptions', 'LocalMessageDefinitionSource']

rosbag2_py/rosbag2_py/_compression_options.pyi

@@ -5,20 +5,17 @@ MESSAGE: CompressionMode
 NONE: CompressionMode
 
 class CompressionMode:
-    __doc__: ClassVar[str] = ...  # read-only
     __members__: ClassVar[dict] = ...  # read-only
     FILE: ClassVar[CompressionMode] = ...
     MESSAGE: ClassVar[CompressionMode] = ...
     NONE: ClassVar[CompressionMode] = ...
     __entries: ClassVar[dict] = ...
     def __init__(self, value: int) -> None: ...
     def __eq__(self, other: object) -> bool: ...
-    def __getstate__(self) -> int: ...
     def __hash__(self) -> int: ...
     def __index__(self) -> int: ...
     def __int__(self) -> int: ...
     def __ne__(self, other: object) -> bool: ...
-    def __setstate__(self, state: int) -> None: ...
     @property
     def name(self) -> str: ...
     @property

rosbag2_py/rosbag2_py/_message_definitions.pyi

@@ -1,5 +1,3 @@
-from typing import Any
-
 class LocalMessageDefinitionSource:
     def __init__(self) -> None: ...
-    def get_full_text(self, *args, **kwargs) -> Any: ...
+    def get_full_text(self, *args, **kwargs): ...

rosbag2_py/rosbag2_py/_reader.pyi

@@ -1,10 +1,8 @@
-from typing import Any
-
 class SequentialCompressionReader:
     def __init__(self) -> None: ...
-    def get_all_message_definitions(self, *args, **kwargs) -> Any: ...
-    def get_all_topics_and_types(self, *args, **kwargs) -> Any: ...
-    def get_metadata(self, *args, **kwargs) -> Any: ...
+    def get_all_message_definitions(self, *args, **kwargs): ...
+    def get_all_topics_and_types(self, *args, **kwargs): ...
+    def get_metadata(self, *args, **kwargs): ...
     def has_next(self) -> bool: ...
     def open(self, arg0, arg1) -> None: ...
     def open_uri(self, arg0: str) -> None: ...
@@ -16,9 +14,9 @@ class SequentialCompressionReader:
 
 class SequentialReader:
     def __init__(self) -> None: ...
-    def get_all_message_definitions(self, *args, **kwargs) -> Any: ...
-    def get_all_topics_and_types(self, *args, **kwargs) -> Any: ...
-    def get_metadata(self, *args, **kwargs) -> Any: ...
+    def get_all_message_definitions(self, *args, **kwargs): ...
+    def get_all_topics_and_types(self, *args, **kwargs): ...
+    def get_metadata(self, *args, **kwargs): ...
     def has_next(self) -> bool: ...
     def open(self, arg0, arg1) -> None: ...
     def open_uri(self, arg0: str) -> None: ...

rosbag2_py/rosbag2_py/_storage.pyi

@@ -1,12 +1,11 @@
-from typing import ClassVar, Dict, List
-
 import datetime
+from typing import ClassVar, Dict, List
 
 class BagMetadata:
     bag_size: int
     compression_format: str
     compression_mode: str
-    custom_data: Dict[str,str]
+    custom_data: Dict[str, str]
     duration: object
     files: List[FileInformation]
     message_count: int
@@ -16,7 +15,7 @@ class BagMetadata:
     storage_identifier: str
     topics_with_message_count: List[TopicInformation]
     version: int
-    def __init__(self, version: int = ..., bag_size: int = ..., storage_identifier: str = ..., relative_file_paths: List[str] = ..., files: List[FileInformation] = ..., duration: object = ..., starting_time: object = ..., message_count: int = ..., topics_with_message_count: List[TopicInformation] = ..., compression_format: str = ..., compression_mode: str = ..., custom_data: Dict[str,str] = ..., ros_distro: str = ...) -> None: ...
+    def __init__(self, version: int = ..., bag_size: int = ..., storage_identifier: str = ..., relative_file_paths: List[str] = ..., files: List[FileInformation] = ..., duration: object = ..., starting_time: object = ..., message_count: int = ..., topics_with_message_count: List[TopicInformation] = ..., compression_format: str = ..., compression_mode: str = ..., custom_data: Dict[str, str] = ..., ros_distro: str = ...) -> None: ...
 
 class ConverterOptions:
     input_serialization_format: str
@@ -71,20 +70,17 @@ class ReadOrder:
     def __init__(self, sort_by: ReadOrderSortBy = ..., reverse: bool = ...) -> None: ...
 
 class ReadOrderSortBy:
-    __doc__: ClassVar[str] = ...  # read-only
     __members__: ClassVar[dict] = ...  # read-only
     File: ClassVar[ReadOrderSortBy] = ...
     PublishedTimestamp: ClassVar[ReadOrderSortBy] = ...
     ReceivedTimestamp: ClassVar[ReadOrderSortBy] = ...
     __entries: ClassVar[dict] = ...
     def __init__(self, value: int) -> None: ...
     def __eq__(self, other: object) -> bool: ...
-    def __getstate__(self) -> int: ...
     def __hash__(self) -> int: ...
     def __index__(self) -> int: ...
     def __int__(self) -> int: ...
     def __ne__(self, other: object) -> bool: ...
-    def __setstate__(self, state: int) -> None: ...
     @property
     def name(self) -> str: ...
     @property
@@ -100,7 +96,7 @@ class StorageFilter:
     def __init__(self, topics: List[str] = ..., services_events: List[str] = ..., regex: str = ..., exclude_topics: List[str] = ..., exclude_service_events: List[str] = ..., regex_to_exclude: str = ...) -> None: ...
 
 class StorageOptions:
-    custom_data: Dict[str,str]
+    custom_data: Dict[str, str]
     end_time_ns: int
     max_bagfile_duration: int
     max_bagfile_size: int
@@ -111,7 +107,7 @@ class StorageOptions:
     storage_id: str
     storage_preset_profile: str
     uri: str
-    def __init__(self, uri: str, storage_id: str = ..., max_bagfile_size: int = ..., max_bagfile_duration: int = ..., max_cache_size: int = ..., storage_preset_profile: str = ..., storage_config_uri: str = ..., snapshot_mode: bool = ..., start_time_ns: int = ..., end_time_ns: int = ..., custom_data: Dict[str,str] = ...) -> None: ...
+    def __init__(self, uri: str, storage_id: str = ..., max_bagfile_size: int = ..., max_bagfile_duration: int = ..., max_cache_size: int = ..., storage_preset_profile: str = ..., storage_config_uri: str = ..., snapshot_mode: bool = ..., start_time_ns: int = ..., end_time_ns: int = ..., custom_data: Dict[str, str] = ...) -> None: ...
 
 class TopicInformation:
     message_count: int
@@ -129,7 +125,6 @@ class TopicMetadata:
     def equals(self, arg0: TopicMetadata) -> bool: ...
 
 class rmw_qos_durability_policy_t:
-    __doc__: ClassVar[str] = ...  # read-only
     __members__: ClassVar[dict] = ...  # read-only
     RMW_QOS_POLICY_DURABILITY_SYSTEM_DEFAULT: ClassVar[rmw_qos_durability_policy_t] = ...
     RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL: ClassVar[rmw_qos_durability_policy_t] = ...
@@ -138,19 +133,16 @@ class rmw_qos_durability_policy_t:
     __entries: ClassVar[dict] = ...
     def __init__(self, value: int) -> None: ...
     def __eq__(self, other: object) -> bool: ...
-    def __getstate__(self) -> int: ...
     def __hash__(self) -> int: ...
     def __index__(self) -> int: ...
     def __int__(self) -> int: ...
     def __ne__(self, other: object) -> bool: ...
-    def __setstate__(self, state: int) -> None: ...
     @property
     def name(self) -> str: ...
     @property
     def value(self) -> int: ...
 
 class rmw_qos_history_policy_t:
-    __doc__: ClassVar[str] = ...  # read-only
     __members__: ClassVar[dict] = ...  # read-only
     RMW_QOS_POLICY_HISTORY_KEEP_ALL: ClassVar[rmw_qos_history_policy_t] = ...
     RMW_QOS_POLICY_HISTORY_KEEP_LAST: ClassVar[rmw_qos_history_policy_t] = ...
@@ -159,19 +151,16 @@ class rmw_qos_history_policy_t:
     __entries: ClassVar[dict] = ...
     def __init__(self, value: int) -> None: ...
     def __eq__(self, other: object) -> bool: ...
-    def __getstate__(self) -> int: ...
     def __hash__(self) -> int: ...
     def __index__(self) -> int: ...
     def __int__(self) -> int: ...
     def __ne__(self, other: object) -> bool: ...
-    def __setstate__(self, state: int) -> None: ...
     @property
     def name(self) -> str: ...
     @property
     def value(self) -> int: ...
 
 class rmw_qos_liveliness_policy_t:
-    __doc__: ClassVar[str] = ...  # read-only
     __members__: ClassVar[dict] = ...  # read-only
     RMW_QOS_POLICY_LIVELINESS_AUTOMATIC: ClassVar[rmw_qos_liveliness_policy_t] = ...
     RMW_QOS_POLICY_LIVELINESS_MANUAL_BY_TOPIC: ClassVar[rmw_qos_liveliness_policy_t] = ...
@@ -180,19 +169,16 @@ class rmw_qos_liveliness_policy_t:
     __entries: ClassVar[dict] = ...
     def __init__(self, value: int) -> None: ...
     def __eq__(self, other: object) -> bool: ...
-    def __getstate__(self) -> int: ...
     def __hash__(self) -> int: ...
     def __index__(self) -> int: ...
     def __int__(self) -> int: ...
     def __ne__(self, other: object) -> bool: ...
-    def __setstate__(self, state: int) -> None: ...
     @property
     def name(self) -> str: ...
     @property
     def value(self) -> int: ...
 
 class rmw_qos_reliability_policy_t:
-    __doc__: ClassVar[str] = ...  # read-only
     __members__: ClassVar[dict] = ...  # read-only
     RMW_QOS_POLICY_RELIABILITY_BEST_EFFORT: ClassVar[rmw_qos_reliability_policy_t] = ...
     RMW_QOS_POLICY_RELIABILITY_RELIABLE: ClassVar[rmw_qos_reliability_policy_t] = ...
@@ -201,12 +187,10 @@ class rmw_qos_reliability_policy_t:
     __entries: ClassVar[dict] = ...
     def __init__(self, value: int) -> None: ...
     def __eq__(self, other: object) -> bool: ...
-    def __getstate__(self) -> int: ...
     def __hash__(self) -> int: ...
     def __index__(self) -> int: ...
     def __int__(self) -> int: ...
     def __ne__(self, other: object) -> bool: ...
-    def __setstate__(self, state: int) -> None: ...
     @property
     def name(self) -> str: ...
     @property

rosbag2_py/rosbag2_py/_transport.pyi

@@ -1,8 +1,7 @@
-from typing import Any, ClassVar, List
-
-from typing import overload
 import datetime
 import rosbag2_py._storage
+from _typeshed import Incomplete
+from typing import ClassVar, List, overload
 
 class PlayOptions:
     clock_publish_frequency: float
@@ -22,7 +21,7 @@ class PlayOptions:
     rate: float
     read_ahead_queue_size: int
     regex_to_filter: str
-    service_requests_source: Any
+    service_requests_source: Incomplete
     services_to_filter: List[str]
     start_offset: float
     start_paused: bool
@@ -38,7 +37,8 @@ class Player:
     @overload
     def __init__(self, arg0: str) -> None: ...
     def burst(self, storage_options: rosbag2_py._storage.StorageOptions, play_options: PlayOptions, num_messages: int) -> None: ...
-    def cancel(self, *args, **kwargs) -> Any: ...
+    @staticmethod
+    def cancel() -> None: ...
     def play(self, storage_options: rosbag2_py._storage.StorageOptions, play_options: PlayOptions) -> None: ...
 
 class RecordOptions:
@@ -74,23 +74,21 @@ class Recorder:
     def __init__(self) -> None: ...
     @overload
     def __init__(self, arg0: str) -> None: ...
-    def cancel(self, *args, **kwargs) -> Any: ...
+    @staticmethod
+    def cancel() -> None: ...
     def record(self, storage_options: rosbag2_py._storage.StorageOptions, record_options: RecordOptions, node_name: str = ...) -> None: ...
 
 class ServiceRequestsSource:
-    __doc__: ClassVar[str] = ...  # read-only
     __members__: ClassVar[dict] = ...  # read-only
     CLIENT_INTROSPECTION: ClassVar[ServiceRequestsSource] = ...
     SERVICE_INTROSPECTION: ClassVar[ServiceRequestsSource] = ...
     __entries: ClassVar[dict] = ...
     def __init__(self, value: int) -> None: ...
     def __eq__(self, other: object) -> bool: ...
-    def __getstate__(self) -> int: ...
     def __hash__(self) -> int: ...
     def __index__(self) -> int: ...
     def __int__(self) -> int: ...
     def __ne__(self, other: object) -> bool: ...
-    def __setstate__(self, state: int) -> None: ...
     @property
     def name(self) -> str: ...
     @property

rosbag2_py/rosbag2_py/_writer.pyi

@@ -1,6 +1,6 @@
-from typing import overload
 import rosbag2_py._compression_options
 import rosbag2_py._storage
+from typing import overload
 
 class SequentialCompressionWriter:
     def __init__(self, arg0: rosbag2_py._compression_options.CompressionOptions) -> None: ...