タンヤオマシーンの実装について
インターネットにズブズブの皆さんこんばんは、元タンヤオです。
ところで、タンヤオマシーン、ご存知ですか？
前回のエントリでこのタンヤオマシーンを作った経緯について書きました（記事ちょっと修正しました）。
今回はその実装について書きます。ブログの普段の読者層（？）を考えるとこういう話題に特にニーズはなさそうなんですが、自分の作業をまとめる意味でもブログに残しておこうと思います。特に目新しいことや珍しいことをしているわけではないので、環境や実装の詳細には特に触れません。
作業内容まとめ
- 仮想環境上にpython実行環境を作った
- 和了判定（＋タンヤオ判定）を行うpython scriptを書いた
- 和了形の牌姿画像を生成できるようにした
- その画像をTwitterにpostするpython scriptを書いた
- 以上の処理を自宅PCから10分に1回実行するようにした
実行環境
- Windows7 64bit, Intel Core i7-3770@3.40GHz, RAM 8GB
- VirtualBox 4.3.28 for Windows hosts
- vagrant 1.7.2
- CentOS7 x64 (Minimal, Shrinked, Guest Additions 4.3.26)(Monthly updates)
参考：
Windows上でVirtualBox+Vagrant+CentOSによる仮想環境構築 - Qiita
CentOS7なのは単純に慣れているからというだけの理由です。vagrantめちゃめちゃ便利ですね。壊しても痛くないLinux仮想環境が十数分で用意できる時代やばい。
和了判定（＋タンヤオ判定）script
python 2.7.5 (CentOS7 default)で作成しました。pythonはこれまでいくつかのsource codeを読んだことがあるとか、10行くらいで書けるちょっとしたtext整形用のscript（bash+sed+awkでもささっと書ける程度）を作ったことくらいしかなくて、本格的？に書くのは初めてというレベルです。ちなみに大学の専攻も情報系ではないし職種もプログラマーではないです（プログラムを書いたり読んだりすることはありますが）
そもそも、pythonを勉強したくてこれを作ることにしたみたいなところがあり、前記事に書いた動機半分、pythonへの興味半分というところでとにかくやってみるbyゴセイジャーという流れです。
というわけでまったくのpython初心者でして、そもそもLLとか動的型付け言語と呼ばれる言語自体あまり使ったことないレベルであり、まだいろいろとわかってないところがありますのでその点ご了承ください。
一応公式チュートリアルを適当に眺めて、理解できたお作法についてはそれに則って書いたつもりですが、明らかに誤っている箇所や（趣味嗜好のレベルでなく）より良い実装がありましたら教えて頂けると助かります。
と一通りの前言い訳したところで、以下が今回作ったscriptです。
check_tenho_mp.py
#!/usr/bin/python # -*- coding: utf-8 -*- """ #------------------------------------------------------------------------------- # Name : check_tenho_mp.py # Description : Make tehai by random and check agari (Tenho) # Author : @Heipod # Created : 09/06/2015 # Copyright : (c) Heipod 2015 # Usage : # % python check_tenho_mp.py [-n NUM] # -n NUM : MAX try count for checking. (default=330536) # For using 4 processes, NUM must be a multiple of 4. # Output : # String of "agari tehai and yaku name" to stdout # e.g.) 3m 3m 4m 5m 6m 2p 3p 4p 5s 6s 6s 7s 7s 8s :TannYao #------------------------------------------------------------------------------- """ import sys import argparse import random import itertools import copy import multiprocessing as mp ##### Global definitions ##### proc_num = 4 # For multiprocessing hai = { 11:'1m', 12:'2m', 13:'3m', 14:'4m', 15:'5m', 16:'6m', 17:'7m', 18:'8m', 19:'9m', 21:'1p', 22:'2p', 23:'3p', 24:'4p', 25:'5p', 26:'6p', 27:'7p', 28:'8p', 29:'9p', 31:'1s', 32:'2s', 33:'3s', 34:'4s', 35:'5s', 36:'6s', 37:'7s', 38:'8s', 39:'9s', 41:'ton', 42:'nan', 43:'sha', 44:'pee', 51:'hak', 52:'hat', 53:'chu' } yaochu = frozenset([11, 19, 21, 29, 31, 39, 41, 42, 43, 44, 51, 52, 53]) ##### Functions ##### def output_hai(h, yaku): """ Convert hai to human-readable string and print to stdout with yaku name. Args : h : array of hai (List) yaku : string of yaku """ for i in h: print "%s" % hai.get(i), print ":%s" % yaku def make_tehai(): """ Make tehai from all of hai by random. Returns: tehai (List, 14 hais, sorted) """ haiall = [ key for key in hai.keys() for i in range(4) ] # hai * 4 return sorted(random.sample(haiall, 14)) ### TODO: If agari, this function should return the mentsu for checking yaku def check_agari(tehai): """ Check tehai is agari or not. Args : tehai (List, 14 hais) Returns : agari : True Not agari : None """ # Count kind of hai hai_accum = { x[0]: len(list(x[1])) for x in itertools.groupby(tehai) } # Search candidate of atama kind_hai = hai_accum.keys() cand_atama = [] for i in range(len(hai_accum)): if hai_accum[kind_hai[i]] >= 2: cand_atama.append(kind_hai[i]) cand_atama.sort() if len(cand_atama) == 0: return None # Check chitoitsu # TODO: Check ryanpeko case if len(cand_atama) == 7: return True # Check kokushi-musou if len(cand_atama) == 1: if frozenset(tehai) == yaochu: return True # Repeat while cand_atama remains while cand_atama: tmp_tehai = copy.deepcopy(tehai) # Remove tmp_atama from cand_atama and temp_tehai tmp_atama = cand_atama.pop(0) for atama_cnt in range(2): # remove 2 hais tmp_tehai.remove(tmp_atama) tmp_accum = \ { x[0]: len(list(x[1])) for x in itertools.groupby(tmp_tehai) } tmp_kind_hai = tmp_accum.keys() # TODO: Prepare another check order : [kotsu->shuntu], [shuntu->kotsu] # Search kotsu kotsu_list = [] kotsu_append = kotsu_list.append for i in xrange(len(tmp_accum)): if tmp_accum[tmp_kind_hai[i]] >= 3: kotsu_append(tmp_kind_hai[i]) # Remove kotsu from tmp for kotsu_cnt in range(3): # remove 3 hais tmp_tehai.remove(tmp_kind_hai[i]) kotsu_list.sort() # Search shuntu shuntsu = [] shuntsu_list = [] while tmp_tehai: target_hai = tmp_tehai[0] # If jihai(>40) remains, not shuntsu if target_hai > 40: break # If shuntsu exists if target_hai+1 in tmp_tehai and \ target_hai+2 in tmp_tehai: for k in range(3): shuntsu.append(target_hai+k) tmp_tehai.remove(target_hai+k) else: break shuntsu_list.append(shuntsu) shuntsu = [] # clear current shuntsu else: return True ### TODO: For now Only TannYao, add other yaku def check_yaku(tehai): """ Check tehai has yaku or not. Args : tehai (List, 14 hais) Returns : yaku ari : String of yaku yaku nashi : None """ # TannYao check if yaochu.isdisjoint(frozenset(tehai)): return "TannYao" else: return None def check_mp(try_count): """ Check agari by multiprocessing. Args : try_count : MAX try count for checking Returns : agari : tehai(List) Not agari : None """ for i in xrange(try_count): tehai = make_tehai() if check_agari(tehai) is not None: return tehai else: return None def main(try_count): """ Do main process for checking agari and yaku. Args : try_count : MAX try count for checking """ # try_count is divided into 4 ranges and add list (it must be iterable) each_range = [try_count/proc_num for i in xrange(proc_num)] # Check agari by multiprocessing pool = mp.Pool(proc_num) results = pool.map(check_mp, each_range) # get list of agari tehai # Check yaku tmp_agari_tehai = None for agari_tehai in results: if agari_tehai is not None: yaku = check_yaku(agari_tehai) if yaku is not None: output_hai(agari_tehai, yaku) break else: tmp_agari_tehai = agari_tehai else: if tmp_agari_tehai is not None: output_hai(tmp_agari_tehai, None) if __name__ == '__main__': # Parse arguments prs = argparse.ArgumentParser(description='Tenho') # -n option: MAX Try count prs.add_argument('-n', nargs='?', const=330536, default=330536) prs.add_argument('-v', '--version', \ action='version', version='%(prog)s : 0.2') args = prs.parse_args() try_count = int(args.n) main(try_count)
処理の概要
牌全種全数から14枚からなる手牌を生成し、それが和了形かどうかを判定、和了形の場合はタンヤオかどうかを判定する、それをtry_count分繰り返す、というscriptです。
手牌生成はrandom.sample()に丸投げですが、pythonの乱数生成器はMersenne twisterなので、この用途であれば問題無いと考えています。
判定アルゴリズムは人間が直感的に理解しやすそうなものを採用しています。
- 頭候補をすべて抽出（2枚以上ある牌種を候補に入れる）
- 最初の頭候補に着目し、その牌を2枚取り除く
- 残りの手牌から刻子を抽出
- 残りの手牌から順子を抽出
- すべての牌を抽出できて、残り手牌が0になれば和了り
- 和了りでない場合は次の頭候補に着目し、頭候補がなくなるまで2.から繰り返し
（他にも特殊役判定とかloop終了判定もありますが略）
役判定は現状タンヤオだけしか実装していませんが、もしきっちりやる場合は、下記を追加する必要があります。
- 和了判定の戻り値を True or None にするのではなく、面子をlistなどで返す方が良いです（再度面子抽出しなくても良いように）。
kotsu_listとかshuntsu_listはそれをやろうとした名残ですが、結局タンヤオだけでいいやってなったので使ってません。 - 刻子→順子の順で面子抽出を行っていますが、役判定を行う場合は逆順（順子→刻子の順）での抽出も実施した上で、それぞれの面子の捉え方によって異なる役を判別すべきです（純チャンor三暗刻、みたいに捉え方によって翻数が変わる和了形を考慮）
classに切り出すとかクロージャ使うとかもっとより良い設計にすべき余地はあると思うのですが、とにかくまず書き始めた状態からほぼ最後まで書けてしまったのでこれでOKとしました。 牌種を定義するdict型の部分で、keyとvalueを逆のような使い方にしてるのはちょっとトリッキーな感じもしますが、これは単純に処理するには数値のほうが便利だが最終的な出力はhuman-readable（？）にしたい、という前提で、処理に使ったvalueからkeyを引くよりも最初からこの形（keyを算術演算に使う）にするほうが簡単だと思ったためです。
基本的にsource codeはコメント含め（文法ガバガバでもいいから）英語で書くようにしているのですが、麻雀用語に関しては日本語のローマ字表記にしています。だってhonours（字牌）とかset（面子）とかchow（順子）とかpung（刻子）とか直感的にわかりにくくないですか。setに至っては予約語だし。。。
並列処理について
当初の実装はmultiprocessでなく、single processで330530回実行していました。この場合、私の環境だとだいたい最大43秒程度でした（和了りが発生しない場合。和了りが見つかった場合はそこで打ち切るので早く終わる）。
上記のように4並列のmultiprocessにした場合は、11秒程度まで短縮しました。ただ前者と異なり、和了りが発生しないprocessがあるとそれを待ち合わせるので、これ以上早く終わる確率はかなり低く、和了りがあろうがなかろうがだいたいこの処理時間になります。。ちなみに実行環境上では仮想コア合わせて8つコアがあるので8並列が最速になりますが、1秒程度しか速くならない割に、全コアでCPU使用率が100%になるため、定期的に実行するには他のprocessへの影響が若干不安なので4並列にしています。
備考
なお、pythonでの他の実装例もWebにあります（参考： Python - 麻雀のアガリ形の判別 - Qiita ）。これで十分やんけという感じですが、前述のとおりpythonの勉強が目的でもあるので、今回は最初から自分で書きました。 また、実用性を考慮してとにかく高速に判定したい場合は和了形をすべてindex化しておくのが妥当だと思います（参考： 麻雀 和了判定（役の判定）アルゴリズム ）
所感
きちんと記録していたわけではないですが、作業時間は設計・実装・refactoringで5日程度（環境構築やら含む）でした。 チュートリアルやらGoogleやらを駆使しながらだったのですが、思ったよりスムーズに書けた感じです。きちんと書こうと思ったら全然知識が足りていないという自覚はありますが、とにかく動く（そしてある程度読みやすいコードにする）ものを作るという用途ならば、学習コストは低めなのかなと感じました。ちなみにひと通り作った後、Numbaとかそういうものを使って高速化しようと試みたりもしましたが、さくっとできなかったのですぐ諦めました。
画像の生成
和了形があった場合、和了判定scriptは下記のような出力をstdoutに書き出します。
3m 3m 4m 5m 6m 2p 3p 4p 5s 6s 6s 7s 7s 8s :TannYao
これをbash+sedで整形（画像file名に変換など）してから、ImageMagickに食わせてconvertすることで牌姿画像を生成しています。
ImageMagick: Convert, Edit, Or Compose Bitmap Images
牌の画像はこちらから無料素材をお借りしました。ありがとうございます。
Twitterへpost
TwitterAPIを使用するscriptは、下記を参考にして python + Requests-OAuthlibを使って書きました。第1引数にtext、第2引数に画像fileを渡してTweetするようにしています。和了判定とは関係なく他の用途にも使えるので別のscriptとして用意しました。script略。
自動運用
以上で作成した一連の処理を実行するbash scriptを作成しました。script略。
- 和了判定scriptを実行
- その結果を受けて、和了形があれば牌姿画像生成
- TwitterAPI使用scriptを実行し、text+牌姿画像fileをpost
- output_img directoryに牌姿画像fileを移動し保存
このbash scriptをcronie-anacronで10分おきに実行させています。 牌姿画像生成やらすべてをpythonでやることもできるとは思いますが、今回は面倒だったのでありもので済ましてしまいました。bashを使用しているのは、生成した牌姿画像をdirectoryに入れて整理するなどのfile操作をするのが楽という理由だけです。
手牌を表す文字列をTwitterからreplyしたら、それを受けて牌姿画像に変換して返す、等の便利機能も当初は実装しようと思っていましたが、他に作りたいものができたのでひとまずこれで終わりです。 しばらく適当に自動運用するつもりですが、気分次第でいきなり止めたりするかもしれませんのでその点ご承知おきください。もしダブル役満とか美しい天和が出たらその時点で止めたい。
実行結果の例
個人的に一番「おっ」となった和了形はこれです。タンヤオじゃないけど。クソ惜しい。
ただの天和 pic.twitter.com/xAaJIR3Rsd
— ﾀﾝﾔｵ (@tannyao_saikou) June 13, 2015まとめ
python最高～(✌ ◜◡◝ )✌
今後の展望
とりあえずタンヤオはこれで終わりにして、今後は（仲間内でブームが来てる）とあるゲームをプレイしやすくできるかもしれないWebアプリを作ってみようと画策中です。 とはいえ、サーバサイドやフロントエンドなどこれまであまりやってこなかった分野のプログラミングスキルが必要で、現状ではだいぶ大変そうなのでどれだけ出来るかわかりませんが、やってみようと思います。
