Open source rewrite of rqt_topic

package.xml

@@ -19,12 +19,13 @@
   <author>Dorian Scholz</author>
   <author email="[email protected]">Scott K Logan</author>
 
-  <exec_depend version_gte="0.2.19">python_qt_binding</exec_depend>
-  <exec_depend>rclpy</exec_depend>
-  <exec_depend>ros2topic</exec_depend>
-  <exec_depend>rqt_gui</exec_depend>
-  <exec_depend>rqt_gui_py</exec_depend>
-  <exec_depend>rqt_py_common</exec_depend>
+  <depend version_gte="0.2.19">python_qt_binding</depend>
+  <depend>rclpy</depend>
+  <depend>ros2topic</depend>
+  <depend>rqt_gui</depend>
+  <depend>rqt_gui_py</depend>
+  <depend>rqt_py_common</depend>
+  <depend>python3-pydantic</depend>
 
   <test_depend>ament_flake8</test_depend>
   <test_depend>ament_xmllint</test_depend>

requirements.txt

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

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')

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

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')

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

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 → 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()

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, 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)
+
+    @validator('topic')
+    def validate_topic(cls, value):
+        assert TOPIC_RE.match(value) is not None, f'Given topic is not valid: {value}'
+        return value
+
+    @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)
+    ]