WebアプリケーションによるROS2データの可視化1 基本部分の作成

import os
from glob import glob
from setuptools import find_packages, setup

package_name = 'visualize_sample_server'

setup(
    name=package_name,
    version='0.0.0',
    packages=find_packages(exclude=['test']),
    data_files=[
        ('share/ament_index/resource_index/packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
        (os.path.join('share', package_name, 'launch'), glob(os.path.join('launch', '*launch.[pxy][yma]*'))),  # <-- 追加
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='k-nagasawa',
    maintainer_email='k-nagasawa@isp.co.jp',
    description='TODO: Package description',
    license='TODO: License declaration',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'sample_sender_node = scripts.sample_sender_node:main',  # <-- ディレクトリ名変更
        ],
    },
)

import rclpy
from rclpy.node import Node
from std_msgs.msg import String

### Node name
NODE_NAME = 'sample_sender'

### Topic name
TOPIC_NAME = '/test/message'

### Timer period
TIMER_PERIOD = 1.0  # 1 秒おきに呼び出し

### Sample sender node
class SampleSenderNode(Node) :
    def __init__(self) :
        super().__init__(NODE_NAME)

        ### Initialize
        self.count = 0
        self.publisher = self.create_publisher(String, TOPIC_NAME, 10)

        ### Set timer
        self.timer = self.create_timer(TIMER_PERIOD, self.on_timer)

    def on_timer(self) :
        
        ### Send message
        self.count += 1
        msg = String()
        msg.data = f'Hello ({self.count})'
        self.publisher.publish(msg)

def main(args = None):
    # Initialize rclpy
    print("Start sample sender node.")
    rclpy.init(args = args)

    # Create node
    node = SampleSenderNode()

    # Spin
    try :
        rclpy.spin(node)
    except KeyboardInterrupt :  # Ctrl + C
        pass
    print("Shutdown sample sender node.")
    rclpy.shutdown()

if __name__ == '__main__':
    main()

import os
import launch
from launch import LaunchDescription
from launch_ros.actions import Node

def generate_launch_description() :
    ld = LaunchDescription()

    # ROS bridge
    node_rosbridge = Node(
        package = 'rosbridge_server',
        executable = 'rosbridge_websocket',
        arguments = [
            '--address', '0.0.0.0', '--port', '8888',
        ],
    )
    ld.add_action(node_rosbridge)

    # Sample text sender node
    node_sample_text_sender = Node(
        package = 'visualize_sample_server',
        executable = 'sample_sender_node',
        name = 'sample_text_sender',
    )
    ld.add_action(node_sample_text_sender)

    return ld

import React, { forwardRef, useImperativeHandle, useEffect, useRef, useState } from 'react';
import * as ROSLIB from 'roslib';

///////////////////////////////////////////////////
/// Settings                                                                        ///
///////////////////////////////////////////////////

// ROS接続URL
const ROS_CONNECTION_URL = 'ws://127.0.0.1:8888'

// Test string topic name
const TOPIC_NAME_TEST_STR = '/test/message';

///////////////////////////////////////////////////
/// Interfaces                                                                     ///
///////////////////////////////////////////////////

// String 型
export interface String {
    data: string;
}

///////////////////////////////////////////////////
/// ROS Operator                                                               ///
///////////////////////////////////////////////////

const RosOperator = forwardRef((props, ref) => {

    // 上位モジュールから呼び出される処理
    useImperativeHandle(ref, () => ({
        // ROSとの接続
        connect: () => {
            connectToROS();
        },
    }));

    // ROSLIB object
    const [ros, setRosObject] = useState<ROSLIB.Ros | null>(null);

    // Test string topic listener
    const [testStrListener, setTestStrListener] = useState<ROSLIB.Topic | null>(null);

    // Connect to ROS ... ROSとの接続を開始する
    const connectToROS = () => {
    
        // 接続開始
        console.log('ROS Operator: Try connection...');
        props.setRosConnected?.(false);  // NOTE: 上位モジュールのROS接続確認フラグをOFF

        // Initialize ROS connection object
        setRosObject(new ROSLIB.Ros({url:ROS_CONNECTION_URL}));
    };

    // ROS オブジェクト更新時に Listener & Sender を更新
    useEffect(() => {

        // 初回表示時は ROS オブジェクトが null なのでスキップ
        if (ros === null) { return; }

        // 接続完了検知 or エラー検知
        // NOTE: 検知時に上位モジュールにフラグ通知
        ros.on('connection', () => {
            console.log('ROS Operator: ROS connected.');
            props.setRosConnected?.(true);
        });
        ros.on('error', (err) => {
            console.log('ROS Operator: ROS connection error, ', err);
            props.setRosConnected?.(false);
        });
        ros.on('close', () => {
            console.log('ROS Operator: ROS connection closed.');
            props.setRosConnected?.(false);
        });

        // Test string topic listener
        setTestStrListener(
            new ROSLIB.Topic({
                ros: ros,
                name: TOPIC_NAME_TEST_STR,
                messageType: 'std_msgs/String',
            })
        );
    }, [ros]);

    // testStrListener更新時 に subscribe 設定
    useEffect(() => {

        // 初回表示時はオブジェクトが空なのでスキップ
        if (testStrListener === null) { return; }

        // Test string topic msg subscription event
        testStrListener.subscribe((msg: ROSLIB.Message) => {
            // 上位モジュールへ通知
            props.updateTestStr?.((msg as String).data);
        });
    }, [testStrListener]);

    return (
        <>< />    // 表示は無し
    );
});
RosOperator.displayName = 'RosOperator';
export default RosOperator;

// React
import React, { useRef, useState } from 'react';

// Bootstrap
import { Container, Row, Col } from 'react-bootstrap';
import { Button, Form } from 'react-bootstrap';
import 'bootstrap/dist/css/bootstrap.min.css';

// ROS Operator
import  RosOperator from '../libs/ros_operation';


//////////////////////////////////////////////////
/// Viewer                                                                        ///
//////////////////////////////////////////////////

const Viewer = () => {

    // ROS operator ref
    const rosOpRef = useRef<any>(null!);

    // ROS 接続状態 (ROS Operatorにて更新)
    const [rosConnected, setRosConnected] = useState<boolean>(false);

    // Test string
    const [testStr, setTestStr] = useState<string>('');

    // ROS接続開始
    const startConnect = () => {
        rosOpRef.current.connect();
    };

    // 表示
    return (
        <>
            {/* ROS Operator */}
            <RosOperator ref={rosOpRef}
                setRosConnected={setRosConnected}
                updateImage={updateImage}
                updatePointCloud={updatePointCloud}
                updateTestStr={setTestStr}
            />

            <Container>
                {/* Title */}
                <Row>
                    <Col>
                        <h1>ROS2 data visualization sample</h1>
                    </Col>
                </Row>

                {/* Header ... ROS接続ボタンおよび接続状態確認 */}
                <Row>
                    <Col>
                        <Button 
                            variant={!rosConnected ? "primary" : "secondary"}
                            disabled={rosConnected}
                            onClick={!rosConnected ? startConnect : null}
                        >
                            ROS接続開始
                        </Button>
                        <h2>ROS接続状態 ... {rosConnected ? 'ON' : 'OFF'}</h2>
                    </Col>
                </Row>

                {/* Logger ... ログ表示等 */}
                <Row>
                    <Col>
                        <h2>ログ表示</h2>
                        <div>{testStr}</div>
                    </Col>
                </Row>
            </Container>
        </>
    );
};
export default Viewer;