Python: Do Not Use Pool.map Method in multiprocessing Module

Python: multiprocessing モジュールの Pool.map を使ったときの罠

 

Pool.map を使った並行処理をバッチ処理などで実行した際、キーボードによる中断(KeyboardInterrupt)をすると
プロセスがハングアップすることがある。

コード

#!/usr/bin/env python

from multiprocessing import Pool, current_process
import time
from datetime import datetime

def f(x):
    print('[%s] start %d: %s' % (datetime.now(), x, current_process().name))
    time.sleep(5)
    print('[%s] end   %d: %s' % (datetime.now(), x, current_process().name))
    return x

pool = Pool(2)
ret = pool.map(f, range(4))
print('[%s] result: %s' % (datetime.now(), ret))

通常の実行例

[2015-12-31 22:10:02.319094] start 0: PoolWorker-1
[2015-12-31 22:10:02.319267] start 1: PoolWorker-2
[2015-12-31 22:10:07.320538] end   1: PoolWorker-2
[2015-12-31 22:10:07.320538] end   0: PoolWorker-1
[2015-12-31 22:10:07.321345] start 2: PoolWorker-2
[2015-12-31 22:10:07.321434] start 3: PoolWorker-1
[2015-12-31 22:10:12.321765] end   2: PoolWorker-2
[2015-12-31 22:10:12.321765] end   3: PoolWorker-1
[2015-12-31 22:10:12.322721] result: [0, 1, 2, 3]

Ctrl-C を押した場合

[2015-12-31 22:05:42.108238] start 0: PoolWorker-1
[2015-12-31 22:05:42.109395] start 1: PoolWorker-2
^CProcess PoolWorker-1:
Process PoolWorker-2:
Traceback (most recent call last):
Traceback (most recent call last):
(snip)
    return map(*args)
    time.sleep(5)
  File "./x.py", line 9, in f
KeyboardInterrupt
    time.sleep(5)
KeyboardInterrupt
[2015-12-31 22:05:42.631444] start 2: PoolWorker-3
[2015-12-31 22:05:42.632162] start 3: PoolWorker-4
[2015-12-31 22:05:47.632713] end   3: PoolWorker-4
[2015-12-31 22:05:47.632664] end   2: PoolWorker-3

プログラムが停止せず、kill コマンドでプロセスを終了する必要に迫られる。
何度 Ctrl-C を押しても駄目である。

回避策1

try 節で KeyboardInterrupt をトラップすれば、プログラムのハングアップは防げる。
しかし、中断した後もプログラムが続行してしまう。

#!/usr/bin/env python

from multiprocessing import Pool, current_process
import time
from datetime import datetime

def f(x):
    try:
        print('[%s] start %d: %s' % (datetime.now(), x, current_process().name))
        time.sleep(5)
        print('[%s] end   %d: %s' % (datetime.now(), x, current_process().name))
    except KeyboardInterrupt:
        pass
    return x

pool = Pool(2)
ret = pool.map(f, range(4))
print('[%s] result: %s' % (datetime.now(), ret))

実行例

[2015-12-31 22:21:41.647775] start 0: PoolWorker-1
[2015-12-31 22:21:41.647928] start 1: PoolWorker-2
^C[2015-12-31 22:21:42.716623] start 2: PoolWorker-2
[2015-12-31 22:21:42.716755] start 3: PoolWorker-1
[2015-12-31 22:21:47.717815] end   2: PoolWorker-2
[2015-12-31 22:21:47.717832] end   3: PoolWorker-1
Traceback (most recent call last):
(snip)
    waiter.acquire()
KeyboardInterrupt

回避策2

map の代わりに map_async を使えば、問題を回避できる。この場合は即座にプログラムが終了する。
ただし、結果を利用する際には get メソッドとともにタイムアウトの時間(秒数)を指定する必要がある。

#!/usr/bin/env python

from multiprocessing import Pool, current_process
import time
from datetime import datetime

def f(x):
    print('[%s] start %d: %s' % (datetime.now(), x, current_process().name))
    time.sleep(5)
    print('[%s] end   %d: %s' % (datetime.now(), x, current_process().name))
    return x

pool = Pool(2)
p = pool.map_async(f, range(4))
ret = p.get(86400)
print('[%s] result: %s' % (datetime.now(), ret))

実行例

[2015-12-31 22:24:32.482984] start 0: PoolWorker-1
[2015-12-31 22:24:32.483816] start 1: PoolWorker-2
^CTraceback (most recent call last):
(snip)
    time.sleep(5)
KeyboardInterrupt

 

References

• Issue 8296: multiprocessing.Pool hangs when issuing KeyboardInterrupt - Python tracker

color-ssh: Runs Remote Commands, Colorfully!

color-ssh: カラフルなリモートコマンドの実行ツール

 

color-ssh というコマンドライン・ツールを作った。

• mogproject/color-ssh

 

インストール

# pip install color-ssh

• 環境によっては「sudo」コマンドが必要
• バージョンアップする場合は、「pip install --upgrade color-ssh」

 

「color-cat」と「color-ssh」という 2個のコマンドがインストールされる。
以下に、基本的な使い方を説明する。

$ color-cat --version
color-cat 0.1.0
$ color-ssh --version
color-ssh 0.1.0

 

color-cat

UNIX/Linux の cat コマンドにターミナル上で色を付けるだけのツール。

• 引数なしで実行すれば、標準入力から読み取った文字列が、ターミナルに赤色で表示される。
• 「-l」オプションでラベルを指定。各行の左側に反転色で表示される。
  • ラベルを変えれば色も変わる。
    ラベルの文字列によって、色が一意に決まるというのが最大の特徴。
• 「-c」オプションで色を直接指定することも可能。
• 「-s」オプションでセパレータ文字列も変えられる。
• もちろん、通常の cat コマンド同様、引数にファイルパスを指定すれば、その内容を全て表示する。
• 「color-cat --help」でヘルプを表示。

 

color-ssh

ssh コマンドの出力を色付きで表示するためのツール。
内部では color-cat コマンドが実行されている。

• ssh コマンドを使ってリモートサーバ上でコマンドを実行し、その出力を色付きで表示する。
• サーバの名前が color-cat のラベルとして利用される。そのため同じサーバは同じ色で出力される。
• 標準エラー出力はセパレータが「+」に変わる。
• 複数のサーバ上で、同じコマンドを同時に実行することが可能。(parallel-ssh と同じような機能)
  • ホスト名のリストは、「-h」オプションでファイルから読み込むか、「-H」オプションで文字列として指定する。
• 「-p」オプションで多重度を制限可能。
• 「color-ssh --help」でヘルプを表示。

 

コマンドライン引数の「分配」

「--distribute」オプションを指定すれば、引数を各サーバに分割・分配して実行することができる。
(これをやりたかった)

上記の例では、「a」「b」「c」「d」「e」という 5個の引数を、各サーバにほぼ均等になるように割り当てて、そのコマンドを並行実行する。

server-1 では「echo xxx a b c」
server-2 では「echo xxx d e」

というコマンドが同時に実行されている。

