diff --git a/.github/workflows/snap-promote.yaml b/.github/workflows/snap-promote.yaml new file mode 100644 index 0000000..006c352 --- /dev/null +++ b/.github/workflows/snap-promote.yaml @@ -0,0 +1,24 @@ +# ============================================================================= +# Snap Promote — manual gate: jazzy/candidate -> jazzy/stable +# +# Run via the Actions tab once a candidate revision has soaked. +# Uses the canonical reusable promote workflow. +# ============================================================================= + +name: Snap Promote + +on: + workflow_dispatch: + +permissions: {} + +jobs: + promote: + permissions: + contents: read + uses: canonical/robotics-actions-workflows/.github/workflows/promote.yaml@main + secrets: + snapstore-login: ${{ secrets.SNAPSTORE_LOGIN }} + with: + from-channel: jazzy/candidate + to-channel: jazzy/stable diff --git a/.github/workflows/snap.yaml b/.github/workflows/snap.yaml new file mode 100644 index 0000000..391ff87 --- /dev/null +++ b/.github/workflows/snap.yaml @@ -0,0 +1,81 @@ +# ============================================================================= +# Snap Package — build, test and publish via the canonical reusable workflow. +# +# Branch push -> jazzy/edge, tag push -> jazzy/candidate. PRs build+test only. +# Promotion jazzy/candidate -> jazzy/stable lives in snap-promote.yaml. +# +# SETUP: store the credentials from +# snapcraft export-login --snaps pj-bridge +# as the SNAPSTORE_LOGIN repository secret. +# ============================================================================= + +name: Snap Package + +on: + push: + branches: [main] + tags: ['*'] + pull_request: + workflow_dispatch: + +permissions: {} + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true + +jobs: + snap: + permissions: + contents: read + actions: write + uses: canonical/robotics-actions-workflows/.github/workflows/snap.yaml@main + secrets: + snapstore-login: ${{ secrets.SNAPSTORE_LOGIN }} + with: + runs-on: '["ubuntu-latest", "ubuntu-24.04-arm"]' + snap-track: jazzy + # Packaging check: start the bridge snap, have a ros2-cli snap publish a + # topic, drive one WebSocket subscribe, and confirm the bridge served it. + snap-test-script: | + #!/usr/bin/env bash + set -euxo pipefail + + sudo snap install ros2-cli --channel jazzy + # websocat has no snap; grab the pinned static musl binary for this runner. + case "$(uname -m)" in + x86_64) websocat_arch=x86_64 ;; + aarch64) websocat_arch=aarch64 ;; + *) echo "Unsupported architecture: $(uname -m)" >&2; exit 1 ;; + esac + sudo curl -fsSL -o /usr/local/bin/websocat \ + "https://github.com/vi/websocat/releases/download/v1.13.0/websocat.${websocat_arch}-unknown-linux-musl" + sudo chmod +x /usr/local/bin/websocat + + pj-bridge.pj-bridge-ros2 --ros-args -p port:=9090 \ + > /tmp/pj_bridge.log 2>&1 & + BRIDGE_PID=$! + + ros2-cli.ros2 topic pub -r 1 /hello_world std_msgs/msg/String \ + "data: 'hello world'" > /tmp/pub_hello.log 2>&1 & + PUB_PID=$! + + # Retry while the WebSocket starts up and DDS discovers the topic. + for attempt in {1..10}; do + echo '{"command":"subscribe","topics":["/hello_world"],"id":"s1"}' \ + | timeout 5 websocat ws://localhost:9090 || true + if grep -q "subscribed to topic '/hello_world'" /tmp/pj_bridge.log; then + break + fi + sleep 1 + done + + set +e + grep -q "subscribed to topic '/hello_world'" /tmp/pj_bridge.log + TEST_EXIT=$? + set -e + + kill "${BRIDGE_PID}" "${PUB_PID}" 2>/dev/null || true + echo "--- Bridge log ---"; cat /tmp/pj_bridge.log + echo "--- Publisher log ---"; cat /tmp/pub_hello.log + exit "${TEST_EXIT}" diff --git a/.gitignore b/.gitignore index ce8fd53..2f4a7b9 100644 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,6 @@ install_tsan/ build_asan/ install_asan/ log/ + +# Built snap artifacts +*.snap diff --git a/README.md b/README.md index 0913a82..58788e9 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,20 @@ pixi global install pj-bridge-ros2-humble \ pj_bridge_ros2 --ros-args -p port:=9090 ``` +### Snap + +Pre-built snaps are available from the [Snap Store](https://snapcraft.io/pj-bridge). + +```bash +sudo snap install pj-bridge + +# Run +pj-bridge.pj-bridge-ros2 --ros-args -p port:=9090 +``` + +The snap bundles ROS 2 Jazzy and uses UDP-only DDS transport automatically. +No extra ROS 2 installation is required on the machine running the bridge. + ### AppImage Pre-built AppImages are available from [GitHub Releases](https://github.com/PlotJuggler/plotjuggler_bridge/releases). diff --git a/snap/local/CMakeLists.txt b/snap/local/CMakeLists.txt new file mode 100644 index 0000000..aad2c68 --- /dev/null +++ b/snap/local/CMakeLists.txt @@ -0,0 +1,5 @@ +cmake_minimum_required(VERSION 3.14) +project(pj_bridge_snap) + +find_package(ament_cmake REQUIRED) +ament_package() diff --git a/snap/local/fastrtps_no_shm.xml b/snap/local/fastrtps_no_shm.xml new file mode 100644 index 0000000..50acb94 --- /dev/null +++ b/snap/local/fastrtps_no_shm.xml @@ -0,0 +1,17 @@ + + + + + UDPv4Transport + UDPv4 + + + + + + UDPv4Transport + + false + + + diff --git a/snap/local/package.xml b/snap/local/package.xml new file mode 100644 index 0000000..86ca212 --- /dev/null +++ b/snap/local/package.xml @@ -0,0 +1,143 @@ + + + + pj_bridge_snap + 0.0.0 + Additional runtime middleware and interfaces for the pj-bridge snap + ubuntu-robotics + AGPL-3.0 + + ament_cmake + + + rmw_cyclonedds_cpp + rmw_fastrtps_cpp + + ackermann_msgs + action_msgs + actionlib_msgs + automotive_autonomy_msgs + automotive_navigation_msgs + automotive_platform_msgs + autoware_auto_msgs + can_msgs + cartographer_ros_msgs + cascade_lifecycle_msgs + control_msgs + controller_manager_msgs + depthai_ros_msgs + diagnostic_msgs + dwb_msgs + dynamixel_workbench_msgs + four_wheel_steering_msgs + foxglove_msgs + gazebo_msgs + geographic_msgs + geometry_msgs + gps_msgs + graph_msgs + grasping_msgs + grbl_msgs + grid_map_msgs + irobot_create_msgs + leo_msgs + lgsvl_msgs + lifecycle_msgs + map_msgs + marti_can_msgs + marti_common_msgs + marti_dbw_msgs + marti_introspection_msgs + marti_nav_msgs + marti_perception_msgs + marti_sensor_msgs + marti_status_msgs + marti_visualization_msgs + mavros_msgs + micro_ros_diagnostic_msgs + micro_ros_msgs + microstrain_inertial_msgs + moveit_msgs + mrpt_msgs + nao_command_msgs + nao_sensor_msgs + nav_2d_msgs + nav_msgs + nav2_msgs + nmea_msgs + novatel_gps_msgs + object_recognition_msgs + octomap_msgs + pal_statistics_msgs + pcl_msgs + pendulum_msgs + phidgets_msgs + plansys2_msgs + plotjuggler_msgs + radar_msgs + rc_common_msgs + rc_reason_msgs + rclpy_message_converter_msgs + rcss3d_agent_msgs + realsense2_camera_msgs + rmf_api_msgs + rmf_building_map_msgs + rmf_charger_msgs + rmf_dispenser_msgs + rmf_door_msgs + rmf_fleet_msgs + rmf_ingestor_msgs + rmf_lift_msgs + rmf_obstacle_msgs + rmf_scheduler_msgs + rmf_site_map_msgs + rmf_task_msgs + rmf_traffic_msgs + rmf_visualization_msgs + rmf_workcell_msgs + robot_calibration_msgs + rosapi_msgs + rosbridge_msgs + rosgraph_msgs + rviz_2d_overlay_msgs + sensor_msgs + shape_msgs + smacc2_msgs + soccer_vision_2d_msgs + soccer_vision_3d_msgs + soccer_vision_attribute_msgs + statistics_msgs + std_msgs + stereo_msgs + system_modes_msgs + teleop_tools_msgs + tf2_msgs + trajectory_msgs + turtlebot3_msgs + ublox_msgs + ublox_ubx_msgs + udp_msgs + unique_identifier_msgs + ur_dashboard_msgs + ur_msgs + urg_node_msgs + velodyne_msgs + vision_msgs + vision_msgs_layers + visualization_msgs + webots_ros2_msgs + wiimote_msgs + + + builtin_interfaces + composition_interfaces + example_interfaces + rcl_interfaces + ros_gz_interfaces + rosbag2_interfaces + type_description_interfaces + + + ament_cmake + + diff --git a/snap/snapcraft.yaml b/snap/snapcraft.yaml new file mode 100644 index 0000000..5b4c7df --- /dev/null +++ b/snap/snapcraft.yaml @@ -0,0 +1,73 @@ +name: pj-bridge +adopt-info: pj_bridge # version and grade derived from git tags (see pj_bridge part) +version: git +title: PlotJuggler Bridge +summary: ROS2 to WebSocket bridge server — forwards topic content at 50 Hz +license: AGPL-3.0 +contact: https://github.com/PlotJuggler/plotjuggler_bridge/issues +issues: https://github.com/PlotJuggler/plotjuggler_bridge/issues +source-code: https://github.com/PlotJuggler/plotjuggler_bridge +website: https://github.com/PlotJuggler/plotjuggler_bridge +description: > + ROS 2 bridge server that forwards middleware topic content over WebSocket to + PlotJuggler clients and any other WebSocket consumer. + + Clients subscribe to topics and receive ZSTD-compressed aggregated messages at + 50 Hz without needing a full middleware installation. + + Call the snap with: + pj-bridge.pj-bridge-ros2 --ros-args -p port:=9090 + +confinement: strict +base: core24 + +# ROS 2 typesupport and middleware libraries are loaded via dlopen at runtime +# by the ROS 2 plugin system; the library linter cannot detect this. +lint: + ignore: + - library + +parts: + pj_bridge: + plugin: colcon + source: . + override-pull: | + craftctl default + # Version/grade from git tag: clean tag -> stable, ahead of tag -> devel. + git config --global --add safe.directory '*' + version="$(git describe --always --tags | sed -e 's/^v//;s/-/+git/;y/-/./')" + if echo "$version" | grep -q '+git'; then grade=devel; else grade=stable; fi + craftctl set version="$version" + craftctl set grade="$grade" + build-packages: + - git + - libssl-dev + - zlib1g-dev + + snap-dependencies: + # A separate manifest keeps the broad runtime dependencies (msgs & rmw) + plugin: colcon + source: snap/local + after: [pj_bridge] + colcon-packages: [pj_bridge_snap] + stage: + - -opt/ros/snap/local_setup.sh + - -opt/ros/snap/setup.sh + + fastrtps-config: + plugin: dump + source: snap/local + organize: + fastrtps_no_shm.xml: etc/pj-bridge/fastrtps_no_shm.xml + stage: + - etc/pj-bridge/fastrtps_no_shm.xml + +apps: + pj-bridge-ros2: + command: opt/ros/snap/lib/pj_bridge/pj_bridge_ros2 + extensions: [ros2-jazzy] + plugs: + - network + - network-bind + environment: + FASTRTPS_DEFAULT_PROFILES_FILE: $SNAP/etc/pj-bridge/fastrtps_no_shm.xml diff --git a/tests/integration/test_client.py b/tests/integration/test_client.py deleted file mode 100755 index 1f91564..0000000 --- a/tests/integration/test_client.py +++ /dev/null @@ -1,490 +0,0 @@ -#!/usr/bin/env python3 -""" -ROS2 Bridge Test Client (WebSocket) - -This script connects to the pj_ros_bridge server over WebSocket and allows -testing of all API operations including topic discovery, subscription, and -receiving aggregated messages. - -Dependencies: - pip install websocket-client zstandard -""" - -import argparse -import json -import struct -import sys -import threading -import time -from collections import defaultdict, deque -from datetime import datetime - -import websocket -import zstandard as zstd - - -class BridgeClient: - """Client for pj_ros_bridge server over WebSocket""" - - # Binary frame header constants - HEADER_SIZE = 16 - MAGIC_PJRB = 0x42524A50 # "PJRB" in ASCII (little-endian) - - def __init__(self, host="localhost", port=9090): - """ - Initialize the bridge client - - Args: - host: Server hostname - port: Server WebSocket port - """ - self.url = f"ws://{host}:{port}" - self.ws = None - - # Buffer for binary frames received while waiting for a text response - self._pending_binary = deque() - - # Lock for send operations (recv is single-threaded from main loop) - self._send_lock = threading.Lock() - - # Heartbeat thread - self.heartbeat_thread = None - self.heartbeat_running = False - - # Statistics - self.stats = defaultdict(lambda: {"count": 0, "bytes": 0}) - self.start_time = None - - # ZSTD decompressor - self.decompressor = zstd.ZstdDecompressor() - - def connect(self): - """Connect to the bridge server""" - self.ws = websocket.create_connection(self.url) - print(f"Connected to WebSocket server: {self.url}") - self.start_time = time.time() - - def disconnect(self): - """Disconnect from the server""" - self.stop_heartbeat() - - if self.ws: - self.ws.close() - - print("Disconnected from server") - - def send_request(self, command_data): - """ - Send a JSON request and wait for the JSON response. - - Any binary frames received while waiting are buffered for later - retrieval via receive_messages(). - - Args: - command_data: Dictionary containing the request - - Returns: - Dictionary containing the response - """ - with self._send_lock: - self.ws.send(json.dumps(command_data)) - - # Read frames until we get a text response - while True: - opcode, data = self.ws.recv_data() - if opcode == websocket.ABNF.OPCODE_TEXT: - return json.loads(data.decode("utf-8")) - elif opcode == websocket.ABNF.OPCODE_BINARY: - # Buffer binary data for later - self._pending_binary.append(data) - - def get_topics(self): - """ - Get list of available topics from the server - - Returns: - List of topic dictionaries with 'name' and 'type' keys - """ - request = {"command": "get_topics"} - response = self.send_request(request) - - if response.get("status") == "success": - return response.get("topics", []) - else: - error_msg = response.get("message", "Unknown error") - raise Exception(f"Failed to get topics: {error_msg}") - - def subscribe(self, topics): - """ - Subscribe to topics - - Args: - topics: List of topic names to subscribe to - - Returns: - Dictionary mapping topic names to schema objects. - Each schema object has: - - "encoding": Schema encoding (e.g., "ros2msg") - - "definition": The actual schema definition string - """ - request = {"command": "subscribe", "topics": topics} - response = self.send_request(request) - - if response.get("status") in ("success", "partial_success"): - return response.get("schemas", {}) - else: - error_msg = response.get("message", "Unknown error") - raise Exception(f"Failed to subscribe: {error_msg}") - - def send_heartbeat(self): - """Send a heartbeat to the server (thread-safe for send, but not recv)""" - with self._send_lock: - self.ws.send(json.dumps({"command": "heartbeat"})) - # Note: response will be picked up by the main recv loop - - def start_heartbeat(self, interval=1.0): - """ - Start heartbeat thread. - - The heartbeat thread only sends; responses are consumed by the - main receive loop and silently discarded. - - Args: - interval: Heartbeat interval in seconds (default: 1.0) - """ - self.heartbeat_running = True - - def heartbeat_loop(): - while self.heartbeat_running: - try: - self.send_heartbeat() - except Exception as e: - print(f"Heartbeat error: {e}") - time.sleep(interval) - - self.heartbeat_thread = threading.Thread(target=heartbeat_loop, daemon=True) - self.heartbeat_thread.start() - print(f"Heartbeat thread started (interval: {interval}s)") - - def stop_heartbeat(self): - """Stop heartbeat thread""" - if self.heartbeat_running: - self.heartbeat_running = False - if self.heartbeat_thread: - self.heartbeat_thread.join(timeout=2.0) - print("Heartbeat thread stopped") - - def deserialize_aggregated_messages(self, data): - """ - Deserialize aggregated messages from binary format (streaming, no header). - - Format per message: - - Topic name length (uint16_t) - - Topic name (N bytes UTF-8) - - Timestamp (uint64_t nanoseconds) - - Message data length (uint32_t) - - Message data (N bytes) - - Args: - data: Decompressed binary data - - Returns: - List of message dictionaries - """ - messages = [] - offset = 0 - - while offset < len(data): - if offset + 2 > len(data): - break - - # Topic name length - topic_len = struct.unpack("10} {'Bytes':>12} {'Rate (Hz)':>12}") - print(f"{'-' * 80}") - - total_count = 0 - total_bytes = 0 - - for topic in sorted(self.stats.keys()): - count = self.stats[topic]["count"] - bytes_val = self.stats[topic]["bytes"] - rate = count / elapsed if elapsed > 0 else 0 - - print(f"{topic:<40} {count:>10} {bytes_val:>12} {rate:>12.2f}") - - total_count += count - total_bytes += bytes_val - - print(f"{'-' * 80}") - total_rate = total_count / elapsed if elapsed > 0 else 0 - print(f"{'TOTAL':<40} {total_count:>10} {total_bytes:>12} {total_rate:>12.2f}") - print(f"{'=' * 80}\n") - - -def main(): - """Main entry point""" - parser = argparse.ArgumentParser(description="pj_ros_bridge test client (WebSocket)") - - parser.add_argument( - "--server", - default="localhost", - help="Server hostname (default: localhost)", - ) - parser.add_argument( - "--port", - type=int, - default=9090, - help="WebSocket port (default: 9090)", - ) - parser.add_argument( - "--command", - choices=["get_topics", "subscribe"], - help="Command to execute", - ) - parser.add_argument( - "--subscribe", - nargs="+", - metavar="TOPIC", - help="Topics to subscribe to", - ) - parser.add_argument( - "--duration", - type=float, - default=10.0, - help="Duration to receive messages in seconds (default: 10)", - ) - parser.add_argument( - "--verbose", - action="store_true", - help="Print each received message", - ) - - args = parser.parse_args() - - # Create client - client = BridgeClient(args.server, args.port) - - try: - # Connect to server - client.connect() - - # Execute command - if args.command == "get_topics" or (args.command is None and args.subscribe is None): - # Get and display topics - print("\nQuerying available topics...") - topics = client.get_topics() - - print(f"\nAvailable topics ({len(topics)}):") - print(f"{'Topic Name':<50} {'Type':<50}") - print("-" * 100) - for topic in topics: - print(f"{topic['name']:<50} {topic['type']:<50}") - - elif args.subscribe or args.command == "subscribe": - # Subscribe to topics - topics = args.subscribe if args.subscribe else [] - - if not topics: - print("Error: No topics specified for subscription") - return 1 - - print(f"\nSubscribing to {len(topics)} topic(s)...") - schemas = client.subscribe(topics) - - print(f"Successfully subscribed to {len(schemas)} topic(s)") - for topic, schema_obj in schemas.items(): - # Handle both old (string) and new (object) schema formats - if isinstance(schema_obj, dict): - encoding = schema_obj.get("encoding", "unknown") - definition = schema_obj.get("definition", "") - def_preview = definition[:50] + "..." if len(definition) > 50 else definition - def_preview = def_preview.replace("\n", " ") - print(f" - {topic} (encoding={encoding}, schema={def_preview})") - else: - # Legacy string format - print(f" - {topic}") - - # Start heartbeat - client.start_heartbeat() - - # Receive messages - print(f"\nReceiving messages for {args.duration} seconds...") - print("Press Ctrl+C to stop\n") - - start_time = time.time() - msg_count = 0 - - try: - while time.time() - start_time < args.duration: - messages = client.receive_messages(timeout=0.1) - - if messages: - msg_count += len(messages) - - if args.verbose: - for msg in messages: - now_ns = int(time.time() * 1_000_000_000) - latency_ns = now_ns - msg["timestamp_ns"] - latency_ms = latency_ns / 1_000_000 - - print( - f"[{datetime.now().strftime('%H:%M:%S.%f')[:-3]}] " - f"{msg['topic']}: {len(msg['data'])} bytes, " - f"latency: {latency_ms:.2f} ms" - ) - - except KeyboardInterrupt: - print("\n\nInterrupted by user") - - # Print statistics - client.print_statistics() - - except Exception as e: - print(f"Error: {e}") - return 1 - - finally: - client.disconnect() - - return 0 - - -if __name__ == "__main__": - sys.exit(main())