Jupyter notebook ならば、 ipywidgets と IPython.display で処理しつつ、結果を動画としてモニタすることが可能
こんな感じ
%% bash
git clone https://github.com/NVIDIA-AI-IOT/jetcam.git
cd jetcam
pip3 install -e .
import ipywidgets
from IPython.display import display
from jetcam.utils import bgr8_to_jpeg
import gym
env = gym.make('Pong-v0') # GUI環境の開始(***)
img = env.reset()
# img = env.render(mode="rgb_array")
image_widget = ipywidgets.Image(format='jpeg')
image_widget.value = bgr8_to_jpeg(img)
display(image_widget)
for episode in range(20):
observation = env.reset() # 環境の初期化
for _ in range(100):
# img = env.render(mode = 'rgb_array')
action = env.action_space.sample() # 行動の決定
observation, reward, done, info = env.step(action)
image_widget.value = bgr8_to_jpeg(observation[:,:,::-1])
おお、これなら Colab でも、って期待したんだけど、 Colab は jupyter notebook とは仕様が違うらしくて、動いてくれない。
gym-notebook-wrapper も試してみたのだけれど、いずれもあまり速くなく、自分で作ったラッパーをかけると動かないケースが多い。
最後（かどうかわからんけど）に行き着いた方法をメモしておく。
参考にしたのは、webCamGoogleColabという、Webcam のモニタ画面を Colab 上に表示する方法。
方法を２つ公開してくださっていて、ひとつは Colab だけで閉じた方法、もう一つは ngrok の転送を使う方法。意外なことに、後者の方が速い。
WebCam の画面と、それをリアルタイムに処理した結果を表示するようなサンプルを書いてくださっているのだが、処理結果の方を自分のプログラムの処理結果に置き換えて、カメラ入力の画面を表示しなければいいんじゃないか、というのがアイデア。
それができるんなら、きっと転送も不要なのではないかと思うし、無駄なコードが残っているに違いないんだけど、目的は達成できているのでよしとして、記録に残す。
wget -q https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip > /dev/null
unzip ngrok-stable-linux-amd64.zip
pip install bottle > /dev/null
pip install bottle_websocket > /dev/null
カーネルを起動したあとでインストールしたライブラリは機能しないので、ここで一旦ランタイムをリセットする。
from IPython.display import display, Javascript
from google.colab.output import eval_js
from time import sleep
def game_mon(url, quality=0.8):
sleep(3)
print("start monitor") # このプリント文は必須
js = Javascript('''
async function gameMon(url, quality) {
const div = document.createElement('div');
document.body.appendChild(div);
const canvas = document.createElement('canvas');
canvas.width = 240;
canvas.height = 315;
const canvasCtx = canvas.getContext('2d');
div.appendChild(canvas);
//exit button （効かないです）
const btn_div = document.createElement('div');
document.body.appendChild(btn_div);
const exit_btn = document.createElement('button');
exit_btn.textContent = 'Exit';
var exit_flg = false
exit_btn.onclick = function() {exit_flg = true};
btn_div.appendChild(exit_btn);
//log
let jsLog = function(abc) {
document.querySelector("#output-area").appendChild(document.createTextNode(`${abc} `));
}
google.colab.output.setIframeHeight(document.documentElement.scrollHeight, true);
//for websocket connection.
var connection = 0
var flag_count = 0
var send_flg = false
// loop
_canvasUpdate();
jsLog("Connect_start");
async function _canvasUpdate() {
flag_count += 1
// 少し（1カウント）待ってから WebSocket を実行
if (flag_count == 10){
connection = new WebSocket(url);
jsLog("Connect_start");
}
if (flag_count == 100){
print("100");
connection.onmessage = function(e) {
var image = new Image()
image.src = e.data;
image.onload = function(){canvasCtx.drawImage(image, 0, 0, 160, 210, 0, 0, canvas.width, canvas.height)}
send_flg=false
};
jsLog("Set_recieve")
}
if(flag_count > 100){
const img = canvas.toDataURL('image/jpeg', quality);
if (send_flg==false){
connection.send(img);
send_flg = true
}
}
if (!exit_flg){
requestAnimationFrame(_canvasUpdate);
}else{
connection.close();
}
};
}
''')
display(js)
data = eval_js('gameMon("{}", {})'.format(url, quality))
画面サイズが、openAI gym[atari] 用に決め打ちになってて生の数値が埋め込まれていますが、実験的プログラムなので、ご勘弁。
ngrok の Web 公開用 url を取得。 サインイン不要。
get_ipython().system_raw('./ngrok http 6006 &')
"""
! curl -s http://localhost:4040/api/tunnels | python3 -c \
"import sys, json;url = json.load(sys.stdin)['tunnels'][0]['public_url'];print(url);f = open('url.txt', 'w');f.write(url);f.close()"
"""
import subprocess
url = "wss"
while len(url)==3:
result = subprocess.run(["curl -s http://localhost:4040/api/tunnels | python3 -c \"import sys, json; print(json.load(sys.stdin)['tunnels'][0]['public_url'])\""], encoding='utf-8',shell=True, stdout=subprocess.PIPE)
url = "wss"+result.stdout[5:-1]
url
参考にしたプログラムでは Linux シェルで ngrok の url を一旦テキストに書き出して、後で python で読み込んでいたが、全部 pythonで書いてみた。
import numpy as np
import cv2
import bottle
from bottle.ext.websocket import GeventWebSocketServer
from bottle.ext.websocket import websocket
from multiprocessing import Pool
import base64
import gym
env = gym.make('BattleZone-v0')
socket = bottle.Bottle()
@socket.route('/', apply=[websocket])
def wsbin(ws):
kbreak = False
while not kbreak:
done = False
env.reset()
while not done:
try:
action = env.action_space.sample() # ランダムな行動選択
out_img, reward, done, info = env.step(action) # １ステップ実行
if done:
break
#encode to string
_, encimg = cv2.imencode(".jpg", out_img, [int(cv2.IMWRITE_JPEG_QUALITY), 80])
img_str = encimg.tostring()
img_str = "data:image/jpeg;base64," + base64.b64encode(img_str).decode('utf-8')
ws.send(img_str)
except:
kbreak = True
done = True
pass
if __name__ == '__main__':
with Pool(2) as p:
p.apply_async(game_mon, (url, 0.8)) #
socket.run(host='0.0.0.0', port=5000, server=GeventWebSocketServer)
もともとが cam からの画像を表示するプログラムをなぞっているので、その名残が残っているが、ご容赦。