この機能は、多数の入力ファイルと複数のサーバがある場合、手軽に仕事を分割するのに役立つと思う。

How to Handle Binary-data stdin/stdout and Command-line Arguments in Python3

Python: 標準入出力およびコマンドライン引数でバイナリデータを取り扱う方法

 

Python2 の場合

str = bytes なので、普通に書けばバイナリデータにも対応できる。

import sys

sys.stdout.write('### sys.argv\n')
for arg in sys.argv:
    sys.stdout.write(arg)
    sys.stdout.write('\n')

sys.stderr.write('### sys.stdin\n')
for line in iter(sys.stdin.readline, ''):
    sys.stderr.write(line.rstrip())
    sys.stderr.write('\n')

• 実行例 (?? の箇所は表示非対応)

$ echo $'abc\nあいう\n\xff\xfe' | python2 ./bin_stdin.py abc あいう $'\xff\xfe'
### sys.argv
./bin_stdin.py
abc
あいう
??
### sys.stdin
abc
あいう
??

 

Python3 の場合

上記のコードはエンコードエラーになる。
これは、Python3 の sys.stdout.write が str = unicode を引数に取るため。

$ echo $'abc\nあいう\n\xff\xfe' | python3 ./bin_stdin.py abc あいう $'\xff\xfe'
### sys.argv
./bin_stdin.py
abc
あいう
Traceback (most recent call last):
  File "./bin_stdin.py", line 5, in
    sys.stdout.write(arg)
UnicodeEncodeError: 'utf-8' codec can't encode character '\udcff' in position 0: surrogates not allowed

 

ポイント

• 1. sys.stdXXX.buffer を利用する
• 2. sys.args の要素を os.fsencode でエンコードする
  • ただしOptionParserなどのパース処理はエンコード前に適用しないと正しく処理されない模様

以下のようなコードを書けば、Python2/3 両方に対応できる。 (flush は必ずしも必須ではない)

import sys
import os

PY3 = sys.version_info >= (3, )

stdin = sys.stdin.buffer if PY3 else sys.stdin
stdout = sys.stdout.buffer if PY3 else sys.stdout
stderr = sys.stderr.buffer if PY3 else sys.stderr

stdout.write(b'### sys.argv\n')
for arg in sys.argv:
    stdout.write(os.fsencode(arg) if PY3 else arg)
    stdout.write(b'\n')
    stdout.flush()

stderr.write(b'### sys.stdin\n')
for line in iter(stdin.readline, b''):
    stderr.write(line.rstrip())
    stderr.write(b'\n')
    stderr.flush()

• 実行例

$ echo $'abc\nあいう\n\xff\xfe' | python3 ./bin_stdin.py abc あいう $'\xff\xfe'
### sys.argv
./bin_stdin.py
abc
あいう
??
### sys.stdin
abc
あいう
??

Python: How to Execute Parallel Processing using multiprocessing.Pool

Python: multiprocessing.Pool を使った並列処理の実行

 

Python で並列処理を行う場合、自分でスレッドを書くこともできるが、グローバルインタプリタロック (GIL) の制約を受けて意図しない結果に陥ることが少なくない。

multithreading モジュールの Pool を利用するのが最も簡単かつ安全である。

 

基本形

引数を1個取る関数を定義し、Pool#map を実行する。

import time
from datetime import datetime
from multiprocessing import Pool, current_process

def f(x):
    proc_name = current_process().name
    print('(%s) [%s] Started: x=%d' % (proc_name, datetime.now(), x))
    time.sleep(1)
    print('(%s) [%s] Ended  : x=%d' % (proc_name, datetime.now(), x))
    return x * x

pool = Pool(4)
print(pool.map(f, range(8)))

尚、Python 3.4 以降なら with 句を使った表現もできる。

実行結果は以下のようになるはずだ。

(PoolWorker-1) [2015-12-08 01:49:03.666114] Started: x=0
(PoolWorker-2) [2015-12-08 01:49:03.666246] Started: x=1
(PoolWorker-3) [2015-12-08 01:49:03.666400] Started: x=2
(PoolWorker-4) [2015-12-08 01:49:03.666620] Started: x=3
(PoolWorker-1) [2015-12-08 01:49:04.667421] Ended  : x=0
(PoolWorker-3) [2015-12-08 01:49:04.667421] Ended  : x=2
(PoolWorker-2) [2015-12-08 01:49:04.667423] Ended  : x=1
(PoolWorker-4) [2015-12-08 01:49:04.667511] Ended  : x=3
(PoolWorker-3) [2015-12-08 01:49:04.668070] Started: x=4
(PoolWorker-2) [2015-12-08 01:49:04.668151] Started: x=5
(PoolWorker-1) [2015-12-08 01:49:04.668282] Started: x=6
(PoolWorker-4) [2015-12-08 01:49:04.668382] Started: x=7
(PoolWorker-3) [2015-12-08 01:49:05.669206] Ended  : x=4
(PoolWorker-2) [2015-12-08 01:49:05.669206] Ended  : x=5
(PoolWorker-4) [2015-12-08 01:49:05.669218] Ended  : x=7
(PoolWorker-1) [2015-12-08 01:49:05.669230] Ended  : x=6
[0, 1, 4, 9, 16, 25, 36, 49]

実行中はプロセスがフォークされているので、特にメモリ使用量は気にかける必要がある。

 

複数のパラメータを取る

タプルを渡して、ワーカー側で分解するのが常套手段。

import time
from datetime import datetime
from multiprocessing import Pool, current_process

def f(args):
    x, y = args
    proc_name = current_process().name
    print('(%s) [%s] Started: x=%d, y=%d' % (proc_name, datetime.now(), x, y))
    time.sleep(1)
    print('(%s) [%s] Ended  : x=%d, y=%d' % (proc_name, datetime.now(), x, y))
    return x * y

pool = Pool(4)
print(pool.map(f, zip(range(8), range(10, 18))))

実行例

(PoolWorker-1) [2015-12-08 01:55:01.700005] Started: x=0, y=10
(PoolWorker-2) [2015-12-08 01:55:01.700078] Started: x=1, y=11
(PoolWorker-3) [2015-12-08 01:55:01.700196] Started: x=2, y=12
(PoolWorker-4) [2015-12-08 01:55:01.700463] Started: x=3, y=13
(PoolWorker-3) [2015-12-08 01:55:02.701283] Ended  : x=2, y=12
(PoolWorker-2) [2015-12-08 01:55:02.701275] Ended  : x=1, y=11
(PoolWorker-1) [2015-12-08 01:55:02.701261] Ended  : x=0, y=10
(PoolWorker-4) [2015-12-08 01:55:02.701293] Ended  : x=3, y=13
(PoolWorker-3) [2015-12-08 01:55:02.701914] Started: x=4, y=14
(PoolWorker-1) [2015-12-08 01:55:02.702046] Started: x=5, y=15
(PoolWorker-2) [2015-12-08 01:55:02.702165] Started: x=6, y=16
(PoolWorker-4) [2015-12-08 01:55:02.702361] Started: x=7, y=17
(PoolWorker-1) [2015-12-08 01:55:03.703123] Ended  : x=5, y=15
(PoolWorker-3) [2015-12-08 01:55:03.703106] Ended  : x=4, y=14
(PoolWorker-4) [2015-12-08 01:55:03.703131] Ended  : x=7, y=17
(PoolWorker-2) [2015-12-08 01:55:03.703140] Ended  : x=6, y=16
[0, 11, 24, 39, 56, 75, 96, 119]

 

