PymunkでPinJointを使う例
振り子運動のようなケースをシミュレーションしたい場合。 PymunkのExampleには複数の振り子を同時にシミュレーションする例(newton_cradle)があるが，単純に１つの場合の例。 初期のBodyとPinのアンカー配置で紐の長さが変わる。
import sys import pymunk import pymunk.pygame_util import pygame as pg from pygame.locals import QUIT, KEYDOWN, K_ESCAPE SCR_W, SCR_H = 200, 200 TICK = 100. def setup_space(): space = pymunk.Space() space.gravity = 0, -100 space.damping = 0.99 return space def setup_pygame(): # Initialize pygame screen for pymunk debug. pg.init() scr = pg.display.set_mode((SCR_W, SCR_H)) clk = pg.time.Clock() draw_options = pymunk.pygame_util.DrawOptions(scr) return scr, clk, draw_options def add_pinjoint_ball(space): R = 20 b = pymunk.Body(mass=1, moment=10) b.position = SCR_W/4, SCR_H/2 s = pymunk.Circle(b, radius=R) pin_joint = pymunk.PinJoint(space.static_body, b, (SCR_W/2, SCR_H), (0, 0)) space.add(b, s, pin_joint) return def run(): scr, clk, draw_options = setup_pygame() # Pygame setup space = setup_space() # Pymunk setup # Add objects add_pinjoint_ball(space) # Run simulation i = 0 while True: for e in pg.event.get(): if e.type == QUIT: sys.exit(0) elif e.type == KEYDOWN and e.key == K_ESCAPE: sys.exit(0) scr.fill((20, 20, 20)) space.debug_draw(draw_options) space.step(1/TICK) pg.display.flip() clk.tick(TICK) pg.image.save(scr, f'images/scr_{i:04}.png') i += 1 pg.quit() if __name__ == '__main__': run()
Pymunkメモ
準備
- 内部的にはChipmunk(Cベースの２次元物理エンジン)を使用している。Rubyなども公式サポートされている。
- 類似のライブラリとして(Py)Box2Dがあるが，ドキュメントはPymunkの方が良い印象。
- 画面表示にはPygame, Pyglet, Matplotlibが使える模様。
$ pip install pymunk
基礎知識
単位系は無い
- Pymunkでシミュレーションをする時にまず悩むのが，各性質の単位。例えば，massはgなのかkgなのか？など。でも，これは悩む必要はない。というのも，Pymunkでは単位は無い。自分で想定して設定するしか無い。バネなどの制約は実際にやってみながら値を設定することになる。
画面座標
- スクリーンサイズはW, Hの順で与える。NumpyなどはH,Wの順なのでちょっと注意が注意。
- 左下が(0, 0)，そこからxは右に+，yは上に+。Pygameは左上が(0,0)なので注意(ただし，debug_printでPygameを指定した場合は良きに変換してやってくれる）
BodyとShape
- Chipmunk(を含む多くの物理エンジン)の重要な構成要素は，剛体(Rigid body)，衝突形状(Collision shape), 関節/ジョイント(制約)，空間。はじめは，剛体に形状情報を含むと勘違いしてしまうが，剛体自体は質量，位置，回転向き，速度などからなり，形は含まない。
剛体の運動のシミュレーションでは，形は持たず，質量，位置，慣性などが理想的な状態を仮定してシミュレーションする。その際に利用されるのがBodyクラス。そして，それと並行してShapeクラスを用いて(衝突を含むシミュレーションでは)衝突判定を行う。
Bodyは６つの特徴から記述される。
- mass(質量)
- moment(慣性)
- position(位置)
- angle(向き)
- verocity(速度) *速度自体も向きを持つ
- angular_verocity(角速度)
Body(ボール)の作成例
大まかな流れは下記になる。
- Bodyを生成。この際に，質量だけでなくモーメントを設定する。
- 初期位置(position)を設定する。
- 衝突シミュレーションの必要があれば，形状shapeを設定する(絶対に必要？)。
- シミュレーション空間に追加。
単純なオブジェクトの場合には，予め定義された関数も用いてモーメントを設定可能。例えば，円の場合には，moment_for_circle。
b1 = pymunk.Body(mass=1, moment=10) b1.position = 100, 100 s1 = pymunk.Circle(b1, radius=20) space.add(b1, s1, ground
Vec2d
-2D空間上で位置(position)や向き(vector)を表すための専用クラスVec2dが提供されている。
ImpulseとForceの違い
- ２つのAPI(apply_forceとapply_impulse)がある。forceとimpulseの違いは，impulseは大きな力が短い時間に与えられるケース，例えばボールを打ったり，壁に衝突したり，キャノンから発射されたり，というようなケース。このようなケースでは速度が直接変化させられる。(逆に，forceは速度は直接変化させられずに，f=maに従って加速度を変化させるものだと想像される。)
Template
- 最も単純なケースとして，ボール１つが固定された地面に落ちる場合を載せておく。
- Pygameの画面を保存する方法はシミュレーションは遅くなるけれど，pygame.image.save(screen, name)でスクリーンを画像として保存して，あとでffmpeg, imagemagickなどでgif化するのが手っ取り早い。
import sys import pymunk import pymunk.pygame_util import pygame as pg from pygame.locals import QUIT, KEYDOWN, K_ESCAPE SCR_W, SCR_H = 400, 200 TICK = 100. def setup_space(): space = pymunk.Space() space.gravity = 0, -100 space.damping = 0.99 return space def setup_pygame(): # Initialize pygame screen for pymunk debug. pg.init() scr = pg.display.set_mode((SCR_W, SCR_H)) clk = pg.time.Clock() draw_options = pymunk.pygame_util.DrawOptions(scr) return scr, clk, draw_options def add_ground(space): b0 = space.static_body ground = pymunk.Segment(b0, (0, 0), (SCR_W, 0), 4) ground.elasticity = 1.0 space.add(ground) return def add_ball(space): b = pymunk.Body(mass=1, moment=10) b.position = 100, 100 b.apply_impulse_at_local_point((30, 80)) s = pymunk.Circle(b, radius=20) s.elasticity = 1.0 space.add(b, s) return def run(): # Initialize Pygame and Pymunk scr, clk, draw_options = setup_pygame() # Pygame setup space = setup_space() # Pymunk setup # Add objects add_ground(space) add_ball(space) # Run simulation t = 0 while True: for e in pg.event.get(): if e.type == QUIT: sys.exit(0) elif e.type == KEYDOWN and e.key == K_ESCAPE: sys.exit(0) scr.fill((20, 20, 20)) space.debug_draw(draw_options) space.step(1/TICK) pg.display.flip() clk.tick(TICK) pg.image.save(scr, f'./images/scr_{t:04}.png') t += 1 pg.quit() if __name__ == '__main__': run()
順列，組合せ，直積でのループ
順列数や組合せ数ではなく，順列，組わせでループしたい時がある。 意外と組むのが面倒だけれど，Pythonだととても便利なライブラリが標準で与えられている。 使い方は要素となるものをリストに入れて（下の例だとa），itertoolsの，組合せ＝combination, 順列＝ permutations, 直積＝product，で簡単にループを組める。 直積は重複順列に使える。
import itertools a = ['a', 'b', 'c'] for comb in itertools.combinations(a, r=2): print(comb) for comb in itertools.permutations(a, r=2): print(comb) for comb in itertools.product(a, repeat=2): print(comb)
週末プロジェクトネタ
休みの日にプログラミングしたいなぁ，新しい言語で何か作ってみたいなぁ，と思った時に，「さて，何を作ろうか？」で悩むことがある。 ということで，よくあるお題などを気づいたら書いていく。
注意点としては，ここではあまり言語に依存しない問題を上げているけど，WebをやってみたいからJavascriptとか，AndroidやってみたいからKotlinとか，そもそも得たい知識が先にある程度決まっている場合は，その応用例を実際にやってみるのが一番良い。それが具体的に無い場合でも，下の例と組み合わせると例が浮かぶかも知れない，Webのテトリス，Androidでテトリスとかね。
趣味系
何か趣味があるなら，その趣味の中であったらいいな，と思うものをメモしておく。
Web系
Blogサイト HTTPサーバからDBから一通り触れる。
HTTPサーバ(FTPクライアント) ライブラリを使わずにHTTPサーバを書いてみるのはHTTPの良い勉強になる。
Webスクレイピング 特定のサイトをスクレイピング出来るとやれることが増えて楽しい。でも，攻撃にならないように気をつけること。株価や特定の商品の値段とか。
チャットアプリ オンラインチャットアプリをもサーバの練習になる。
シンプルゲーム
迷路ゲーム 単純な迷路だけでなく，倉庫番にしてみたり，ボンバーマンみたいなものを作ってみたりも面白い。迷路のマップ生成も面白い。
レトロゲームクローン テトリス，Pong，Breakoutなど。最近だとOpenAIのAPI付けておくと強化学習の環境としても使える。
Sudoku 問題を作って，ソルバーを書いて。特にソルバは良い勉強になる。
ピクロス生成 画像からピクロス問題を作る。解ける問題かどうかを判定するのが難しいのかな？数独みたいにソルバーも書けるのかな？英語だとピクセルアートとかノノグラムとかイラストロジックとか言うのかな。
その他
シンプルチャットボット すごいチャットボットを作るのは難しいけど，会話内容を限定(ニッチ領域)すれば，それなりに動くようにみえるものは意外と作れたりする。あくまでも，そう見えるだけだけど。
統計アルゴリズム カルマンフィルタとかパーティクルフィルタなどを書いてみる。
離散アルゴリズム ここは問題の宝庫。いきなり難しい問題に取り組まず，時間に合わせて適時問題の難易度を考えてチョイスすると良いかも。
処理系・言語作成 チャレンジしては途中でやめてしまう系No1な気がするテーマ。僕も最後までやったことはない・・・。いつかやりきってみたいな。
Unixコマンドの再実装 これもよくある例かな。オプションとかも含めて完全なものを作ろうとはしないほうが良い。
プログラミングサイト Project EulerとかRozetta Codeの例とか。
OpenCVによる物体追跡
OpenCVには物体追跡アルゴリズムがcontrib-libとして提供されている。 Web上にたくさんサンプルが落ちているので参考にやってみる。
インストール
- 追跡アルゴリズムはcontrib提供なので別途インストールが必要。
- VideoCaptureを使うにはOpenCVのコンパイル時にVideo-I/Oがアクティブでなければならない，などの注意があるらしいが今回は問題が発生しなかったので詳細は不明。
$ pip3 install opencv-python $ pip3 install opencv-contlib-python
サンプルとメモ
- 's'キーを押すとROI(Region of interest)選択モードになる。これにはopenCVが提供するselectROI関数を利用する(便利な関数だなぁ)。fromCenterをFalseにしないとマウスの位置が矩形の中心になるので，左上角から矩形を描きたい場合にはFalseにしておく。
- Webの例だとROIの選択は一回のみで，再度's'キーで止めてROIを選択しても適切に動作しない場合が多い。そこを修正しようとしたが，ROIを上書きして，Trackerをinitするだけでは適切にリセットされないようで，'s'キーを押すたびにTrackerをコンストラクトしている。実用的ではないけど実験には十分。
- imshow()はwaitKeyで指定される秒数しか描画されない，描画を継続したい場合にはwaitKey(0)を指定する(引数無しでも良さそうだが，手元の環境ではうまく行かなかった)。また，waitKeyが無くimshow単体では適切に描画さえもされないので注意する。
- destroyAllWindowsにウィンドウを個別に渡している例も見られたが，destroyAllWindowsは引数が無くても，全てのウィンドウを破棄してくれるみたいだ。
- 使用するアルゴリズムの基本的な選択基準は，精度最優先=CSRT, 速度最優先=MOSSE。KCFはBoostingとMILの改良版なので，Boosting, MILは基本的には使わない。起動の予測が容易な場合にはMedianFlow。TLDは他のアルゴリズムと異なる特性を持つようだが，現在のOpenCVの実装にはバグがあるのか，適切に動かない(らしい)。試した印象ではCSRTと他で大きく精度が違う印象。
import argparse import imutils import cv2 def main(args): # Generate tracker TRACKERS = { "csrt": cv2.TrackerCSRT_create, "kcf": cv2.TrackerKCF_create, "boosting": cv2.TrackerBoosting_create, "mil": cv2.TrackerMIL_create, "tld": cv2.TrackerTLD_create, "median": cv2.TrackerMedianFlow_create, "mosse": cv2.TrackerMOSSE_create } tracker = TRACKERS[args.tracker]() # Load video stream cap = cv2.VideoCapture(args.video) assert cap.isOpened(), "Given video can't be opened." window = 'tracking test' target = None while True: is_read, frame = cap.read() if is_read: frame = imutils.resize(frame, width=500) # ndarray, (H,W,C) H, W, _ = frame.shape if target is not None: is_success, bbox = tracker.update(frame) if is_success: x, y, w, h = [int(d) for d in bbox] cv2.rectangle(frame, (x, y), (x+w, y+h), (0, 255, 0), 2) else: print('Tracking is failed.') cv2.imshow(window, frame) k = cv2.waitKey(10) & 0xFF if k == ord('q'): break elif k == ord('s'): # select target # After selection, ENTER or SPACE. Re-select by ESC del(tracker) tracker = TRACKERS[args.tracker]() target = cv2.selectROI(window, frame, fromCenter=False, showCrosshair=True) tracker.init(frame, target) else: break cap.release() cv2.destroyAllWindows() if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('-v', '--video', type=str, help='video file path') parser.add_argument('-t', '--tracker', type=str, default='csrt', help='tracker type') args = parser.parse_args() main(args)
CNNによるMNISTの埋め込み（t-SNEも）
- 前回に続いて，学習済みモデルの中間層の出力を取り出す例として，学習済みCNNを自己符号化(オートエンコーダ)として使う例をやってみる。
- コードはPyTorch公式exampleのmnistを少しいじって，モデルの中間層が取り出しやすくして，t-SNEによる埋め込みを行った。と言っても，t-SNEはsklearnを使っているだけ。t-SNEはアルゴリズム的に学習済みモデルを使って予測，という類の使い方ではない。つまり，他のsklearnのモデルみたいに学習データでfitして，評価したいデータをpredict，みたいな使い方は(基本的には)できない。
from __future__ import print_function import numpy as np from sklearn.manifold import TSNE import matplotlib.pyplot as plt import argparse import torch import torch.nn as nn import torch.nn.functional as F import torch.optim as optim from torchvision import datasets, transforms class Net(nn.Module): def __init__(self): super(Net, self).__init__() self.encoder = nn.Sequential( # Conv1 nn.Conv2d(1, 10, kernel_size=5), nn.MaxPool2d(kernel_size=2), nn.ReLU(), # Conv2 nn.Conv2d(10, 20, kernel_size=5), nn.Dropout2d(), nn.MaxPool2d(kernel_size=2), nn.ReLU(), ) self.fc1 = nn.Linear(320, 50) self.fc2 = nn.Linear(50, 10) def forward(self, x): x = self.encoder(x) x = x.view(-1, 320) x = F.relu(self.fc1(x)) x = F.dropout(x, training=self.training) x = self.fc2(x) return F.log_softmax(x, dim=1) def train(args, model, device, train_loader, optimizer, epoch): model.train() for batch_idx, (data, target) in enumerate(train_loader): data, target = data.to(device), target.to(device) optimizer.zero_grad() output = model(data) loss = F.nll_loss(output, target) loss.backward() optimizer.step() if batch_idx % args.log_interval == 0: print('Train Epoch: {} [{}/{} ({:.0f}%)]\tLoss: {:.6f}'.format( epoch, batch_idx * len(data), len(train_loader.dataset), 100. * batch_idx / len(train_loader), loss.item())) def test(args, model, device, test_loader): model.eval() test_loss = 0 correct = 0 with torch.no_grad(): for data, target in test_loader: data, target = data.to(device), target.to(device) output = model(data) # sum up batch loss test_loss += F.nll_loss(output, target, reduction='sum').item() # get the index of the max log-probability pred = output.max(1, keepdim=True)[1] correct += pred.eq(target.view_as(pred)).sum().item() test_loss /= len(test_loader.dataset) print('\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n'.format( test_loss, correct, len(test_loader.dataset), 100. * correct / len(test_loader.dataset))) def auto_encode(args, model, device, test_loader): model.eval() with torch.no_grad(): for data, target in test_loader: data, target = data.to(device), target.to(device) latent_vecs = model.encoder(data) latent_vecs = latent_vecs.view(-1, 320) latent_vecs = model.fc1(latent_vecs) return latent_vecs, target def main(train=False): # Training settings parser = argparse.ArgumentParser(description='PyTorch MNIST Example') parser.add_argument('--batch-size', type=int, default=64, metavar='N', help='input batch size for training (default: 64)') parser.add_argument('--test-batch-size', type=int, default=10000, metavar='N', help='input batch size for testing (default: 10000)') parser.add_argument('--epochs', type=int, default=10, metavar='N', help='number of epochs to train (default: 10)') parser.add_argument('--lr', type=float, default=0.01, metavar='LR', help='learning rate (default: 0.01)') parser.add_argument('--momentum', type=float, default=0.5, metavar='M', help='SGD momentum (default: 0.5)') parser.add_argument('--no-cuda', action='store_true', default=False, help='disables CUDA training') parser.add_argument('--seed', type=int, default=1, metavar='S', help='random seed (default: 1)') parser.add_argument('--log-interval', type=int, default=10, metavar='N', help='how many batches to wait before logging training\ status') args = parser.parse_args() use_cuda = not args.no_cuda and torch.cuda.is_available() torch.manual_seed(args.seed) device = torch.device("cuda" if use_cuda else "cpu") kwargs = {'num_workers': 1, 'pin_memory': True} if use_cuda else {} train_loader = torch.utils.data.DataLoader( datasets.MNIST('/home/hoge/PyTorch_MLdata/MNIST', train=True, download=True, transform=transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,)) ])), batch_size=args.batch_size, shuffle=True, **kwargs) test_loader = torch.utils.data.DataLoader( datasets.MNIST('/home/hoge/PyTorch_MLdata/MNIST', train=False, transform=transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))])), batch_size=args.test_batch_size, shuffle=True, **kwargs) model = Net().to(device) if train: optimizer = optim.SGD(model.parameters(), lr=args.lr, momentum=args.momentum) for epoch in range(1, args.epochs + 1): train(args, model, device, train_loader, optimizer, epoch) test(args, model, device, test_loader) torch.save(model.state_dict(), './model_param_autoencoder.pkl') # Auto Encode using trained model model.load_state_dict(torch.load('./model_param_autoencoder.pkl')) latent_vecs, target = auto_encode(args, model, device, test_loader) latent_vecs, target = latent_vecs.numpy(), target.numpy() print(latent_vecs.shape, target.shape) latent_vecs_reduced = TSNE(n_components=2,\ random_state=0).fit_transform(latent_vecs[:1000]) plt.scatter(latent_vecs_reduced[:, 0], latent_vecs_reduced[:, 1], c=target[:1000], cmap='jet') plt.colorbar() plt.show() if __name__ == '__main__': main(train=False)
PyTorchのモデルの書き方の整理
サマリ
いくつかの書き方があるが，混乱するのは下記の種類が色々Webに見られるからか。
- ① initにはパラメタありだけ，パラメタが無いものはforwardに。これが一番柔軟性があるように思う。モデルの中間層を細かく取り出したり，活性化関数を変えたりしたいならこれが良いと思う。
- ②initにReLuなども書いてしまう。print(model)でReLuなども含めて出力してくれる点が良い点。
- ③ Sequencialを使う。特定の塊ごとに意味を持たせられる。
これらは，それぞれ排他的に使う必要はなくて，適材適所でMixして使う。方針としては，学習後に分解することのない特定の塊があればどんどんSequencialに入れて，initにReLuなどのパラメタが無い層も含めて書いて良いと思う。もし，特定のレイヤを置換したりしたい場合は，それは独立したSequentialか，独立したレイヤにする。
基本事項
- nn.Moduleを継承して、構成要素をinitに定義して順方向をforwardに記載。callを描かなくても、モデルに入力を渡せば自動的にforwardを実行してくれる。
- Sequentialを使ったり、Activationを関数に渡してしまったり、色々なパタンがあるので整理しておく。
パタン１(最もシンプルなパタン)
- 学習パラメタを持つものはinitに、関数はforwardに。
- 公式チュートリアルはこの書き方。(Poolingもパラメタは無いからforward)
- print(model)で出力されるのはinitに描かれたもの。
class MLP(nn.Module): def __init__(self, input_size, hidden_size, num_classes): super(MultiLayerPerceptron, self).__init__() self.fc1 = nn.Linear(input_size, hidden_size) self.fc2 = nn.Linear(hidden_size, hidden_size) self.fc3 = nn.Linear(hidden_size, num_classes) def forward(self, x): h = F.relu(self.fc1(x)) h = F.relu(self.fc2(h)) out = self.fc3(h) return out
パタン2(ReLUなどもinitに集約)
- パラメタ付きか否かの分離ができないが、print文でReLUなども出力されるのでその点は良い。この場合はReLUは関数(F.relu)ではなくクラス(nn.ReLU)を使う。
class MLP(nn.Module): def __init__(self, input_size, hidden_size, num_classes): super(MultiLayerPerceptron, self).__init__() self.fc1 = nn.Linear(input_size, hidden_size) self.ac1 = nn.ReLU() self.fc2 = nn.Linear(hidden_size, hidden_size) self.ac2 = nn.ReLU() self.fc3 = nn.Linear(hidden_size, num_classes) def forward(self, x): h = self.ac1(self.fc1(x)) h = self.ac2(self.fc2(h)) out = self.fc3(h) return out
パタン3(Sequential)
- KerasのSequentialのような書き方が可能。単純なMLPやCNNならこれで良い。一方で多入力、多出力なネットワークの記述は出来ない。
model = nn.Sequential() model.add_module('fc1', nn.Linear(input_size, hidden_size)) model.add_module('relu1', nn.ReLU()) model.add_module('fc2', nn.Linear(hidden_size, hidden_size)) model.add_module('relu2', nn.ReLU()) model.add_module('fc3', nn.Linear(hidden_size, n_classes))
- 先にリストに層をまとめておくことも可能。 layers = [nn.Linear(input_size, hidden_size), nn.ReLU(), nn.Linear(hidden_size, hidden_size), nn.ReLU(), nn.Linear(hiddensize, n_classes)] model = nn.Sequential(*layers)
