python ライブラリ等の雑多なサンプルコード
[Last Modified] (2019/09/17 14:38:51)
ここは
趣味のプログラミングを楽しむための情報共有サービス。記事の一部は有料設定にして公開できます。 詳しくはこちらをクリック📝
Recent posts
Popular pages

概要

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

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 で確認できるとおり、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 threading import Thread
from threading import Event
from time import sleep

event = Event()

threads = []

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

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

sleep(10)

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

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'

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

If you want to read the rest of this page

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

残り文字数は全体の約 33 %
なつかしのねこ
100 円
Related pages
    概要 pcap (packet capture) というパケットキャプチャ用APIの実装には、UNIX系のlibpcapとWindowsのWinPcapがあります。これらのライブラリを利用したアプリケーションとしては、UNIX系のtcpdumpとWindowsのWiresharkが有名です。また、スクリプト言語にはlibpcap/WinPcapのラッパーが存在しており、パケットキャプチャアプリケ
    概要 立方体を二つ配置して回転させてみます。ライブラリを用いずに OpenGL API を直接利用します。描画部分のみを IPython で検証するためのソースコードはこちらです。 wget https://gist.githubusercontent.com/harubot/df886254396a449038ee542ed317f7b3/raw/92216e02d0210b9d8177056
    概要 OpenGL のフレームバッファには複数のカラーバッファを割り当てることができます。フレームバッファはカラーバッファの他にデプスバッファとステンシルバッファを持ちます。これらバッファはすべてレンダーバッファとよばれるメモリ領域です。フラグメントシェーダで out として得られる出力はカラーバッファに格納されます。一つのフラグメントシェーダに
    概要 コンピュータグラフィックスのレンダリングライブラリの一つである OpenGL を Python3 から利用するための Linux 環境を準備して、設定例およびサンプルコードを記載します。特にここでは Debian9 を利用します。 Getting Started Language bindings / Python
    概要 C++ を Python から利用する方法の一つに pybind11 があります。C++11 をサポートするコンパイラが必要です。サンプルコードを記載します。 pybind11 — Seamless operability between C++11 and Python Reference 簡単なサンプル
    バイナリファイルから文字列を抽出する (strings) 文字列らしい部分をバイナリファイルから抽出して出力するコマンドです。得体の知れないコマンドのオプションを調べたり strings /usr/bin/gcc | grep ^-- --help --target-help --sysroot= --all-warnings --ansi --assemble --assert --cla
    概要 GNU Debugger (GDB) の簡単な使い方を記載します。 適宜参照するための公式ドキュメント 起動方法 main.cpp #include <iostream> #include <unistd.h> using namespace std; int main() { int i = 0; while(true) { sleep(1);