github.com
VarSpeedServoを使おうとしたが現時点では使えないらしい。
github.com
Arduino Nano Everyを買うときは気を付けましょう。
Arduinoを使ってボールを追いかける車を作った
はじめに
カメラ画像をWi-FiでPCに送信。
PCでカメラ画像からボールの位置を確認、XBeeを使ってArduinoにモータを動かす指示を送信。
上記を繰り返すことによってボールを追いかけることができた。
ボールの認識はMXNetの物体検出モデルを使用した。（こちらを参照）
購入したもの
Arduino Uno R3（The Arduino Starter Kitとして購入） Hブリッジモータードライバ [L293DNE] （The Arduino Starter Kitに付属） サンハヤト SAD-101 ニューブレッドボード XBee ZB（S2C）ワイヤアンテナ型 ×2 XBee エクスプローラ USB （Sparkfun Electronics） Arduino用XBee スタック可能シールド V2 （SeeedStudio） APEMAN A79 アクションカメラ Hitecコネクタ付き、6Vニッケル水素2000mAhバッテリーパック（モーター用に購入） cheero Canvas 3200mAh IoT機器対応 モバイルバッテリー（Arduino用に購入） 変換名人 USB A(オス) -USB B(オス)変換ケーブル [ 約20cm ] タミヤ ロングユニバーサルアームセット （オレンシﾞ） タミヤ ユニバーサルプレート用スライドアダプター タミヤ スポーツタイヤセット （56mm径） タミヤ ボールキャスター （2セット入） タミヤ ユニバーサルプレート （2枚セット） タミヤ ダブルギヤボックス（左右独立4速タイプ） タミヤ 1.5A 平行コード （5m） ダイソー すべりどめシート 適当なボール
- Arduinoはユニバーサルプレート用スライドアダプターを使ってユニバーサルプレートに固定した。適当なスペーサーがあればスライドアダプターはいらないと思う。
- ブレッドボードは両面テープでユニバーサルプレートに張り付けた。
- バッテリー類はすべりどめシートを下に敷いて置いただけ。平地を走行する分にはこれで十分であった。
- APEMAN A79のカバーをユニバーサルプレートにボンドで固定した。
- タミヤ 1.5A 平行コード （5m）の商品説明には「芯線が30本と多く、ハンダを使わずに手でひねった結線でも接触不良になりにくいのもポイントです。」と書かれている。それを信じてはんだ付けはしていないが今のところ問題ない。そもそもはんだごてを持っていない（笑）。
Pythonコード
import mxnet as mx import gluoncv import serial, time import cv2, queue, threading class VideoCapture: def __init__(self, name): self.cap = cv2.VideoCapture(name) self.q = queue.Queue() t = threading.Thread(target=self._reader) t.daemon = True t.start() # read frames as soon as they are available, keeping only most recent one def _reader(self): while True: ret, frame = self.cap.read() if not ret: break if not self.q.empty(): try: self.q.get_nowait() # discard previous (unprocessed) frame except queue.Empty: pass self.q.put(frame) def read(self): return self.q.get() def motion(r_PWM, l_PWM, delay_time): a = r_PWM.to_bytes(1, 'big') b = l_PWM.to_bytes(1, 'big') c = delay_time.to_bytes(1, 'big') ser.write(a) ser.write(b) ser.write(c) ser =serial.Serial("COM4", 9600) time.sleep(2) ctx = mx.gpu() # Load the model classes = ['ball'] net = gluoncv.model_zoo.get_model('ssd_512_mobilenet1.0_custom', classes=classes, pretrained=False, root='./model') net.load_parameters('ssd_512_mobilenet1.0_ball.params') net.collect_params().reset_ctx(ctx) # Compile the model for faster speed net.hybridize() cap = VideoCapture('rtsp://192.72.1.1:554/liveRTSP/av4/track0') time.sleep(1) #最初にストップした状態を送る motion(0,0,100) while True: frame = cap.read() frame = mx.nd.array(cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)).astype('uint8') rgb_nd, frame = gluoncv.data.transforms.presets.ssd.transform_test(frame, short=320) # Run frame through network class_IDs, scores, bounding_boxes = net(rgb_nd.as_in_context(ctx)) if ser.in_waiting != 0: ser.reset_input_buffer() if scores[0][0] > 0.55: x_min = bounding_boxes[0][0][0] y_min = bounding_boxes[0][0][1] x_max = bounding_boxes[0][0][2] y_max = bounding_boxes[0][0][3] width = x_max - x_min height = y_max - y_min if max(width, height) < 135: #右旋回 if x_min > 360: l_PWM = 150 if x_min > 480: r_PWM = 90 else: r_PWM = 120 motion(r_PWM, l_PWM, 50) #左旋回 elif x_max < 280: r_PWM = 150 if x_max <160: l_PWM = 90 else: l_PWM = 120 motion(r_PWM, l_PWM, 50) #前進 else: r_PWM = 150 l_PWM = 150 motion(r_PWM, l_PWM, 50) else: motion(0, 0, 25) #ボールが近いのでストップ else: motion(0, 0, 25) #ボールを見失ったのでストップ # Display the result img = gluoncv.utils.viz.cv_plot_bbox(frame, bounding_boxes[0], scores[0], class_IDs[0], class_names=net.classes, thresh=0.55) gluoncv.utils.viz.cv_plot_image(img) # escを押したら終了 if cv2.waitKey(1) == 27: break ser.close() cv2.destroyAllWindows()
Arduinoスケッチ
const int PIN_1A = 4; const int PIN_2A = 5; const int PIN_3A = 6; const int PIN_4A = 7; const int motor_right = 9; const int motor_left = 10; int delay_time; int right_pwm; int left_pwm; void setup() { pinMode(PIN_1A, OUTPUT); pinMode(PIN_2A, OUTPUT); pinMode(PIN_3A, OUTPUT); pinMode(PIN_4A, OUTPUT); pinMode(motor_right, OUTPUT); pinMode(motor_left, OUTPUT); Serial.begin(9600); //right_wheel前進 digitalWrite(PIN_1A, LOW); digitalWrite(PIN_2A, HIGH); //right_wheel前進 digitalWrite(PIN_3A, HIGH); digitalWrite(PIN_4A, LOW); } void loop() { if(Serial.available()>2){ right_pwm = Serial.read(); left_pwm = Serial.read(); delay_time = Serial.read()*4; analogWrite(motor_right, right_pwm); analogWrite(motor_left, left_pwm); delay(delay_time); Serial.write(255); } }
Arduino関連の記事
2020-05-16ArduinoとPython
2020-05-17カメラに人が写り込んだらLEDを光らせる
2020-05-20Arduinoでサーボを動かす（Pythonとのシリアル通信）
2020-05-21Arduinoで複数のサーボを動かす（Pythonとのシリアル通信）
2020-05-26ArduinoとPythonのシリアル通信を無線化する（XBee）
2020-05-28人がいる方向をむく（iPhone）
2020-06-02人がいる方向をむく（アクションカメラ APEMAN A79）
2020-06-05人がいる方向をむく（アクションカメラ APEMAN A79）OpenCV-Python
2020-06-11ボールがある方向を向く（パン＆チルト）
2020-06-25Arduinoを使ってボールを追いかける車を作った
MXNetで深層強化学習（A2CでCartPole-v0）
参考にさせて頂いたサイト
環境
Windows10 Pro GPUなし Python 3.8.2
バージョン情報
インストールしたのは「mxnet」「matplotlib」「gym」のみ。
pip install mxnet pip install matplotlib pip install gym
その他は勝手についてくる。
certifi==2020.6.20 chardet==3.0.4 cloudpickle==1.3.0 cycler==0.10.0 future==0.18.2 graphviz==0.8.4 gym==0.17.2 idna==2.6 kiwisolver==1.2.0 matplotlib==3.2.2 mxnet==1.6.0 numpy==1.16.6 pyglet==1.5.0 pyparsing==2.4.7 python-dateutil==2.8.1 requests==2.18.4 scipy==1.5.0 six==1.15.0 urllib3==1.22
強化学習のPythonコード
上記GitHubから「A2C.py」を持ってきた。
max_episodesを400から100に変更。
import gym import numpy as np import matplotlib.pyplot as plt import mxnet as mx from mxnet import gluon, nd, autograd, init from mxnet.gluon import loss as gloss, nn class Actor(nn.Block): def __init__(self, action_dim): super(Actor, self).__init__() self.action_dim = action_dim self.dense0 = nn.Dense(400, activation='relu') self.dense1 = nn.Dense(300, activation='relu') self.dense2 = nn.Dense(self.action_dim) def forward(self, state): _ = self.dense2(self.dense1(self.dense0(state))) probs = nd.softmax(_, axis=1) return probs class Critic(nn.Block): def __init__(self): super(Critic, self).__init__() self.dense0 = nn.Dense(400, activation='relu') self.dense1 = nn.Dense(300, activation='relu') self.dense2 = nn.Dense(1) def forward(self, state): v_values = self.dense2(self.dense1(self.dense0(state))) return v_values class A2C(object): def __init__(self, gamma, action_dim, observation_dim, ctx): self.gamma = gamma self.action_dim = action_dim self.observation_dim = observation_dim self.ctx = ctx self.actor_network = Actor(self.action_dim) self.critic_network = Critic() self.actor_network.initialize(init=init.Xavier(), ctx=self.ctx) self.critic_network.initialize(init=init.Xavier(), ctx=self.ctx) self.actor_optimizer = gluon.Trainer(self.actor_network.collect_params(), 'adam') self.critic_optimizer = gluon.Trainer(self.critic_network.collect_params(), 'adam') self.states = [] self.actions = [] self.rewards = [] self.dones = [] self.next_states = [] self.total_reward = [] def compute_returns(self, next_return): r = next_return self.total_reward = [0] * len(self.rewards) for step in reversed(range(len(self.rewards))): r = self.rewards[step] + self.gamma * r * (1 - self.dones[step]) self.total_reward[step] = r def store_transition(self, state, action, reward, done, next_state): self.states.append(state) self.actions.append(action) self.rewards.append(reward) self.dones.append(done) self.next_states.append(next_state) def choose_action(self, state): state = nd.array([state], ctx=self.ctx) all_action_prob = self.actor_network(state) action = int(nd.sample_multinomial(all_action_prob).asnumpy()) return action def update(self): states = nd.array(self.states, ctx=self.ctx) actions = nd.array(self.actions, ctx=self.ctx) total_reward = nd.array(self.total_reward, ctx=self.ctx) # ------------optimize actor----------- with autograd.record(): values = self.critic_network(states) probs = self.actor_network(states) advantages = (total_reward - values).detach() loss = -nd.pick(probs, actions).log() * advantages self.actor_network.collect_params().zero_grad() loss.backward() self.actor_optimizer.step(batch_size=len(states)) # -----------optimize critic------------ with autograd.record(): values = self.critic_network(states) l2_loss = gloss.L2Loss() loss = l2_loss(values, total_reward) self.critic_network.collect_params().zero_grad() loss.backward() self.critic_optimizer.step(batch_size=len(states)) self.states = [] self.actions = [] self.rewards = [] self.dones = [] self.next_states = [] self.total_reward = [] def save(self): self.actor_network.save_parameters('A2C_CartPole_actor_network.params') self.critic_network.save_parameters('A2C_CartPole_critic_network.params') def load(self): self.actor_network.load_parameters('A2C_CartPole_actor_network.params') self.critic_network.load_parameters('A2C_CartPole_critic_network.params') if __name__ == '__main__': seed = 77777777 np.random.seed(seed) mx.random.seed(seed) env = gym.make('CartPole-v0').unwrapped env.seed(seed) ctx = mx.cpu() render = False agent = A2C(gamma=0.99, action_dim=env.action_space.n, observation_dim=env.observation_space.shape[0], ctx=ctx) episode_reward_list = [] max_episodes = 100 max_episode_steps = 500 for episode in range(max_episodes): state = env.reset() episode_reward = 0 for episode_step in range(max_episode_steps): if render: env.render() action = agent.choose_action(state) next_state, reward, done, info = env.step(action) if done: reward = -1 agent.store_transition(state, action, reward, done, next_state) episode_reward += reward if done: break state = next_state print('episode %d ends with reward %d' % (episode, episode_reward)) episode_reward_list.append(episode_reward) agent.compute_returns(reward) agent.update() agent.save() env.close() plt.plot(episode_reward_list) plt.xlabel('episode') plt.ylabel('episode reward') plt.title('A2C_CartPole_v0') plt.savefig('./A2C_CartPole_v0.png') plt.show()
結果の確認
- 学習後
import mxnet as mx from A2C import Actor import gym env = gym.make('CartPole-v0') agent = Actor(env.action_space.n) agent.load_parameters('A2C_CartPole_actor_network.params') for i in range(10): observation = env.reset() for t in range(300): #env.render() state = mx.nd.array([observation]) all_action_prob = agent(state) action = int(mx.nd.sample_multinomial(all_action_prob).asnumpy()) observation, reward, done, info = env.step(action) if done: print("Episode{} finished after {} timesteps".format(i, t+1)) break env.close()
Episode0 finished after 200 timesteps Episode1 finished after 200 timesteps Episode2 finished after 200 timesteps Episode3 finished after 200 timesteps Episode4 finished after 200 timesteps Episode5 finished after 200 timesteps Episode6 finished after 200 timesteps Episode7 finished after 200 timesteps Episode8 finished after 200 timesteps Episode9 finished after 200 timesteps
MXNetで深層強化学習（Double_DQNでCartPole-v0）
「電子工作×深層学習」をテーマとした書籍である。
やはり強化学習の知識は欠かせない。
今回はMXNetで強化学習をしてみた。（GitHubのコードを動かしただけ）
参考にさせて頂いたサイト
環境
Windows10 Pro GPUなし Python 3.8.2
バージョン情報
インストールしたのは「mxnet」「matplotlib」「gym」のみ。
pip install mxnet pip install matplotlib pip install gym
その他は勝手についてくる。
certifi==2020.6.20 chardet==3.0.4 cloudpickle==1.3.0 cycler==0.10.0 future==0.18.2 graphviz==0.8.4 gym==0.17.2 idna==2.6 kiwisolver==1.2.0 matplotlib==3.2.2 mxnet==1.6.0 numpy==1.16.6 pyglet==1.5.0 pyparsing==2.4.7 python-dateutil==2.8.1 requests==2.18.4 scipy==1.5.0 six==1.15.0 urllib3==1.22
強化学習のPythonコード
上記GitHubから「Double_DQN.py」を持ってきた。
episodesを400から360に変更。
あらかじめ「utils.py」と実行コードを同じフォルダに入れておく。
import random import gym import numpy as np import matplotlib.pyplot as plt import mxnet as mx from mxnet import gluon, nd, autograd, init from mxnet.gluon import loss as gloss from utils import MemoryBuffer class DoubleQNetwork(gluon.nn.Block): def __init__(self, n_action): super(DoubleQNetwork, self).__init__() self.n_action = n_action self.dense0 = gluon.nn.Dense(400, activation='relu') self.dense1 = gluon.nn.Dense(300, activation='relu') self.dense2 = gluon.nn.Dense(self.n_action) def forward(self, state): q_value = self.dense2(self.dense1(self.dense0(state))) return q_value class DoubleDQN: def __init__(self, n_action, init_epsilon, final_epsilon, gamma, buffer_size, batch_size, replace_iter, annealing, learning_rate, ctx ): self.n_action = n_action self.epsilon = init_epsilon self.init_epsilon = init_epsilon self.final_epsilon = final_epsilon # discount factor self.gamma = gamma # memory buffer size self.buffer_size = buffer_size self.batch_size = batch_size # replace the parameters of the target network every T time steps self.replace_iter = replace_iter # The number of step it will take to linearly anneal the epsilon to its min value self.annealing = annealing self.learning_rate = learning_rate self.ctx = ctx self.total_steps = 0 self.replay_buffer = MemoryBuffer(self.buffer_size, ctx) # use deque # build the network self.target_network = DoubleQNetwork(n_action) self.main_network = DoubleQNetwork(n_action) self.target_network.collect_params().initialize(init.Xavier(), ctx=ctx) # initialize the params self.main_network.collect_params().initialize(init.Xavier(), ctx=ctx) # optimize the main network self.optimizer = gluon.Trainer(self.main_network.collect_params(), 'adam', {'learning_rate': self.learning_rate}) def choose_action(self, state): state = nd.array([state], ctx=self.ctx) if nd.random.uniform(0, 1) > self.epsilon: # choose the best action q_value = self.main_network(state) action = int(nd.argmax(q_value, axis=1).asnumpy()) else: # random choice action = random.choice(range(self.n_action)) # anneal self.epsilon = max(self.final_epsilon, self.epsilon - (self.init_epsilon - self.final_epsilon) / self.annealing) self.total_steps += 1 return action def update(self): state_batch, action_batch, reward_batch, next_state_batch, done_batch = self.replay_buffer.sample( self.batch_size) with autograd.record(): # get the Q(s,a) all_current_q_value = self.main_network(state_batch) main_q_value = nd.pick(all_current_q_value, action_batch) # different from DQN # get next action from main network, then get its Q value from target network all_next_q_value = self.target_network(next_state_batch).detach() # only get gradient of main network max_action = nd.argmax(all_current_q_value, axis=1) target_q_value = nd.pick(all_next_q_value, max_action).detach() target_q_value = reward_batch + (1 - done_batch) * self.gamma * target_q_value # record loss loss = gloss.L2Loss() value_loss = loss(target_q_value, main_q_value) self.main_network.collect_params().zero_grad() value_loss.backward() self.optimizer.step(batch_size=self.batch_size) def replace_parameters(self): self.main_network.save_parameters('Double_DQN_temp_params') self.target_network.load_parameters('Double_DQN_temp_params') print('Double_DQN parameters replaced') def save_parameters(self): self.target_network.save_parameters('Double_DQN_target_network_parameters') self.main_network.save_parameters('Double_DQN_main_network_parameters') def load_parameters(self): self.target_network.load_parameters('Double_DQN_target_network_parameters') self.main_network.load_parameters('Double_DQN_main_network_parameters') if __name__ == '__main__': seed = 7777777 mx.random.seed(seed) random.seed(seed) np.random.seed(seed) ctx = mx.cpu() env = gym.make('CartPole-v0').unwrapped env.seed(seed) render = False episodes = 360 agent = DoubleDQN(n_action=env.action_space.n, init_epsilon=1, final_epsilon=0.1, gamma=0.99, buffer_size=3000, batch_size=32, replace_iter=1000, annealing=3000, learning_rate=0.0001, ctx=ctx ) episode_reward_list = [] for episode in range(episodes): state = env.reset() episode_reward = 0 while True: if render: env.render() action = agent.choose_action(state) next_state, reward, done, info = env.step(action) episode_reward += reward agent.replay_buffer.store_transition(state, action, reward, next_state, done) if agent.total_steps > 1000: agent.update() if agent.total_steps % agent.replace_iter == 0: agent.replace_parameters() if done: print('episode %d ends with reward %d at steps %d' % (episode, episode_reward, agent.total_steps)) episode_reward_list.append(episode_reward) break state = next_state agent.save_parameters() env.close() plt.plot(episode_reward_list) plt.xlabel('episode') plt.ylabel('episode reward') plt.title('Double_DQN CartPole-v0') plt.savefig('./Double-DQN-CartPole-v0.png') plt.show()
結果の確認
- 未学習
import gym env = gym.make('CartPole-v0') for i in range(10): observation = env.reset() for t in range(300): #env.render() observation, reward, done, info = env.step(env.action_space.sample()) if done: print("Episode{} finished after {} timesteps".format(i, t+1)) break env.close()
Episode0 finished after 19 timesteps Episode1 finished after 25 timesteps Episode2 finished after 11 timesteps Episode3 finished after 19 timesteps Episode4 finished after 15 timesteps Episode5 finished after 37 timesteps Episode6 finished after 22 timesteps Episode7 finished after 11 timesteps Episode8 finished after 13 timesteps Episode9 finished after 17 timesteps
- 学習後
import mxnet as mx from Double_DQN import DoubleQNetwork import gym ctx = mx.cpu() env = gym.make('CartPole-v0') agent = DoubleQNetwork(env.action_space.n) agent.load_parameters('Double_DQN_main_network_parameters') for i in range(10): observation = env.reset() for t in range(300): #env.render() state = mx.nd.array([observation]) action = int(mx.nd.argmax(agent(state), axis=1).asnumpy()) observation, reward, done, info = env.step(action) if done: print("Episode{} finished after {} timesteps".format(i, t+1)) break env.close()
Episode0 finished after 200 timesteps Episode1 finished after 200 timesteps Episode2 finished after 200 timesteps Episode3 finished after 200 timesteps Episode4 finished after 200 timesteps Episode5 finished after 200 timesteps Episode6 finished after 194 timesteps Episode7 finished after 199 timesteps Episode8 finished after 200 timesteps Episode9 finished after 200 timesteps
PyTorchの「3D human pose estimation」を試してみる
環境
Windows10 Pro GPUなし Python 3.8.2
バージョン確認（pip freeze）
インストールしたのは「torch」「torchvision」「opencv-python」「matplotlib」のみ。
pip install torch==1.5.0+cpu torchvision==0.6.0+cpu -f https://download.pytorch.org/whl/torch_stable.html pip install opencv-python pip install matplotlib
その他は勝手についてくる。
cycler==0.10.0 future==0.18.2 kiwisolver==1.2.0 matplotlib==3.2.1 numpy==1.18.5 opencv-python==4.2.0.34 Pillow==7.1.2 progress==1.5 pyparsing==2.4.7 python-dateutil==2.8.1 six==1.15.0 torch==1.5.0+cpu torchvision==0.6.0+cpu
学習済みモデルのダウンロード
GitHubに示されている通りにGoogle Driveからダウンロードしてくる。
実行
用意されたサンプル画像
python ./src/demo.py --demo ./images/mpi_inf_3dhp_706.png --gpus -1 --load_model ./models/fusion_3d_var.pth
結果
2つの画像が出力される。
2番目の画像はグリグリ回転させることができる。
面白い動画を見つけた（SegmentationとPose Estimateion）
面白い動画を見つけた。
2011年に公開されたものである。
Segmentation, Pose Estimateion の要素が入っているが2011年の時点でどうやってこれを成し遂げたのだろう。
https://www.youtube.com/watch?v=1AiGsAOH-Es
計算問題を解いてダイエット
2桁×2桁の掛け算をただ単に繰り返し暗算してダイエットしようと考えた。
自分の限界はここまでで3桁になると暗算はムリ。
体動かすのもいいが脳をフル回転すればカロリーは消費されるはず。
問題と答えは画面に表示させるのではなく読み上げとした。
音源は以下のサイトからダウンロードさせて頂いた。
www14.big.or.jp
import os cd = os.getcwd() os.add_dll_directory('C:/Program Files/VideoLAN/VLC') import vlc os.chdir(os.path.join(cd, 'sound')) import random def make_sound(num): sound_list = [] num_str = str(num).zfill(4) if not num_str[0]== '0': sound_list.append(num_str[0].ljust(4, '0') + '.wav') if not num_str[1]== '0': sound_list.append(num_str[1].ljust(3, '0') + '.wav') if not num_str[2]== '0': sound_list.append(num_str[2].ljust(2, '0') + '.wav') if not num_str[3]== '0': sound_list.append(num_str[3] + '.wav') return sound_list while True: q1 = random.randint(10,99) q2 = random.randint(10,99) answer = q1 * q2 q1_list = make_sound(q1) q2_list = make_sound(q2) sound_list_q = q1_list + ['kakeru.wav'] + q2_list sound_list_a = make_sound(answer) player = vlc.MediaListPlayer() mediaList = vlc.MediaList(sound_list_q) player.set_media_list(mediaList) player.play() get_key = input() if get_key == 'q': break player = vlc.MediaListPlayer() mediaList = vlc.MediaList(sound_list_a) player.set_media_list(mediaList) player.play() get_key = input() if get_key == 'q': break