【シストレ開発#2】ROS2の独自メッセージ型を作成してみた！

$ ros2 pkg create --build-type ament_cmake oanda_api_msgs

# Pricing Definitions of OANDA-API.
# Reference:
#    https://developer.oanda.com/rest-live-v20/pricing-df/
#    https://oanda-api-v20.readthedocs.io/en/latest/endpoints/pricing/pricingstream.html

# The date/time when the Price was created
string time

# The Price’s Instrument.
string instrument

# The closeout bid Price.
float32 closeout_bid

# The closeout ask Price.
float32 closeout_ask

# Flag indicating if the Price is tradeable or not
bool tradeable

<?xml version="1.0"?>
<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
<package format="3">
  <name>oanda_api_msgs</name>
  <version>0.1.0</version>
  <description>
    oanda_api msg package.
  </description>
  <maintainer email="takkin.takilog@gmail.com">takkin</maintainer>
  <license>Apache License 2.0</license>

  <buildtool_depend>ament_cmake</buildtool_depend>
  <buildtool_depend>rosidl_default_generators</buildtool_depend>

  <exec_depend>std_msgs</exec_depend>
  <exec_depend>action_msgs</exec_depend>
  <exec_depend>rosidl_default_runtime</exec_depend>

  <member_of_group>rosidl_interface_packages</member_of_group>

  <export>
    <build_type>ament_cmake</build_type>
  </export>
</package>

13:  <buildtool_depend>rosidl_default_generators</buildtool_depend>

17:  <exec_depend>rosidl_default_runtime</exec_depend>

19:  <member_of_group>rosidl_interface_packages</member_of_group>

cmake_minimum_required(VERSION 3.5)
project(oanda_api_msgs)

set(CMAKE_CXX_STANDARD 14)

if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
  add_compile_options(-Wall -Wextra -Wpedantic)
endif()

# find dependencies
find_package(ament_cmake REQUIRED)
find_package(std_msgs REQUIRED)
find_package(action_msgs REQUIRED)
find_package(rosidl_default_generators REQUIRED)

rosidl_generate_interfaces(${PROJECT_NAME}
  "msg/Pricing.msg"
  DEPENDENCIES std_msgs action_msgs
)

ament_export_dependencies(rosidl_default_runtime)

ament_package()

import rclpy
from rclpy.node import Node
from std_msgs.msg import Bool
from oanda_api_msgs.msg import Pricing
from oandapyV20 import API
from oandapyV20.endpoints.pricing import PricingStream
from oandapyV20.exceptions import V20Error, StreamTerminated


class StreamApi(Node):

    def __init__(self):
        super().__init__("stream_api")

        # Set logger lebel
        self.__logger = super().get_logger()
        self.__logger.set_level(rclpy.logging.LoggingSeverity.DEBUG)

        PRMNM_ACCOUNT_NUMBER = "account_number"
        PRMNM_ACCESS_TOKEN = "access_token"
        PRMNM_INSTRUMENTS = "instruments"
        TPCNM_PRICING = "pricing_"
        TPCNM_ACT_FLG = "activate_flag"

        # Declare parameter
        self.declare_parameter(PRMNM_ACCOUNT_NUMBER)
        self.declare_parameter(PRMNM_ACCESS_TOKEN)
        self.declare_parameter(PRMNM_INSTRUMENTS)

        # initialize
        self.__act_flg = True

        account_number = self.get_parameter(PRMNM_ACCOUNT_NUMBER).value
        access_token = self.get_parameter(PRMNM_ACCESS_TOKEN).value
        instrumentslist = self.get_parameter(PRMNM_INSTRUMENTS).value
        self.__logger.debug("[OANDA]Account Number:%s" % account_number)
        self.__logger.debug("[OANDA]Access Token:%s" % access_token)
        self.__logger.debug("[OANDA]Instruments:%s" % instrumentslist)

        self.__api = API(access_token=access_token)
        instruments = ",".join(instrumentslist)
        params = {"instruments": instruments}
        self.__ps = PricingStream(account_number, params)

        # Declare publisher and subscriber
        self.__pub_dict = {}
        for instrument in instrumentslist:
            suffix = instrument.replace("_", "").lower()
            pub = self.create_publisher(Pricing, TPCNM_PRICING + suffix)
            self.__pub_dict[instrument] = pub.publish

        self.__sub_act = self.create_subscription(Bool, TPCNM_ACT_FLG,
                                                  self.__act_flg_callback)

    def background(self):
        if self.__act_flg:
            self.__request()

    def __act_flg_callback(self, msg):
        if msg.data:
            self.__act_flg = True
        else:
            self.__act_flg = False

    def __request(self):
        TIME = "time"
        INSTRUMENT = "instrument"
        BID = "closeoutBid"
        ASK = "closeoutAsk"
        TRADEABLE = "tradeable"

        try:
            for rsp in self.__api.request(self.__ps):

                rclpy.spin_once(self, timeout_sec=0)
                if not self.__act_flg:
                    self.__logger.debug("terminate PricingStream")
                    self.__ps.terminate()

                if TRADEABLE in rsp.keys():
                    msg = Pricing()
                    msg.time = rsp[TIME]
                    msg.instrument = rsp[INSTRUMENT]
                    msg.closeout_bid = float(rsp[BID])
                    msg.closeout_ask = float(rsp[ASK])
                    msg.tradeable = rsp[TRADEABLE]

                    # Publish topics
                    self.__pub_dict[msg.instrument](msg)

        except V20Error as e:
            print("Error: {}".format(e))
        except StreamTerminated as e:
            print("StreamTerminated: {}".format(e))

    def __handler(self, func, msg):
        func(msg)


def main(args=None):
    rclpy.init(args=args)
    stream_api = StreamApi()
    try:
        while True:
            rclpy.spin_once(stream_api, timeout_sec=1.0)
            stream_api.background()
    except KeyboardInterrupt:
        stream_api.destroy_node()
        rclpy.shutdown()

 4:    from oanda_api_msgs.msg import Pricing

81:    msg = Pricing()
82:    msg.time = rsp[TIME]
83:    msg.instrument = rsp[INSTRUMENT]
84:    msg.closeout_bid = float(rsp[BID])
85:    msg.closeout_ask = float(rsp[ASK])
86:    msg.tradeable = rsp[TRADEABLE]

stream_api:
    ros__parameters:
        account_number: "999-999-9999999-999"
        access_token: "zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz-zzzzzzzzzzzzzzzzzzzzzzzzzzzzzzzz"
        instruments: ["USD_JPY", "EUR_JPY", "EUR_USD"]

$ ros2 run oanda_api price_stream_publisher __params:=ファイルの絶対パス/oanda_params.yaml