ワーカーのスタックトレースを出力する

何らかの例外が発生する場合を考える。

import time
from datetime import datetime
from multiprocessing import Pool, current_process

def f(args):
    x, y = args
    proc_name = current_process().name
    print('(%s) [%s] Started: x=%d, y=%d' % (proc_name, datetime.now(), x, y))
    time.sleep(1)
    print('(%s) [%s] Ended  : x=%d, y=%d' % (proc_name, datetime.now(), x, y))
    return y / x

pool = Pool(4)
print(pool.map(f, zip(range(8), range(10, 18))))

そのままだと、スタックトレースの内容が分かりづらい。

(PoolWorker-1) [2015-12-08 01:59:54.088732] Started: x=0, y=10
(PoolWorker-2) [2015-12-08 01:59:54.088835] Started: x=1, y=11
(PoolWorker-3) [2015-12-08 01:59:54.088960] Started: x=2, y=12
(PoolWorker-4) [2015-12-08 01:59:54.089304] Started: x=3, y=13
(PoolWorker-3) [2015-12-08 01:59:55.090067] Ended  : x=2, y=12
(PoolWorker-2) [2015-12-08 01:59:55.090095] Ended  : x=1, y=11
(PoolWorker-1) [2015-12-08 01:59:55.090049] Ended  : x=0, y=10
(PoolWorker-4) [2015-12-08 01:59:55.090160] Ended  : x=3, y=13
(PoolWorker-2) [2015-12-08 01:59:55.090750] Started: x=4, y=14
(PoolWorker-3) [2015-12-08 01:59:55.090926] Started: x=5, y=15
(PoolWorker-4) [2015-12-08 01:59:55.091020] Started: x=6, y=16
(PoolWorker-1) [2015-12-08 01:59:55.091336] Started: x=7, y=17
Traceback (most recent call last):
  File "xxx.py", line 14, in
    print(pool.map(f, zip(range(8), range(10, 18))))
  File "/Users/xxxxxx/.pyenv/versions/2.7.9/lib/python2.7/multiprocessing/pool.py", line 251, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/Users/xxxxxx/.pyenv/versions/2.7.9/lib/python2.7/multiprocessing/pool.py", line 558, in get
    raise self._value
ZeroDivisionError: integer division or modulo by zero

デコレータでスタックトレースを表示してみる。副産物として、関数 f が複数のパラメータを取れるようになった。

import time
from datetime import datetime
from multiprocessing import Pool, current_process
import traceback

def with_stacktrace(func):
    import functools

    @functools.wraps(func)
    def wrapper(args):
        try:
            return func(*args)
        except:
            proc_name = current_process().name
            for line in traceback.format_exc().splitlines():
                print('[TRACE:%s] %s' % (proc_name, line))
            raise
    return wrapper

@with_stacktrace
def f(x, y):
    proc_name = current_process().name
    print('(%s) [%s] Started: x=%d, y=%d' % (proc_name, datetime.now(), x, y))
    time.sleep(1)
    print('(%s) [%s] Ended  : x=%d, y=%d' % (proc_name, datetime.now(), x, y))
    return y / x

pool = Pool(4)
print(pool.map(f, zip(range(8), range(10, 18))))

実行例

(PoolWorker-1) [2015-12-08 02:31:36.959575] Started: x=0, y=10
(PoolWorker-2) [2015-12-08 02:31:36.959657] Started: x=1, y=11
(PoolWorker-3) [2015-12-08 02:31:36.959721] Started: x=2, y=12
(PoolWorker-4) [2015-12-08 02:31:36.960174] Started: x=3, y=13
(PoolWorker-4) [2015-12-08 02:31:37.960911] Ended  : x=3, y=13
(PoolWorker-2) [2015-12-08 02:31:37.960904] Ended  : x=1, y=11
(PoolWorker-3) [2015-12-08 02:31:37.960920] Ended  : x=2, y=12
(PoolWorker-1) [2015-12-08 02:31:37.960888] Ended  : x=0, y=10
(PoolWorker-2) [2015-12-08 02:31:37.961423] Started: x=4, y=14
(PoolWorker-3) [2015-12-08 02:31:37.961634] Started: x=5, y=15
(PoolWorker-4) [2015-12-08 02:31:37.961724] Started: x=6, y=16
[TRACE:PoolWorker-1] Traceback (most recent call last):
[TRACE:PoolWorker-1]   File "xxx.py", line 9, in f
[TRACE:PoolWorker-1]     return func(*args)
[TRACE:PoolWorker-1]   File "xxx.py", line 23, in f
[TRACE:PoolWorker-1]     return y / x
[TRACE:PoolWorker-1] ZeroDivisionError: integer division or modulo by zero
(PoolWorker-1) [2015-12-08 02:31:37.962384] Started: x=7, y=17
Traceback (most recent call last):
  File "xxx.py", line 26, in
    print(pool.map(f, zip(range(8), range(10, 18))))
  File "/Users/xxxxxx/.pyenv/versions/2.7.9/lib/python2.7/multiprocessing/pool.py", line 251, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/Users/xxxxxx/.pyenv/versions/2.7.9/lib/python2.7/multiprocessing/pool.py", line 558, in get
    raise self._value
ZeroDivisionError: integer division or modulo by zero

これでデバッグが捗る。

 

 

References

• 16.6. multiprocessing — プロセスベースの “並列処理” インタフェース — Python 2.6ja2 documentation
• moco(beta)'s backup: Python 標準ライブラリ探訪 (23) ~ multiprocessing.Pool 編 ~
• Ned Batchelder: Re-throwing exceptions in Python

C++/Python: Writing Custom Converters in Boost.Python

C++/Python: Boost.Python でカスタムコンバーターを作る

 

C++ のクラスを Python で扱うとき、std::pair などのクラスはデフォルトでは型変換が行われない。自分でコンバーターを書く必要がある。

 

以下はコードの例。C++11のラムダ式を利用したら綺麗にまとまった。

#include <boost/python.hpp>
#include "your_code.hpp"


namespace yourmodulename {
  namespace py = boost::python;

  /*
   * Converter for result set (std::vector of std::pair => pylist of pytuple)
   */
  namespace detail {
    template <typename T, typename U>
    struct pair_vector_to_tuple_list {
      static PyObject* convert(std::vector<std::pair<T, U> > const& ps) {
        py::list ret;
        for (auto it: ps) ret.append(py::make_tuple(it.first, it.second));
        return py::incref(ret.ptr());
      }

      static PyTypeObject const* get_pytype() { return &PyList_Type; }
    };
  }

