python ライブラリ等の雑多なサンプルコード

[最終更新] (2020/07/11 19:47:52)

プログラミング/IoT の関連商品 (Amazonのアソシエイトとして、当メディアは適格販売により収入を得ています。)

Nintendo Switch Lite イエロー

オープンソース・ソフトウェアルータVyatta入門　――実践ルーティングから仮想化まで (Software Design plus)

ROSロボットプログラミングバイブル

ポケットモンスターソード -Switch

回路シミュレータでストンとわかる!最新アナログ電子回路のキホンのキホン

tybot

概要

よく使う 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

コンテキストを切り換える contextlib

contextlib を利用すると with 内でのみ処理を切り換えることができます。

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

from contextlib import contextmanager

def Main():
    with MyContext():
        print('hi')

@contextmanager
def MyContext():
    print('context started')
    try:
        yield
    finally:
        print('context ended')

if __name__ == '__main__':
    Main()

実行例

$ python3 sample.py 
context started
hi
context ended

time

import time

# スリープ
time.sleep(1.0)

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

Python2 でタイムスタンプを UTC に変換

from datetime import datetime
from dateutil.tz import tzlocal
from dateutil.tz import tzutc

unixtime = 1558922100
datetime.fromtimestamp(unixtime, tzlocal()).astimezone(tzutc()).strftime('%Y-%m-%d %H:%M:%S')

Python3 でタイムスタンプを UTC/JST に変換

from datetime import datetime
from datetime import timezone
from datetime import timedelta

unixtime = 1558922100

datetime.fromtimestamp(unixtime, timezone(timedelta(hours=+0), 'UTC')).strftime('%Y-%m-%d %H:%M:%S')
datetime.fromtimestamp(unixtime, timezone(timedelta(hours=+9), 'JST')).strftime('%Y-%m-%d %H:%M:%S')
datetime.fromtimestamp(unixtime, timezone(timedelta(hours=+9), 'JST')).isoformat()

Python3 で ISO 時刻文字列をタイムスタンプに変換

from dateutil.parser import parse

isodatetime = '2019-05-27T10:55:00+09:00'

int(parse(isodatetime).timestamp())

組み込み関数

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

SSH 先のターミナルにコピーする場合等、%paste、%cpaste、%run でうまくいかない際は %autoindent を利用すると良いです。

In [7]: %autoindent
Automatic indentation is: OFF

In [8]: if True:
   ...:     print 'ok'
   ...: if True:
   ...:     print 'ok'
   ...: 
ok
ok

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 で確認できるとおり、SIGKILL と SIGSTOP については設定できません。シグナル一覧は 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:

A long-running calculation implemented purely in C (such as regular expression matching on a large body of text) may run uninterrupted for an arbitrary amount of time, regardless of any signals received. The Python signal handlers will be called when the calculation finishes. https://docs.python.org/3/library/signal.html#execution-of-python-signal-handlers

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 threading import Thread
from threading import Event
from time import sleep

event = Event()

threads = []

def target():
    print 'waiting...'
    event.wait(timeout=5)
    print 'done'
    event.clear()

for i in range(2):
    thread = Thread(target=target)
    threads.append(thread)
    thread.start()

sleep(10) #

print 'hi from main'
#event.set()

for thread in threads:
    thread.join()

実行例

waiting...
waiting...
done
done
hi from main

sleep せずに set する場合

waiting...
waiting...
hi from main
done
done

プリントデバッグ

from pprint import pprint
pprint('xxx')

整形

from json import dumps
print dumps({'a':123}, indent=2)

スタックトレースの確認

import traceback
traceback.print_stack()

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'

subprocess の Popen を利用して、パイプを作成して標準入力を Python から与えることもできます。

from subprocess import Popen
from subprocess import PIPE

text = 'hello'
output = Popen(['cat', '-'], stdin=PIPE, stdout=PIPE).communicate(text.encode('utf-8'))

In [5]: output
Out[5]: ('hello', None)

TTY 端末の存在確認

In [2]: from sys import stdin

In [3]: stdin.isatty()
Out[3]: True

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

この続きが気になる方は

python ライブラリ等の雑多なサンプルコード

残り文字数は全体の約 33 %

tybot

100 円

* プログラミング情報の有料公開およびご購入について

* ご購入済みの方はこちらからログイン

概要

コンテキストを切り換える contextlib

Python2 でタイムスタンプを UTC に変換

Python3 でタイムスタンプを UTC/JST に変換

Python3 で ISO 時刻文字列をタイムスタンプに変換

standard または native の指定

pack, unpack

IPython 関連

Re-raise について

%r と %s の違い

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

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

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

実行したスレッドをタイムアウトを設定して待機させる

スタックトレースの確認

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

TTY 端末の存在確認

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

python ライブラリ等の雑多なサンプルコード

`%r` と `%s` の違い