Open source rewrite of rqt_topic

- open sourced from Apex.AI

Signed-off-by: Evan Flynn <[email protected]>

pytest.ini

@@ -1,2 +1,3 @@
 [pytest]
 junit_family=xunit2
+qt_api=pyside2

requirements.txt

@@ -0,0 +1,2 @@
+pydantic>=2.5.3
+pytest-qt>=4.3.1

setup.py

@@ -1,39 +1,44 @@
 from setuptools import setup
 
-package_name = 'rqt_topic'
+package_name = "rqt_topic"
 setup(
     name=package_name,
-    version='1.7.0',
-    package_dir={'': 'src'},
-    packages=[package_name],
+    version="1.7.0",
+    package_dir={"": "src"},
+    packages=[
+        package_name,
+        f"{package_name}/models",
+        f"{package_name}/buttons",
+        f"{package_name}/views",
+        f"{package_name}/workers",
+    ],
     data_files=[
-        ('share/ament_index/resource_index/packages',
-            ['resource/' + package_name]),
-        ('share/' + package_name + '/resource', ['resource/TopicWidget.ui']),
-        ('share/' + package_name, ['package.xml']),
-        ('share/' + package_name, ['plugin.xml']),
+        ("share/ament_index/resource_index/packages", ["resource/" + package_name]),
+        ("share/" + package_name + "/resource", ["resource/TopicWidget.ui"]),
+        ("share/" + package_name, ["package.xml"]),
+        ("share/" + package_name, ["plugin.xml"]),
     ],
-    install_requires=['setuptools'],
+    install_requires=["setuptools"],
     zip_safe=True,
-    author='Dorian Scholz',
-    maintainer='Brandon Ong',
-    maintainer_email='[email protected]',
-    keywords=['ROS'],
+    author="Dorian Scholz",
+    maintainer="Brandon Ong",
+    maintainer_email="[email protected]",
+    keywords=["ROS"],
     classifiers=[
-        'Intended Audience :: Developers',
-        'License :: OSI Approved :: BSD License',
-        'Programming Language :: Python',
-        'Topic :: Software Development',
+        "Intended Audience :: Developers",
+        "License :: OSI Approved :: BSD License",
+        "Programming Language :: Python",
+        "Topic :: Software Development",
     ],
     description=(
-        'rqt_topic provides a GUI plugin for displaying debug information about ROS topics '
-        'including publishers, subscribers, publishing rate, and ROS Messages.'
+        "rqt_topic provides a GUI plugin for displaying debug information about ROS topics "
+        "including publishers, subscribers, publishing rate, and ROS Messages."
     ),
-    license='BSD',
-    tests_require=['pytest'],
+    license="BSD",
+    tests_require=["pytest"],
     entry_points={
-        'console_scripts': [
-            'rqt_topic = ' + package_name + '.main:main',
+        "console_scripts": [
+            "rqt_topic = " + package_name + ".main:main",
         ],
     },
 )

src/rqt_topic/buttons/clear.py

@@ -0,0 +1,15 @@
+from python_qt_binding.QtWidgets import QAction, QStyle
+
+
+class Clear(QAction):
+    def __init__(self, style, name: str = "Clear All"):
+        super(Clear, self).__init__(name)
+
+        # Style is provided by the widget that uses this button
+        self.style = style
+
+        self.clear_icon = self.style.standardIcon(QStyle.SP_DialogResetButton)
+
+        self.setIcon(self.clear_icon)
+        self.setIconText("Clear All")
+        self.setStatusTip("Clear all the views")

src/rqt_topic/buttons/hide_timestamps.py

@@ -0,0 +1,36 @@
+from python_qt_binding.QtWidgets import QAction
+
+
+# TODO(evan.flynn): it'd be better to make a generic "hideColumn" feature directly
+# in the QAbstractTableModel. This is an acceptable work around for now.
+class HideTimestamps(QAction):
+    def __init__(self, style, name: str = "Hide timestamps"):
+        super(HideTimestamps, self).__init__(name)
+
+        # Style is provided by the widget that uses this button
+        self.style = style
+
+        self.setStatusTip("Hide the timestamp columns from all views")
+        self.triggered.connect(self.toggle_hide)
+        self._hidden = False
+
+    def is_hidden(self) -> bool:
+        return self._hidden
+
+    def toggle_hide(self):
+        if self._hidden:
+            self.unhide()
+        else:
+            self.hide()
+
+    def hide(self):
+        """Timestamps are hidden."""
+        self.setText("Unhide timestamps")
+        self.setStatusTip("Unhide the timestamp columns from all views")
+        self._hidden = True
+
+    def unhide(self):
+        """Button is resumed."""
+        self.setText("Hide timestamps")
+        self.setStatusTip("Hide the timestamp columns from all views")
+        self._hidden = False

src/rqt_topic/buttons/resize_columns.py

@@ -0,0 +1,11 @@
+from python_qt_binding.QtWidgets import QAction
+
+
+class ResizeColumns(QAction):
+    def __init__(self, style, name: str = "Resize columns to contents"):
+        super(ResizeColumns, self).__init__(name)
+
+        # Style is provided by the widget that uses this button
+        self.style = style
+
+        self.setStatusTip("Resize columns to contents")

src/rqt_topic/buttons/toggle_highlight.py

@@ -0,0 +1,34 @@
+from python_qt_binding.QtWidgets import QAction
+
+
+class ToggleHighlight(QAction):
+    def __init__(self, style, name: str = "Disable highlighting"):
+        super(ToggleHighlight, self).__init__(name)
+
+        # Style is provided by the widget that uses this button
+        self.style = style
+
+        self.setStatusTip("Disable color highlighting for new messages")
+        self.triggered.connect(self.toggle_highlight)
+        self._is_highlighting = True
+
+    def is_highlighting(self) -> bool:
+        return self._is_highlighting
+
+    def toggle_highlight(self):
+        if self._is_highlighting:
+            self.no_highlight()
+        else:
+            self.highlight()
+
+    def highlight(self):
+        """Turn color highlighting on."""
+        self.setText("Disable highlighting")
+        self.setStatusTip("Disable color highlighting for new messages")
+        self._is_highlighting = True
+
+    def no_highlight(self):
+        """Turn color highlighting off."""
+        self.setText("Highlight new messages")
+        self.setStatusTip("Do not highlight rows for new messages")
+        self._is_highlighting = False

src/rqt_topic/buttons/toggle_pause.py

@@ -0,0 +1,42 @@
+from python_qt_binding.QtWidgets import QAction, QStyle
+
+
+class TogglePause(QAction):
+    def __init__(self, style, name: str = "Pause"):
+        super(TogglePause, self).__init__(name)
+
+        # Style is provided by the widget that uses this button
+        self.style = style
+
+        self.pause_icon = self.style.standardIcon(QStyle.SP_MediaPause)
+        self.play_icon = self.style.standardIcon(QStyle.SP_MediaPlay)
+
+        self.setIcon(self.pause_icon)
+        self.setIconText("Pause")
+        self.setStatusTip("Pause the view")
+        self._paused = False
+
+        self.triggered.connect(self.toggle_pause)
+
+    def is_paused(self) -> bool:
+        return self._paused
+
+    def toggle_pause(self):
+        if self._paused:
+            self.resume()
+        else:
+            self.pause()
+
+    def pause(self):
+        """Button is paused."""
+        self.setIcon(self.play_icon)
+        self.setText("Resume")
+        self.setStatusTip("Resume the view")
+        self._paused = True
+
+    def resume(self):
+        """Button is resumed."""
+        self.setIcon(self.pause_icon)
+        self.setText("Pause")
+        self.setStatusTip("Pause the view")
+        self._paused = False

src/rqt_topic/main.py

@@ -30,13 +30,15 @@
 
 import sys
 
+# setattr(sys, 'SELECT_QT_BINDING', 'pyside')
+
 from rqt_gui.main import Main
 
 
 def main():
     main = Main()
-    sys.exit(main.main(sys.argv, standalone='rqt_topic.topic.Topic'))
+    sys.exit(main.main(sys.argv, standalone="rqt_topic.topic.Topic"))
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

src/rqt_topic/models/message.py

@@ -0,0 +1,62 @@
+from datetime import datetime
+from typing import List
+
+from python_qt_binding.QtGui import QColor
+from pydantic import BaseModel, ConfigDict, field_validator
+import re
+
+TOPIC_RE = re.compile(r"^(\/([a-zA-Z0-9_]+))+$")
+
+
+class MessageModel(BaseModel):
+    timestamp: datetime = datetime.now()
+    topic: str = ""
+    message_type: str = ""
+    content: dict = {}
+
+    # TODO(evan.flynn): implement these later on
+    # recorded_timestamp: Optional[str] = "timestamp this message was recorded"
+    # source_node: Optional[str] = "node that sent this msg"
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    def __str__(self):
+        if not self.content:
+            return ""
+        return str(self.content)
+
+    @field_validator("topic")
+    def validate_topic(cls, value):
+        assert TOPIC_RE.match(value) is not None, f"Given topic is not valid: {value}"
+        return value
+
+    @field_validator("timestamp")
+    def validate_timestamp(cls, value):
+        return value
+
+    def total_seconds_old(self) -> datetime:
+        return (datetime.now() - self.timestamp).total_seconds()
+
+    def color_from_timestamp(self) -> QColor:
+        # multiply by 30 to scale / excentuate the alpha value, clip between 0 and 150
+        alpha = max(0, min(150, 150 - int(self.total_seconds_old() * 30)))
+        return QColor(90, 90, 90, alpha)
+
+    def clear(self):
+        self.timestamp = datetime.now()
+        self.topic = ""
+        self.message_type = ""
+        self.content = {}
+
+
+def generate_test_msgs(number_of_msgs: int = 100) -> List[MessageModel]:
+    """Generate a list of messages for testing."""
+    return [
+        MessageModel(
+            topic=f"/{i}/test_topic",
+            message_type="test_msgs/BasicTypes",
+            timestamp=datetime.now(),
+            content={f"test_{i}_key": f"value_{i}"},
+        )
+        for i in range(number_of_msgs)
+    ]