  template <typename T, typename U>
  void expose_pair_vector_to_tuple_list() {
    py::to_python_converter<std::vector<std::pair<T, U> >, detail::pair_vector_to_tuple_list<T, U>, true>();
  }

  /*
   * Converter for list parameter (pylist => std::vector)
   */
  template <typename T>
  void expose_pylist_to_vector() {
    typedef std::vector<T> VT;

    auto convertible = [](PyObject* obj_ptr) -> void* { return PySequence_Check(obj_ptr) ? obj_ptr : NULL; };

    auto construct = [](PyObject* obj_ptr, py::converter::rvalue_from_python_stage1_data* data) {
      VT* storage = new (reinterpret_cast<py::converter::rvalue_from_python_storage<VT>*>(data)->storage.bytes) VT();
      for (py::ssize_t i = 0, l = PySequence_Size(obj_ptr); i < l; ++i) {
        storage->push_back(py::extract<typename boost::range_value<VT>::type>(PySequence_GetItem(obj_ptr, i)));
      }
      data->convertible = storage;
    };

    py::converter::registry::push_back(convertible, construct, py::type_id<VT>());
  }

}


BOOST_PYTHON_MODULE(yourmodulename) {
  using namespace boost::python;
  using namespace yourmodulename;

  // expose converters
  expose_pair_vector_to_tuple_list<int, double>();
  expose_pylist_to_vector<int>();

  // expose classes
  ...
}

なかなかに魔力を吸い取られそうなコードだ。

 

 

Related Posts

• mog project: C++/Python: Diving Into Boost.Python
• mog project: C++/Python: How to run Boost.Python module using setup.py with Python3.4 on Mac

 

References

• How to write boost.python converters | Misspent
• Boost.Python の機能をざっと紹介してみる - muddy brown thang

Top N Sorter in Python

Python: トップNソートの実装

 

ソート可能な iterable に対して、降順の上位 N件のソート結果を取得したい。

heapq を使えば効率が良さそうだ。

 

コード

import heapq


class TopNSorter(object):
    def __init__(self, limit):
        self._heap = []
        self.limit = limit

    def insert(self, item):
        if len(self._heap) == self.limit:
            heapq.heappushpop(self._heap, item)
        else:
            heapq.heappush(self._heap, item)

    def result(self):
        h = self._heap[:]
        buf = []
        while h:
            buf.insert(0, heapq.heappop(h))
        return buf

 

実行例

>>> s = TopNSorter(5)
>>> s.result()
[]
>>> s.insert(10)
>>> s.result()
[10]
>>> s.insert(20)
>>> s.result()
[20, 10]
>>> s.insert(10)
>>> s.result()
[20, 10, 10]
>>> s.insert(5)
>>> s.result()
[20, 10, 10, 5]
>>> s.insert(30)
>>> s.result()
[30, 20, 10, 10, 5]
>>> s.insert(15)
>>> s.result()
[30, 20, 15, 10, 10]
>>> s.insert(10)
>>> s.result()
[30, 20, 15, 10, 10]
>>> s.insert(25)
>>> s.result()
[30, 25, 20, 15, 10]
>>> s.insert(1)
>>> s.result()
[30, 25, 20, 15, 10]

 

Related Posts

• mog project: Top N Sorter in Scala

Python: How to Avoid 'multiprocessing' TypeError on Exit

Python: multiprocessing TypeError の回避方法

 

Python 2.6 でテスト (python setup.py test) をしたとき、最後に以下のようなエラーが出ることがある。

Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/opt/python/2.6.9/lib/python2.6/atexit.py", line 24, in _run_exitfuncs
    func(*targs, **kargs)
  File "/opt/python/2.6.9/lib/python2.6/multiprocessing/util.py", line 258, in _exit_function
    info('process shutting down')
TypeError: 'NoneType' object is not callable

回避策は、以下に従い setup.py に記述をすること。

• Issue 15881: multiprocessing 'NoneType' object is not callable - Python tracker

try:
    # Work around a traceback on Python < 2.7.4 and < 3.3.1 
    # http://bugs.python.org/issue15881#msg170215
    import multiprocessing  # noqa: unused
except ImportError:
    pass

コメントに noqa を付けると、pep8 などのチェックを回避できる。

Universal Use of 'subprocess' Module in Python

Python: バージョン・OS 透過的な subprocess モジュールの利用

 

Python のバージョン (2.6 / 2.7 / 3.2 / 3.3 / 3.4 / 3.5) や、OS環境 (Linux / Mac / Windows /CygWin) に依存せずに外部コマンドを実行するコードを書きたい。

 

ワークアラウンド

なかなか一筋縄ではいかない。

Python3.2 + POSIX環境 では、コマンドラインを bytes で渡してはいけない

以下のようなエラーが出る。

>>> import subprocess
>>> subprocess.call(b'echo')
TypeError: expect bytes or str, not int

subprocess モジュールの中で、args が文字列かどうかを判定するときに bytes である場合の考慮が不足しており、bytes の個々の要素(int)に対して分割が行われてしまうため。

• Issue 8513: subprocess: support bytes program name (POSIX) - Python tracker

対策は、コマンドラインをリストとして渡せばよい。以下のように対策する。(args が bytes である場合)

import sys

if sys.version_info[:2] == (3, 2):
    args = [args]

 

Python3 + Windows環境では、コマンドラインと環境変数のキー・値を bytes で渡してはいけない

以下のようなエラーが出る。

>>> import subprocess
>>> subprocess.call(b'cmd /C echo x')
    needquote = (" " in arg) or ("\t" in arg) or not arg
TypeError: argument of type 'int' is not iterable

bytes に対する in オペレーションに失敗しているが、これといった対策が見当たらない。bytes で渡すのをやめて、str で渡すことにする。(その場合、コマンドライン文字列のエンコードは指定できない)

import sys
import six
if six.PY2 or sys.platform != 'win32':
    # この場合だけ bytes にエンコード

env の指定についても同様。キーだけではなく、値も bytes が許容されていないので注意。

 

POSIX環境 + shell=True のとき、args をリストとして渡してはいけない

リストの2番目以降の引数が評価されない。

>>> import subprocess
>>> subprocess.call(['echo', '3'], shell=True)

0

ワークアラウンドとしては、subprocess.list2cmdline を使って適切な文字列を生成する。

import sys
import subprocess
if shell and sys.platform != 'win32':
    args = [subprocess.list2cmdline(args)]

 

外部コマンドについて

 

sleep

• How to sleep for 5 seconds in Windows's Command Prompt? (or DOS) - Stack Overflow

Windows 7 以降は「timeout」というコマンドが標準装備されているが、Python から実行すると標準入力が奪われてしまい不都合だった。


結局、Python の time モジュールを使うのが一番簡便だという結論に。

import subprocess
subprocess.call('python -c "import time;time.sleep(2)"')

 

プロセステーブルの取得

• psutil 3.2.2 : Python Package Index を使うか
• 自分で実装するか
  • Windows: TASKLIST /FO CSV /NH /FI PID eq プロセスID
  • POSIX: os.kill(プロセスID, 0)

Python: Terminal I/O Assessment in CygWin

