python ライブラリ等の雑多なサンプルコード
[履歴] [最終更新] (2018/12/23 22:58:40)
1
作品
409
技術情報
最近の投稿
ここは
趣味の電子工作を楽しむ人のためのハードウェア情報共有サイト

技術情報や作品の投稿機能、リアルタイム遠隔操作 API をご利用いただけます。
新着作品

概要

よく使う python ライブラリのサンプルコード集です。

JSON

#!/usr/bin/python
# -*- coding: utf-8 -*-

import json

arr = [1, 2, {'xxx': 3}]

# オブジェクト ←→ JSON 文字列
jsonStr = json.dumps(arr)
arr2 = json.loads(jsonStr)

# オブジェクト ←→ IO
fp_w = open('./tmp.json', 'w') # https://docs.python.org/3/library/functions.html#open
json.dump(arr, fp_w)
fp_w.close()

fp_r = open('./tmp.json', 'r')
arr3 = json.load(fp_r)
fp_r.close()

環境変数の取得

import os
os.environ.get('HOME')

代入もできます。

In [5]: os.environ['MYENV'] = '123'
In [6]: os.environ.get('MYENV')
Out[6]: '123'
In [8]: os.system('env | grep MYENV')
MYENV=123

path 操作

import os
import os.path

# 指定した path 内のファイルおよびディレクトリをリストとして取得 https://docs.python.org/3/library/os.html#os.listdir
# (python3系の場合は標準で、より高機能な os.scandir() が使えます https://docs.python.org/3/library/os.html#os.scandir )
os.listdir('..')

# 絶対パス化
os.path.abspath('.')

# path の結合
os.path.join('/', 'path', 'to', 'somewhere.txt')

オブジェクト等の情報を取得 inspect

以下の例では inspect モジュールのファイルの場所を inspect.getfile() で取得しています。関連して、ファイルの更新日時 mtime も取得できます。

import inspect
import os
print(inspect.getfile(inspect)) #=> /usr/lib/python2.7/inspect.pyc
print(os.path.getmtime(inspect.getfile(inspect))) #=> 1538182044.96

time

import time

# スリープ
time.sleep(1.0)

# unix タイムスタンプ
int(time.time())

組み込み関数

class MyClass(object):
    pass
obj = MyClass()
isinstance(obj, MyClass)
isinstance({}, dict)
isinstance([], list)
isinstance("", str)

ロガー

getattr() を併用して、動的にログレベルを設定できます。

#!/usr/bin/python
# -*- coding: utf-8 -*-

from logging import getLogger, Formatter, StreamHandler
import logging

logger = getLogger(__name__)

def Main():
    loglevel = 'DEBUG' # ERROR, WARNING, INFO, DEBUG
    handler = StreamHandler()
    handler.setFormatter(Formatter('%(asctime)s - %(levelname)s - %(message)s'))
    logger.addHandler(handler)
    logger.setLevel(getattr(logging, loglevel))

    try:
        raise RuntimeError
    except ZeroDivisionError:
        pass
    except:
        logger.exception("xxx")
    else:
        pass
    finally:
        logger.info("xxx")

if __name__ == '__main__':
    Main()

出力例

$ python main.py
2018-06-21 01:11:19,387 - ERROR - xxx
Traceback (most recent call last):
  File "main.py", line 17, in Main
    raise RuntimeError
RuntimeError
2018-06-21 01:11:19,387 - INFO - xxx

バイナリデータの処理

ファイルに格納されたバイナリデータや、ネットワーク経由でやり取りするバイナリデータの処理に利用できます。バイナリデータは bytes 型として扱います。

standard または native の指定

バイナリデータには、CPU 等に依存するバイトオーダアラインメントという概念があります。ネットワーク経由では常にビッグエンディアンでデータをやり取りする必要があります。バイトオーダは以下のようなコードで確認できます。

import sys
print sys.byteorder # little (CPU依存)

無指定時は実行環境に依存した native 設定になりますが、環境に依らない処理を実装するためには @ 以外のフォーマット指定を行います

#!/usr/bin/python
# -*- coding: utf-8 -*-

from struct import calcsize

