<![CDATA[
from rclpy.node import Node
import subprocess
import json
import numpy as np
from my_robot_msgs.msg import PositionStatus
from my_robot_msgs.msg import VisualStatus
from my_robot_msgs.msg import EarthStatus
from my_robot_msgs.msg import MotorStatus
from my_robot_msgs.msg import MotorCommand
class Store(Node):
def __init__(self):
super().__init__('store')
self.nn =
subprocess.Popen(
['/usr/bin/walking_brain_nn'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
text=True,
bufsize=1
)
self.position = None
self.visual = None
self.earth = None
self.motor = None
self.create_subscriber(
PositionStatus,
'/position/status',
self.position_callback,
10
)
self.create_subscriber(
VisualStatus,
'/vision/status',
self.visual_callback,
10
)
self.create_subscriber(
EarthStatus,
'/earth/status',
self.earth_callback,
10
)
self.create_subscriber(
MotorStatus,
'/motor/status',
self.motor_callback,
10
)
self.publisher = self.create_publisher(
MotorCommnad,
'/motor/command',
10
)
self.timer = self.create_timer(
0.01,
self.publish_motor_command
)
def position_callback(self, msg):
self.position = msg
def visual_callback(self, msg):
self.visual = msg
def earth_callback(self, msg):
self.earth = msg
def motor_callback(self, msg):
self.motor = msg
def walking_brain_nn(self, brain_input):
input_json = json.dumps(brain_input)
self.nn.stdin.write(input_json + "\n")
self.nn.stdin.flush()
output_json = self.nn.stdout.readline()
nn_output = json.loads(output_json)
return nn_output
def publish_motor_command(self):
if None in (self.position, self.visual, self.earth, self.motor):
return
brain_input = {
"position": self.position.__dict__,
"visual": self.visual.__dict__,
"earth": self.earth.__dict__,
"motor": self.motor.__dict__
}
nn_output = self.walking_brain_nn(brain_input)
current = np.argmax(nn_output, axis=1).astype(np.float32)
msg = MotorCommand()
msg.current = current.tolist()
self.publisher.publish(msg)
def main(args=None):
rclpy.init(args=args)
node = Store()
rclpy.spin(node)
node.destroy_node()
rclpy.shutdown()
if __name__ == '__main__':
main()
]]>
- プログラムの大枠
Python プログラムは，つぎの３段階構成：
# 類を設定
class Store(Node):
‥‥
# その類の物を自動させるコード
def main(args=None):
‥‥
node = Store()
rclpy.spin(node)
‥‥
# main() を実行するコード
if __name__ == '__main__':
main()
main() は，定型。
即ち，main() を定型にするのが Python の流儀。
プログラムにさせようとすることの記述は，類の設定のところで完結させる。
- 類の属性の設定
類の設定は，これの属性の設定。
属性の設定は，つぎの形式で行う：
class Store(Node):
def __init__(self):
super().__init__('brain')
self.属性名( 属性の内容 )
‥‥
ここで，"self" が出て来る。
"self.＊＊＊" は，｢self の属性＊＊＊｣ と読む。
属性は「何かの属性」となるものなので，その「何か｣ に self を措くというわけ。
実際，
class X :
属性 A
属性 B
と書けたとしたら，これから
class Y :
属性 A
が導かれる。
これは，
｢属性が類を定める｣
→ ｢属性が超越的に独自存在する｣
の存在論──転倒した存在論──になる。
- サブルーチン
"def __init__(self):" と並列の "def ‥‥ :" は，サブルーチン。
サブルーチンは，再帰処理を書く方法。
すべて，def __init__(self) の中から呼び出される。
- "super().__init__('store')"
この１行で何が起こるのかは，不明。
特に，'store’ がプログラムの中でどう効いてくるのか，不明。
２つのアンダーバーで挟む形は，"__init__" の他に
__dict__
__name__
__main__
が出て来る。
"__＊＊__" にはどんな意味/ロジックがあるのか？
- ライブラリ
プログラミングは，ライブラリの中にあるマクロを多用して，プログラムを短くする。
実際，これがプログラミングを省力化する方法。
import lrclpy
from rclpy.node import Node
は，つぎのコマンド：
Python の ROS2ライブラリ lrclpy を紐付けよ
rclpy.node の中のユーザ作成マクロ集 Node を紐付けよ
C だと，：
$ gcc main.c -L/lib -lrclpy
#include <Node.h>
となるところ。
プログラムの中の出処不明な文字列は，すべてライブラリのマクロ名。
特に，Python で Pub/Sub のデータ流通をプログラミングすることは，
lrclpy にどんなマクロがあるかを知っていて，
それらを使う
ことである。
例えば，
create_subscriber
create_publisher
はライブラリから引いているわけだが，
これを書くと何が起こるか
を知らないと，使えないわけである。
なお，
import subprocess
import json
は，＜チャイルドプロセスで NN を実行＞のため。
import numpy as np
は，＜NN の出力を出荷品に成形＞のため。
- "timer"
このプログラムのいちばん難解な箇所が，ここ：
self.timer = self.create_timer(
0.01,
self.publish_motor_command
)
"publish_motor_command" は，つぎの処理をパックしたもの：
入荷品から，NN の入力をつくる
NN に入力する
NN の出力を受ける
出力を出荷品に成形する
publish
ここで，0.01秒周期の timer がどう働くのかが不明。
（｢0.01秒周期の publish｣ の内容になっているのか？)
|