Python: CygWin 環境におけるターミナル I/O 周りの調査

環境

• OS: Windows Server 2012 R2 日本語版
• CygWin: 2.873 (64bit)
• Python: 2.7

 

目的

Python 上で以下の操作が可能か調べる。

• ターミナルのエンコーディングの自動判別
• ターミナル画面のクリア
• ターミナル上のキー入力の検知

 

バリエーション

調査したバリエーションは以下 5 種類。

環境	フロントエンド	シェル	Pythonビルド	エンコーディング
1	cmd.exe	-	Windows	cp932
2	cmd.exe	CygWin bash	Windows	cp932
3	cmd.exe	CygWin bash	CygWin	UTF-8
4	MinTTY	CygWin bash	Windows	UTF-8
5	MinTTY	CygWin bash	CygWin	UTF-8

 

環境情報

いくつかの方法で情報を取得。(要 import platform,os,sys,locale)

#	メソッド	環境1	環境2	環境3	環境4	環境5
a	platform.system()	Windows	Windows	CYGWIN_NT-6.3	Windows	CYGWIN_NT-6.3
b	os.name	nt	nt	posix	nt	posix
c	os.environ.get('TERM')	None	cygwin	cygwin	xterm	xterm
d	os.environ.get('LANG')	None	ja_JP.UTF-8	ja_JP.UTF-8	ja_JP.UTF-8	ja_JP.UTF-8
e	sys.stdin.isatty()	True	True	True	False	True
f	sys.stdin.encoding	cp932	cp932	UTF-8	None	UTF-8
g	sys.stdout.isatty()	True	True	True	False	True
h	sys.stdout.encoding	cp932	cp932	UTF-8	None	UTF-8
i	locale.getpreferredencoding()	cp932	cp932	UTF-8	cp932	UTF-8

明快な判定方法は無いが、sys.stdout.encoding -> 環境変数LANG -> locale.getpreferredencoding() の順に信頼すれば良さそうに思える。

 

画面クリア

以下のメソッドの挙動を確認。(要 import subprocess)

#	メソッド	環境1	環境2	環境3	環境4	環境5
a	subprocess.call('cls', shell=True)	OK	OK	-	-	-
b	subprocess.call('cmd /c cls', shell=True)	-	-	OK	-	-
c	subprocess.call('clear', shell=False)	-	-	-	-	-
d	subprocess.call(['echo', '-en', r'\ec'], shell=False)	-	-	OK	-	OK
e	subprocess.call(r'echo -en "\ec"', shell=False)	-	OK	-	OK	-

なかなか判定条件が悩ましい。

　

キー入力の取得

msvcrt による取得が可能かどうかと、tty、curses が使えるかどうか調査。

#	メソッド	環境1	環境2	環境3	環境4	環境5
a	msvcrt.getch()	OK	OK	-	-	-
b	tty.setraw(0)	-	-	OK	-	OK
b	curses.screen#getch()	-	-	OK	-	OK

環境4 に限っては、1文字単位でキー入力を取得する手段が無いという結論となった。

　

calendar-cli - Command-line Interface for Google Calendar

calendar-cli - Googleカレンダーを操作するためのコマンドライン・インターフェース

 

Google カレンダーをコマンドラインで操作するニーズがあったので、元々あったスクリプトを CLI として作り変えた。




これを使えば、コマンドラインで

• 1日単位のサマリーの表示
• 予定の登録

が、できる。

 

コード

• mogproject/calendar-cli

 

セットアップ

Google の API を使う都合上、導入までに少し手間がかかる。

1. API プロジェクトの作成とclient_secret.json の入手

• Google Developers Console を開く (要Googleアカウントのログイン)
• 新規プロジェクトを作成
• Calendar API を有効化
  • API と認証 -> API -> Google Apps API -> Calendar API -> API を有効にする
• OAuth 2.0 認証のクライアントを作成 (デバイス種別は Other)
  • API と認証 -> 認証情報 -> OAuth 同意画面 -> サービス名 (calendar-cli でよい) を入力し保存
  • API と認証 -> 認証情報 -> 認証情報を追加 -> OAuth 2.0 クライアント ID
    -> アプリケーションの種類: その他として保存
• 画面から client_secret.json をダウンロードし保存
  • 実際のファイル名はもっと長い可能性あり

 

2. calendar-cli のインストール

pip で導入可能。環境によっては要sudo。

$ pip install calendar-cli
$ calendar-cli --version
calendar-cli 0.1.3
$ calendar-cli -h
Usage:
  calendar-cli [options]
                        Print a summary of events on the calendar.

  calendar-cli setup  [--read-only --credential ]
                        Generate a credentials file from the client secret.
                        You need a web browser to proceed.

  calendar-cli create [--date <YYYYMMDD> --start <HHMM> --end <HHMM>
                       --credential <credential_path>] <summary>
                        Create an event onto the calendar.


Options:
  --version             show program's version number and exit
  -h, --help            show this help message and exit
  --calendar=CALENDAR   set calendar id to CALENDAR (default:primary)
  --date=YYYYMMDD       set date to YYYYMMDD in the setup/create command
                        (default:today)
  --credential=CREDENTIAL
                        set credential path to CREDENTIAL
                        (default:/Users/user/.credentials/calendar-cli.json)
  --read-only           create a read-only credential file in the setup
                        command (default: False)
  --start=HHMM          set start time in the create command
  --end=HHMM            set end time in the create command
  --debug               enable debug logging (default: False)

 

3. 認証用ファイルの作成

calendar-cli でセットアップ用のコマンドを用意した。
client_secret.json の部分は、手順1でダウンロードした適切なパスに書き換えること。

$ calendar-cli setup client_secret.json

一度ブラウザ上で OAuth 認証が行われた後、~/.credentials/calendar-cli.json として認証情報が保存される。

 

使い方

 

1. サマリーの表示

• 今日の予定を表示 (デフォルトのカレンダー)
  

  $ calendar-cli

• 指定した日の予定を表示 (デフォルトのカレンダー)
  

  $ calendar-cli --date 12/6

  
  日付として指定可能なフォーマットは以下。年を省略すると現在の年で補完される。
  • 20151206 (YYYYmmdd)
  • 2015-12-6, 2015-12-06 (YYYY-m-d)
  • 2015/12/6, 2015/12/06 (YYYY/m/d)
  • 12-6-2015, 12-06-2015 (m-d-YYYY)
  • 12/6/2015, 12/06/2015 (m/d/YYYY)
  • 12-6, 12-06 (m-d)
  • 12/6, 12/06 (m/d)
• 指定したカレンダーの今日の予定を表示
  

  $ calendar-cli --calendar xxxxxx@group.calendar.google.com
  [終日] 有給休暇 (James LaBrie)
  [11:00-12:00] ○○様来社 (John Petrucci)
  [14:00-15:00] [外出] △△訪問 (John Myung)
  [17:30-22:00] □□勉強会 (Jordan Rudess)
  [17:30-23:00] ☆☆飲み会 (Mike Mangini)

 

2. イベントの登録