print calcsize('@L') # 8 ←sizeof(unsigned long) の値
print calcsize('!L') # 4

pack, unpack

from struct import pack, unpack

mybytes = pack('hhl', 1, 2, 3) # hhl は 2hl とも記載できます。
isinstance(mybytes, bytes) # True

mytuple = unpack('hhl', mybytes) # 必ずタプルで返ります
print mytuple # (1, 2, 3)

コードチェック flake8

Ruby の rubocop のようなものです。

sudo pip install flake8
flake8 /path/to/code/to/check.py

特定のエラーコードを無視したい場合

--ignore E121,E123

特定のファイルを無視したい場合

--exclude tests/*

これらを設定ファイルとして利用したい場合

~.flake8

[flake8]
filename = *.py
ignore = E121,E123
exclude = tests/*

setup.pyプラグインを新規に追加することで、独自のエラーコードを発行できます。プラグイン内では ast で表現されたソースコードの検証を行います

IPython 関連

オブジェクト調査

In [4]: str?
Type:       type
String Form:<type 'str'>
Namespace:  Python builtin
Docstring:
str(object) -> string

Return a nice string representation of the object.
If the argument is a string, the return value is the same object.

Ruby の pry のようにデバッガを仕掛ける

import IPython

...
IPython.embed() # ipython 等が起動される (ctrl-d 等で抜けると continue)

...
IPython.embed() # ipython 等が起動される (ctrl-d 等で抜けると continue)

Out[N] は変数として参照できます。

In [1]: 123
Out[1]: 123

In [2]: 1
Out[2]: 1

In [3]: print Out[1] + Out[2]
124

Re-raise について

raise e とすると例外が新規にコピーされるため、スタックトレースが新規に始まります。raise とすることで、スタックトレースごと re-raise できます。

main.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

import logging

logging.basicConfig()
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

def f():
    raise Exception("from f")

def g():
    try:
        f()
    except Exception as e:
        raise e

def h():
    try:
        f()
    except Exception as e:
        raise

def Main():
    try:
        g()
    except Exception as e:
        logger.exception("%r", e)

    try:
        h()
    except Exception as e:
        logger.exception("%r", e)

if __name__ == '__main__':
    Main()

実行例

$ python main.py 

ERROR:__main__:Exception('from f',)
Traceback (most recent call last):
  File "main.py", line 27, in Main
    g()
  File "main.py", line 17, in g  ←g から始まっています
    raise e
Exception: from f

ERROR:__main__:Exception('from f',)
Traceback (most recent call last):
  File "main.py", line 32, in Main
    h()
  File "main.py", line 21, in h
    f()
  File "main.py", line 11, in f  ←f から始まっています
    raise Exception("from f")
Exception: from f

%r%s の違い

%r__repr__ を実行した結果を利用します。オブジェクトを文字列表現した内容が返ります。%s__str__ を実行した結果を利用します。__repr__ と異なり、オブジェクトを表現する文字列ではなく、文字列として処理する際に便利なフォーマットが返ります。そのため、__repr__ の実装は期待できますが、__str__ の実装は必ずしも期待できません。

main.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

import json

class MyClass(object):
    def __init__(self):
        self.x = { 'xxx': 123 }
    def __repr__(self):
        return "MyClass(x = %s)" % json.dumps(self.x)
    def __str__(self):
        return json.dumps(self.x)

def Main():
    obj = MyClass()
    print "%r" % obj # repr(obj) としても同じ
    print "%s" % obj # str(obj) としても同じ

if __name__ == '__main__':
    Main()

実行例

$ python main.py
MyClass(x = {"xxx": 123})
{"xxx": 123}

相対パスによる、同じパッケージ内モジュールのインポート

同じ階層にある myrelmodule から myfunc をインポートしたい場合の例

from .myrelmodule import myfunc

シグナルハンドラ

システムコールの一つ signal を内部的に利用するライブラリを利用して、main python スレッドにシグナルハンドラを設定できます。

main.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from signal import signal, SIGTERM, SIGINT

def signal_handler(signum, frame):
    print('Signal handler called with signal', signum)

def Main():
    signal(SIGTERM, signal_handler) # kill 等
    signal(SIGINT, signal_handler) # ctrl-c 等
    while True:
        pass

if __name__ == '__main__':
    Main()

実行例

$ python main.py
^C('Signal handler called with signal', 2)  ← ctrl-c
^C('Signal handler called with signal', 2)
^C('Signal handler called with signal', 2)
('Signal handler called with signal', 15)  ← 別ターミナルから kill

man 2 signal で確認できるとおり、SIGKILLSIGSTOP については設定できません。シグナル一覧は kill -l で確認できます。

外部プログラムの実行時はハンドリングできない

libmy.c

#include <stdio.h>
#include <unistd.h>

void hello() {
    int i;
    for(i = 0; i < 10; ++i) {
        printf("hello world!\n");
        sleep(1);
    }
}

ビルド

gcc -shared -o libmy.so libmy.c

外部プログラムを実行中は Python のイベントハンドラは呼ばれません。例えば、上記ライブラリを ctypes 等で実行している間は、設定したシグナルハンドラは実行されません。

A Python signal handler does not get executed inside the low-level (C) signal handler. Instead, the low-level signal handler sets a flag which tells the virtual machine to execute the corresponding Python signal handler at a later point(for example at the next bytecode instruction). This has consequences:

main.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from ctypes import CDLL
from signal import signal, SIGTERM, SIGINT, SIGKILL

def signal_handler(signum, frame):
    print('Signal handler called with signal', signum)

def Main():
    signal(SIGTERM, signal_handler)
    signal(SIGINT, signal_handler)
    libmy = CDLL('./libmy.so')
    libmy.hello()

if __name__ == '__main__':
    Main()

実行例

$ python main.py
hello world!
hello world!  ← ここで SIGTERM を送信
hello world!
hello world!
hello world!
hello world!
hello world!
hello world!
hello world!
hello world!
('Signal handler called with signal', 15)  ← libmy から処理が戻ってからハンドリングされます。

外部プログラム実行中であっても、SIGTERM または SIGSTOP であれば停止できます。そのため、subprocess で子プロセスを生成して外部プログラムを実行すれば、子プロセスごとまとめて SIGKILL できるということになります。

スレッド

マルチスレッドプログラミングを行うためには threading ライブラリ等を利用します。Thread() でスレッドを作成して start() で開始します。開始したスレッドの終了を待つためには join() を利用します。

main.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from threading import Thread
from time import sleep

class MyClass(object):
    def __init__(self):
        self._worker = None

    def RunAndDetach(self):
        self._worker = Thread(target = self._Run, name = "MyClass worker")
        self._worker.start()

    def _Run(self):
        for i in range(5):
            print "hi from MyClass"
            sleep(1)

    def Wait(self):
        self._worker.join()

def Main():
    obj = MyClass()
    obj.RunAndDetach()
    # obj.Wait()
    print "hi from Main()"

if __name__ == '__main__':
    Main()

実行例

$ python main.py
hi from MyClass
hi from Main()
hi from MyClass
hi from MyClass
hi from MyClass
hi from MyClass

Wait する場合

$ python main.py
hi from MyClass
hi from MyClass
hi from MyClass
hi from MyClass
hi from MyClass
hi from Main()

共通リソースを複数のスレッドから利用する場合

グローバル変数や外部 DB 等を複数スレッドから利用する場合は、Java synchronized 等のように Lock() する必要があります。

main.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from threading import Thread, Lock
from time import sleep

count = 0
lock = Lock()

def incr(caller):
    global count
    print "Acquiring %s %d" % (caller, count)
    with lock:
        print "Acquired %s %d" % (caller, count)
        count += 1
        sleep(1)
        print "Releasing %s %d" % (caller, count)

def f():
    while count < 5:
        incr('f')

def g():
    while count < 5:
        incr('g')

def Main():
    workerF = Thread(target = f)
    workerG = Thread(target = g)
    workerF.start()
    workerG.start()

if __name__ == '__main__':
    Main()

実行例

$ python main.py 
Acquiring f 0
Acquired f 0
Acquiring g 1
Releasing f 1
Acquiring f 1
Acquired g 1
Releasing g 2
Acquiring g 2
Acquired g 2
Releasing g 3
Acquiring g 3
Acquired f 3
Releasing f 4
Acquiring f 4
Acquired f 4
Releasing f 5
Acquired g 5
Releasing g 6

CPython の場合、Java 等のスレッドと異なり、一つの CPU コアで複数のスレッドを交互に実行します。そのため、I/O バウンドの場合等は別として、CPU バウンドの場合はマルチスレッド化によっては解決できません。

CPython implementation detail: In CPython, due to the Global Interpreter Lock, only one thread can execute Python code at once (even though certain performance-oriented libraries might overcome this limitation). If you want your application to make better use of the computational resources of multi-core machines, you are advised to use multiprocessing or concurrent.futures.ProcessPoolExecutor. However, threading is still an appropriate model if you want to run multiple I/O-bound tasks simultaneously.
https://docs.python.org/3/library/threading.html

プリントデバッグ

from pprint import pprint
pprint('xxx')

Python モジュールのパスを確認

import json
json.__path__
json.__file__

外部コマンドを実行して結果を受け取る

from os import popen
cmd = 'date'
res = popen(cmd).read().strip()
print res

Sat Sep  1 23:31:21 JST 2018

check_output でも同様のことができます。

In [1]: from subprocess import check_output
In [2]: check_output(['echo', 'hello'])
Out[2]: 'hello\n'

外部コマンドを実行して終了ステータスを取得する

単に外部コマンドを実行するためには os.system() が便利ですが、返り値は子プロセスの終了ステータスとその他の情報をエンコードした値になっており、今回の実行環境では 256 倍された値となっています。

$ echo 'from os import system
> print system("exit 0")' | python
0

$ echo 'from os import system
print system("exit 1")' | python
256

$ echo 'from os import system
print system("exit 2")' | python
512

bash は 256 以上や負の値の終了ステータスを以下のように stauts mod 256 処理します。

$ bash -c 'exit 255'; echo $?
255
$ bash -c 'exit 256'; echo $?
0
$ bash -c 'exit 257'; echo $?
1
$ bash -c 'exit -1'; echo $?
255

そのため、今回の環境で python の system の返り値をシェルにそのまま返すと、異常終了 1 が正常終了 0 として判定されてしまいます。

$ echo 'from os import system
> status = system("exit 1")
> exit(status)' | python; echo $?
0

子プロセスの終了ステータスを正しく処理するためには subprocess を利用します。

python2 call

$ echo 'from subprocess import call
> status = call("exit 1", shell=True)
> print(status)' | /usr/bin/python2
1

python3 run

$ echo 'from subprocess import run
> status = run("exit 1", shell=True).returncode
> print(status)' | /usr/bin/python3
1

ステートマシン transitions

状態遷移図等で表現される有限オートマトン (FMS; finite state machine) の Python による実装の一つに transitions ライブラリがあります。

インストール

pip install transitions

サンプルコード

#!/usr/bin/python
# -*- coding: utf-8 -*-

from transitions import Machine, State

class MyClass(object):

    def __init__(self):
        # 状態と遷移
        self._states = [
            State(name='angry'),
            State(name='happy', on_enter=['say_happy'], on_exit=['say_unhappy'])
        ]
        self._transitions = [
            { 'trigger': 'be_happy', 'source': 'angry', 'dest': 'happy' },
            { 'trigger': 'be_angry', 'source': 'happy', 'dest': 'angry' }
        ]
        # FSM 初期化
        self.machine = Machine(model=self, states=self._states, transitions=self._transitions, initial='angry')

    def say_unhappy(self):
        print('become unhappy');

    def say_happy(self):
        print('become happy');

def Main():
    obj = MyClass()
    print obj.state #=> angry

    obj.be_happy() #=> become happy
    print obj.state #=> happy
    print obj.is_happy() #=> True

if __name__ == '__main__':
    Main()

遷移条件の設定

conditions を設定します。

#!/usr/bin/python
# -*- coding: utf-8 -*-

from transitions import Machine, State

class MyClass(object):

    def __init__(self):
        self._states = [
            State(name='solid'),
            State(name='liquid'),
            State(name='gas')
        ]
        self._transitions = [
            { 'trigger': 'heat', 'source': '*', 'dest': 'gas', 'conditions': 'is_flammable' },
            { 'trigger': 'heat', 'source': '*', 'dest': 'liquid' },
        ]
        self.machine = Machine(model=self, states=self._states, transitions=self._transitions, initial='solid')

    def is_flammable(self, temp):
        return temp >= 100

def Main():
    obj = MyClass()

    obj.heat(temp = 74)
    print obj.state #=> liquid

    obj.heat(temp = 100)
    print obj.state #=> gas

if __name__ == '__main__':
    Main()

遷移時のコールバック

beforeafter を設定します。

#!/usr/bin/python
# -*- coding: utf-8 -*-

from transitions import Machine, State

class MyClass(object):

    def __init__(self):
        self._states = [
            State(name='angry'),
            State(name='happy')
        ]
        self._transitions = [
            { 'trigger': 'be_happy', 'source': 'angry', 'dest': 'happy', 'before': 'say_will_be_happy', 'after': 'say_became_happy' },
            { 'trigger': 'be_angry', 'source': 'happy', 'dest': 'angry' }
        ]
        self.machine = Machine(model=self, states=self._states, transitions=self._transitions, initial='angry')

    def say_will_be_happy(self):
        print('will be happy')

    def say_became_happy(self):
        print('became happy')


def Main():
    obj = MyClass()
    obj.be_happy() #=> will be happy\nbecame happy

if __name__ == '__main__':
    Main()

関数の結果をキャッシュして高速化 lru_cache

Python3.3 以降の functools.lru_cache を利用すると、関数の結果をキャッシュして高速化できます。Python 2 系の場合はバックポートされた backports.functools_lru_cache を利用します。

pip install backports.functools_lru_cache==1.3

例えば以下のようにインポートします。

try:
    from functools import lru_cache
except ImportError:
    from backports.functools_lru_cache import lru_cache

import inspect
print(inspect.getfile(lru_cache)) #=> /usr/local/lib/python2.7/dist-packages/backports/functools_lru_cache.py

デコレータを付与するだけで関数の引数をキーとして実行結果がキャッシュされるようになります。maxsizeNone を指定すると、すべてのキーに対して無限にキャッシュが生成されます。

@lru_cache(maxsize=32)
def fib(n):
    if n < 2:
        return n
    return fib(n-1) + fib(n-2)

def _fib(n):
    if n < 2:
        return n
    return _fib(n-1) + _fib(n-2)

ipython の %%timeit 等で実行時間を計測してみます。キャッシュが効いていることが分かります。

In [5]: %%timeit
   ...: [fib(n) for n in range(32)]
   ...: 
10000 loops, best of 3: 78.4 µs per loop

In [6]: %%timeit
   ...: [_fib(n) for n in range(32)]
   ...: 
1 loop, best of 3: 1.17 s per loop

In [7]: fib.cache_info()
Out[7]: CacheInfo(hits=1315580, misses=32, maxsize=32, currsize=32)

その他のライブラリ例

関連ページ
    概要 pcap (packet capture) というパケットキャプチャ用APIの実装には、UNIX系のlibpcapとWindowsのWinPcapがあります。これらのライブラリを利用したアプリケーションとしては、UNIX系のtcpdumpとWindowsのWiresharkが有名です。また、スクリプト言語にはlibpcap/WinPcapのラッパーが存在しており、パケットキャプチャアプリケ
    概要 コンピュータグラフィックスのレンダリングライブラリの一つである OpenGL を Python3 から利用するための Linux 環境を準備して、設定例およびサンプルコードを記載します。特にここでは Debian9 を利用します。 Getting Started Language bindings / Python
    バイナリファイルから文字列を抽出する (strings) 文字列らしい部分をバイナリファイルから抽出して出力するコマンドです。得体の知れないコマンドのオプションを調べたり $ strings /usr/bin/gcc | grep ^-- --help --target-help --sysroot= --all-warnings --ansi --assemble --assert --cl