• 当日または翌日の 10:30 に 15分のイベントを作成
  

  $ calendar-cli create --start 10:30 内容
  イベントを作成しました: 2015-11-02 月 [10:30-10:45] 内容

  現在時刻が 10:30 以前であれば当日、以降であれば翌日にイベントが作成される。
  時刻指定のフォーマットは以下。
  • 1030 (HHMM)
  • 9:30, 09:30 (H:M)
• 当日または翌日に 12:00-13:00 のイベントを作成
  

  $ calendar-cli create --start 12:00 --end 13:00 内容
  イベントを作成しました: 2015-11-02 月 [12:00-13:00] 内容

  現在時刻が 12:00 以前であれば当日、以降であれば翌日にイベントが作成される。
• 日付を指定してイベントを作成
  

  $ calendar-cli create --date 12/6 --start 12:00 --end 13:00 内容
  イベントを作成しました: 2015-12-06 日 [12:00-13:00] 内容

  --date オプションのフォーマットはサマリー表示と同じ。
• 終日イベントを作成
  

  $ calendar-cli create --date 12/12 内容
  イベントを作成しました: 2015-12-12 土 [終日] 内容

 

Related Posts

• mog project: Reading Event Data on Google Calendar in Python

Type Assertion Decorator in Python

Python: 型チェックのデコレータ

 

Python は動的型付けの言語だが、抽象的なライブラリを書こうとすればどうしても型チェックの需要が生じる。

• PyContracts は非常に良さそうなライブラリだが、独自のクラスには対応していない模様。
• なので、車輪の再発明と知りつつも自前で実装してみた。

 

コード

• mog-commons-python/types.py at master · mogproject/mog-commons-python

 

インストール

mog-commons-python の一部として実験的にリリースしている。

pip install mog-commons

inspect#getcallargs をバックポートしたので、Python 2.6 / 2.7 / 3.2 / 3.3 / 3.4 / 3.5 の全てで動く。

 

使い方

関数定義と一緒に @types デコレータを指定するだけ。

• 1個以下の可変長引数として、戻り値の型を指定 (省略も可能)
• 名前付き引数として、パラメータ名=型 と制約を定義
• ListOf, DictOf などを使えば、コレクションの要素の型もチェックできる
• チェックに失敗したら、TypeError が発生

>>> from mog_commons.types import *

>>> @types(float, x=int, y=float, z=ListOf(int))
... def f(x, y, z):
...     return x * y + sum(z)
...

>>> f(1, 2, 3)
TypeError: y must be float, not int.

>>> f(1, 2.0, 3)
TypeError: z must be list(int), not int.

>>> f(1, 2.0, [3, 4])
9.0

戻り値だけのチェックも可能。

>>> @types(bool)
... def g(x):
...     return x
...

>>> g(3)
TypeError: must return bool, not int.

>>> g(True)
True

簡便のため、Python バージョンに透過的な String, Unicode, Option も定義した。
可変長引数、名前付き引数をチェックする場合は以下のようにする。

>>> @types(args=VarArg(String), kwargs=KwArg(int))
... def h(*args, **kwargs):
...     return 'ok'
...

>>> h('abc', 5, a=123, b=456)
TypeError: args must be tuple((basestring|str)), not tuple.

>>> h('abc', 'def', a=123, b=456)
'ok'

実際にどの要素の型が異なっているのか、までは表示していない。

 

これで、Typesafe Python にまた一歩近づいた。

Running Windows-OS CI for Python Projects on AppVeyor

AppVeyor を使って Windows 環境で Python プロジェクトの CI を回す

 

easy-menu を AppVeyor の CI に載せたので簡単にメモ。

• easy-menu - AppVeyor

AppVeyor とは

• Continuous Integration and Deployment service for Windows developers - Appveyor

Windows OS (デフォルトは Windows Server 2012) 環境で CI を実行してくれるサービス。

OSS は無料で制限なく使えるので、Windows 版 Travis CI のようなものと考えてよい。

セットアップ

プロジェクト追加

GitHub のアカウントで認証し、画面からプロジェクトを追加するだけ。

appveyor.yml

以下のリポジトリから、appveyor.yml と appveyor ディレクトリ配下のスクリプトを拝借した。

• ogrisel/python-appveyor-demo

基本的にそのままで動くが、バリエーションが多すぎるので適宜減らすとよい。
(最初、appv「a」yor.yml と typo していたため全く機能せず、焦った)

ハマりどころ

mock インストール失敗

環境の setuptools が古いせいか、mock ライブラリのインストールで失敗した。

• Requires setuptools >=17.1 to install properly. · Issue #261 · testing-cabal/mock

ワークアラウンドとして、'mock == 1.0.1' とバージョンを固定したら通るようになった。

そもそもテストが通らない

OS に対して透過的なテストを書くことは、本当に難しい。

Python: How to Implement Thread-Safe Auto-Increment in Redis

Python: Redis上でスレッドセーフなオート・インクリメントを実現する

 

目的

Redisで以下のような3種類のDBを使い、色々な名前の登録処理をしたい。

• 名前 -> ID の検索テーブル
• ID -> 名前 の検索テーブル
• 特定の key で ID の登録数 (=払い出したIDの最大値) を保持するテーブル

ID は 1 を起点とするオート・インクリメントなもので、名前ごとにユニークな数値を割り当てる。

 

インターフェース

1個の Redis インスタンスの中にある 3つの DBを使用する想定。それぞれのDB番号とカウンターとして使用するキーの名前を指定して初期化。

register メソッドは文字列として name を受け取り、その name に対応するユニークな ID を返す。name が DB に登録されていない場合のみ、登録処理を行う。

import redis


class Registerer(object):
    def __init__(self, counter_key, db_counter, db, db_invert=None,
                 host='localhost', port=6379):
        self.counter_key = counter_key
        self.db_counter = db_counter
        self.db = db
        self.db_invert = db_invert
        self.host = host
        self.port = port
        self.redis = redis.Redis(host, port, db)
        self.redis_cnt = redis.Redis(host, port, db_counter)
        self.redis_inv = (redis.Redis(host, port, db_invert)
                          if db_invert is not None else None)

    def register(self, name):
        return ???

 

テスト

マルチスレッド・プログラミングをする時はテストがないと不安なので、先にテストを書く。
(実行すると、ローカルの Redis のデータは全て消える)

#!/usr/bin/env python
import unittest
import threading

from repos.registerer import Registerer


class TestRegisterer(unittest.TestCase):
    def _clear(self):
        self.r.redis.flushall()

    def setUp(self):
        self.num_records = 10000
        self.r = Registerer('count', 15, 14, 13)
        self._clear()

    def tearDown(self):
        self._clear()

    def test_register_serial(self):
        r = self.r
        for i in range(self.num_records):
            r.register('user-%04d' % i)

        for i in range(10):
            r.register('user-%04d' % i)

        self.assertEqual(r.redis.dbsize(), self.num_records)
        self.assertEqual(r.redis_inv.dbsize(), self.num_records)
        self.assertEqual(int(r.redis_cnt.get('count')), self.num_records)

        for i in range(100):
            self.assertEqual(int(r.redis.get(r.redis_inv.get(i * 50 + 1))), i * 50 + 1)
            self.assertEqual(r.redis_inv.get(i * 50 + 1), 'user-%04d' % (i * 50))

    def test_register_parallel(self):
        r = self.r
        threads = [threading.Thread(target=r.register, args=('user-%04d' % i,)) for i in range(10000)]

        for t in threads:
            t.start()

        for t in threads:
            t.join()

        self.assertEqual(r.redis.dbsize(), self.num_records)
        self.assertEqual(r.redis_inv.dbsize(), self.num_records)
        self.assertEqual(int(r.redis_cnt.get('count')), self.num_records)

        for i in range(100):
            self.assertEqual(int(r.redis.get(r.redis_inv.get(i * 50 + 1))), i * 50 + 1)

            # this will fail
            # self.assertEqual(r.invert_redis.get(i * 50 + 1), 'user-%04d' % (i * 50))

    def test_register_parallel_same_id(self):
        r = self.r
        threads = [threading.Thread(target=r.register, args=('user-0000',)) for _ in range(self.num_records)]

        for t in threads:
            t.start()

        for t in threads:
            t.join()

        self.assertEqual(r.redis.dbsize(), 1)
        self.assertEqual(r.redis_inv.dbsize(), 1)
        self.assertEqual(int(r.redis_cnt.get('count')), 1)
        self.assertEqual(int(r.redis.get('user-0000')), 1)
        self.assertEqual(r.redis_inv.get(1), 'user-0000')


if __name__ == '__main__':
    unittest.main()

 

ナイーブな実装 (問題あり)

名前 -> ID のテーブルを検索し、値が取得できなかったら Redis の INCR コマンドで値を更新し、その ID を正引き/逆引きのテーブルそれぞれ登録する。

念の為に SETNX コマンドを利用し、キーが既に存在していたら例外を送出するようにする。

    def register_naive(self, name):
        index = self.redis.get(name)
        if index is None:
            index = self.redis_cnt.incr(self.counter_key)
            if not self.redis.setnx(name, index):
                raise Exception(
                    'Failed to register: db=%d, key=%s, value=%s' %
                    (self.db, name, index))
            if self.db_invert is not None:
                if not self.redis_inv.setnx(index, name):
                    raise Exception(
                        'Failed to register: db=%d, key=%s, value=%s' %
                        (self.db_invert, index, name))
        return int(index)

Redis の INCR はアトミックな処理であるし、一見問題はなさそうだが、先程書いたテストが失敗する。

.Exception in thread Thread-10001:
Traceback (most recent call last):
  File "/Users/xxxxxx/.pyenv/versions/2.7.9/lib/python2.7/threading.py", line 810, in __bootstrap_inner
    self.run()
  File "/Users/xxxxxx/.pyenv/versions/2.7.9/lib/python2.7/threading.py", line 763, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/xxxxxx/registerer.py", line 36, in register_naive
    raise Exception('Failed to register: db=%d, key=%s, value=%s' % (self.db, name, index))
Exception: Failed to register: db=14, key=user-0000, value=1

F.
======================================================================
FAIL: test_register_parallel_same_id (__main__.TestRegisterer)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "./tests/test_registerer.py", line 72, in test_register_parallel_same_id
    self.assertEqual(int(r.redis_cnt.get('count')), 1)
AssertionError: 2 != 1

----------------------------------------------------------------------
Ran 3 tests in 9.492s

FAILED (failures=1)

これは、同じ名前を同時に大量に登録した場合にのみ発生する。
名前 -> IDテーブルのチェックと更新の間のタイミングで競合状態が発生し、同じ名前の登録処理が複数実行されているためである。

 

トランザクションを利用した実装

redis-py には transaction というお誂え向きのヘルパーメソッドが用意されている。

パイプラインを引数に取る関数と、楽観的ロックをかけるキーを指定するだけでシンプルにトランザクションを記述できる。

最終的なコードは以下のようになった。

import redis


class Registerer(object):
    def __init__(self, counter_key, db_counter, db, db_invert=None,
                 host='localhost', port=6379):
        self.counter_key = counter_key
        self.db_counter = db_counter
        self.db = db
        self.db_invert = db_invert
        self.host = host
        self.port = port
        self.redis = redis.Redis(host, port, db)
        self.redis_cnt = redis.Redis(host, port, db_counter)
        self.redis_inv = (redis.Redis(host, port, db_invert)
                          if db_invert is not None else None)

    def register(self, name):
        def f(pipe):
            index = self.redis.get(name)
            if index is None:
                index = pipe.incr(self.counter_key)
                if not self.redis.setnx(name, index):
                    raise Exception(
                        'Failed to register: db=%d, key=%s, value=%s' %
                        (self.db, name, index))
                if self.db_invert is not None:
                    if not self.redis_inv.setnx(index, name):
                        raise Exception(
                            'Failed to register: db=%d, key=%s, value=%s' %
                            (self.db_invert, index, name))
            return int(index)

        return self.redis_cnt.transaction(
            f, self.counter_key, value_from_callable=True)

 

無事、テストも通った。

...
----------------------------------------------------------------------
Ran 3 tests in 13.818s

OK

 

References

• andymccurdy/redis-py - Pipelines

Python3.2: TypeError in subprocess Module When Using Bytes Command String

Python3.2: subprocess モジュールでコマンドラインを bytes で書くと TypeError が発生

 

事象

POSIX互換環境の Python 3.2 で subprocess#call を呼び出すとき
コマンドを bytes で書きつつ shell モードを有効にすると、以下の意味不明なエラーが出る。

$ python
Python 3.2.5 (default, Sep  1 2015, 23:07:07)
[GCC 4.2.1 Compatible Apple LLVM 6.1.0 (clang-602.0.53)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import subprocess
>>> subprocess.call(b'echo', shell=True)
(snip)
TypeError: Can't convert 'int' object to str implicitly
>>>

コマンドを str で渡せばよいのだが、それだとコマンドライン自体が SJIS などの非UTF-8で表現されている場合に困ってしまう。

 

原因

子プロセスを生成するところで、args に ['/bin/sh', '-c', 101, 99, 104, 111] というパラメータが渡っていた。
['/bin/sh', '-c'] + list(b'echo') のようなコードで作られたようだ。

1バイトごとに分解され int となったものが暗黙的に str に変換されずに TypeError が送出される。

Python 3.3 では修正済み。

• Issue #8513: On UNIX, subprocess supports bytes command string. · python/cpython@ae1258d
• Issue 8513: subprocess: support bytes program name (POSIX) - Python tracker

 

対応方法

Python 3.2 をサポートしないのが賢明と思われるが、以下のようなワークアラウンドでも回避できそうだ。

import subprocess


NOT_USE_SHELL = sys.version_info[:2] == (3, 2) and not sys.platform == 'win32'

cmd = b'echo'

if NOT_USE_SHELL:
    ret_code = subprocess.call(['/bin/sh', '-c', cmd], shell=False)
else:
    ret_code = subprocess.call(cmd, shell=True)

 

話題になったのが 5年以上前ということもあり、Google検索しても情報が出てこないのがつらい。

Easy Menu 1.0 is Now Released!

新しく生まれ変わった Easy Menu をお試しください

 

Easy Menu とは

• mogproject/easy-menu

Easy Menu とは、設定ファイルを元にターミナル上で動作するコマンドランチャーを立ち上げるだけの、とてもシンプルなツールです。

実に 2 年以上の間ずっと放置していたのですが意外と社内で好評でしたので、この際リニューアルしてみました。

• 動作イメージ
  

 

すぐに試す

今回、パッケージを PyPI に登録することができたので、インストールがとても簡単になりました。

設定ファイルを HTTP(S) 経由で取得すれば、たった 2行のコマンドで Easy Menu をお試しいただけます。

注意: 2015-08-31時点では、まだ Python 3.x系には対応していません
=> 2015-09-02 にリリースした v1.0.2 で Python 3.2/3.3/3.4 に対応しました

pip install easy-menu
easy-menu http://git.io/vGWla

• 短縮URLをお好みでない方は、こちらのURLをご利用ください。
  https://raw.githubusercontent.com/mogproject/easy-menu/master/easy-menu.example.yml
• ファイルの中身は こちら です。コマンドは全て echo コマンドなのでご安心ください。

 

主な起動方法

• easy-menu
  • パラメータを付けずにコマンドを実行した場合、カレントディレクトリから上位に向かって easy-menu.yml が存在するかチェックしていきます。そして最初に見つかった設定ファイルの内容がロードされます。 (Vagrant における Vagrantfile の検索と同じような動きです)
  • 環境変数 EASY_MENU_CONFIG を設定することで、検索対象のファイル名
    (デフォルトは easy-menu.yml) を変更することもできます。
  • プロジェクトのルートディレクトリに easy-menu.yml を置いておくと捗ります。
• easy-menu <設定ファイルのパス>
  • もちろん、設定ファイルの場所を直接指定することもできます。
    尚、コマンド実行時には、設定ファイルの存在するディレクトリにカレントディレクトリを移動してからコマンドが実行されます。
• easy-menu <設定ファイルのURL>
  • http(s) により設定ファイルをネットワーク上から取得します。
  • スキームは http, https のみ対応。省略はできません。

 

新機能

 

1行のコマンドでインストール

前述のとおり、pip コマンドでインストールできるようになりました。

YAML形式のサポート

これまでは JSON のみのサポートでしたが、YAML にも対応しました。

別ファイルのインクルード

"include" というキーワードを使うことで、別のファイルのメニューをマージすることが可能になりました。

設定イメージ:

---
メインメニュー:
  - include: common.yml
  - サブメニュー:
    - コマンド3: echo 3
    - コマンド4: echo 4

---
共通メニュー:
  - コマンド1: echo 1
  - コマンド2: echo 2

この場合、以下のようにメニューが生成されます。

---
メインメニュー:
  - 共通メニュー:
    - コマンド1: echo 1
    - コマンド2: echo 2
  - サブメニュー:
    - コマンド3: echo 3
    - コマンド4: echo 4

 

動的なメニュー生成

"eval" というキーワードを使うことで、任意のコマンドをメニュー生成時に実行し、その標準出力を YAML 形式の設定として読み込むことが可能になりました。

設定イメージ:

---
メインメニュー:
  - eval: scripts/create_ec2_menu.sh

この場合、シェルスクリプト create_ec2_menu.sh の出力がそのままメニューの内容として反映されます。

例えば、AWS で起動中の EC2 インスタンスの一覧を動的に取得するような使い方を想定しています。

国際化対応

現状、対応しているのは英語と日本語のみです。

 

以上、簡単な機能の紹介でした。
不具合、要望などがありましたら、お気軽に GitHub Issue に登録をお願いします。

Python: Getting Started with NetworkX Graph Generator

NetworkX のグラフジェネレーターで遊ぶ

 

out-of-the-box な機能が非常に豊富で感動した。

• Graph generators — NetworkX 1.9.1 documentation
• Gallery — NetworkX 1.9.1 documentation

せっかくなので、IPython Notebook でビジュアライズしてみる。

 

準備

全て必須という訳ではないが、一応インストールしておく。

pip install ipython jinja2 tornado pyzmq numpy scipy pylab matplotlib
pip install networkx

作業ディレクトリに移動して、IPython Notebook 起動。

ipython notebook

ブラウザ上で操作し、 New Notebook を作成。

 

実行例

コードの補完もできる。

ひととおりグラフを作ったあと保存し、そのファイルを Gist にアップロードしてみた。
すると以下のとおり、簡単にコードとイメージを共有することができる。

このカジュアルさに、また感動。Python 素晴らしい。

Writing Set of Sets in Python

Python: set を要素に持つ set を書く

 

普通に書くと、unhashable type (ハッシュ化できないデータ型) だと怒られる。

>>> {set(), {1}, {1, 2}}
Traceback (most recent call last):
  File "", line 1, in
TypeError: unhashable type: 'set'

内側の set(集合) に frozenset を使えばよい。

>>> {frozenset(), frozenset({1}), frozenset({1, 2})}
{frozenset({1, 2}), frozenset(), frozenset({1})}

fronzenset はイミュータブルなので、後から変更を加えることはできない。

>>> s = frozenset()
>>> s.add(1)
Traceback (most recent call last):
  File "", line 1, in
AttributeError: 'frozenset' object has no attribute 'add'

 

 

References

• 4. 組み込み型 — Python 3.4.3 ドキュメント

Python unittest Cheat Sheet

Python: unittest のチートシート

よく使うイディオムをまとめておく。

• Python 2.6 に対応するには、unittest2 を使うのが一番簡単

Safe and Smart Memoization in C++ (or Python)

エレガントなメモ化のやり方

 

同僚から教えてもらったテクニックを忘れそうになったのでメモ。

#include <cstdio>

int memo[100][100];

int main() {
  memset(memo, -1, sizeof memo);
  // call f with parameters
  f(???, ???)
}

int f(int x, int y) {
  int& res = memo[x][y];
  if (res < 0) {
    // do the work when the value has not been calculated
    res = ???
  }
  return res;
}

適当なn次配列をメモ化の領域として使い、1のビット(-1)で初期化するところまでは一般的なやり方。

ポイントは、実際にメモ化したデータを使う部分。
はじめに配列の要素の参照を取り、初期状態でない場合にのみその値を更新する。
いずれのケースでも最後にその参照を返す。

これで処理の流れがシンプルになるし、メモの更新し忘れを防ぐこともできる。




同様に、Python で値が None で初期化されているならこんな感じ。

def f():
    if cache is None:
        # do the work
        cache = ???
    return cache

Counting True in the Python List

Python: リスト内の True の個数を数える

 

sum を使えば、True: 1, False: 0 としてカウントされるので便利。

>>> [False, True, True, False, True].count(True)
3
>>> sum([False, True, True, False, True])
3
>>> sum(x % 2 == 1 for x in range(5))
2

 

 

References

• Counting the number of True Booleans in a Python List - Stack Overflow