[Python] ライブラリ別メモ

単語帳．毎回検索するのが面倒なので転載多め．元URLあり．

beautifulsoup

Qiita@itkr: PythonとBeautiful Soupでスクレイピング

text vs string

いるかのボックス: Beautifulsoup4のtextとstringの違い

画像の保存

ゼロイチ: Python3で画像をスクレイピングしてローカルに保存する|BeautifulSoupを利用
西住工房: 【Python/BeautifulSoup】Webサイトの画像を自動で一括ダウンロード

concurrent.futures

concurrent.futures

テンプレート

ProcessPoolExecutor→ThreadPoolExecutorでマルチスレッド．

from concurrent import futures
with futures.ProcessPoolExecutor(max_workers=None) as executor:
    for *:
        executor.submit(myfunc, arg0, arg1, ...)

並列化のタイミング: submit, mapの後

ProcessPoolExecutorもThreadPoolExecutorも同じ．

並列化できない例1: ジェネレータ自体

from concurrent import futures
import time

max_workers = 8

def mygenerator(args):
    '''重い処理を挟むジェネレータ'''
    for arg in args:
        time.sleep(1)
        yield arg

for i in mygenerator(range(max_workers)):
    print(i) # 8秒かかる

submitでもmapでも所要時間は変わらない:

#with futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
with futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    for i in mygenerator(range(max_workers)):
        print(i)

#with futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
with futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    executor.map(print, mygenerator(range(max_workers)))

勿論submitした後は並列化されるので，それ以降の処理が重い場合は効果あり．

並列化できない例2: submitする引数

以下の例でも所要時間は短縮されない．

def myfunc(arg):
    '''重い処理を挟む関数'''
    time.sleep(1)
    return arg

#with futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
with futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    for i in range(max_workers):
        executor.submit(print, myfunc(i))

重い処理自体を並列化してしまえばOK．
例えば以下の例は並列化される (およそ8秒→1秒になる)

#with futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
with futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    for i in range(max_workers):
        executor.submit(time.sleep, 1)
        executor.submit(print, i)

ベター(?)プラクティス

これら並列化されるタイミングを踏まえると，
共通化可能な処理をラップする関数を別に作り，それを並列化するのが良いと考えられる．
例: 「ファイルを読む (様々な形式や関数・モジュール) →そのトレンドを計算 (これは共通化可能)」を大量にやる場合，
　「」内を行う関数をそれぞれ作成し，それを並列化する:

メモ

submitの返り値はFutureクラス．変数futureで受けるのが一般的．
これをリストなどにまとめるときはfuturesとしたくなるが，名前が競合する場合はfsで．
(公式ドキュメントのfutures.as_completed(fs)参照)

• ThreadPoolExecutor: 軽い処理用, 1つのCPUが使われる
• ProcessPoolExecutor: 重い処理用, 複数のCPUが使われる

※並列状況はtopコマンドで監視可能．
　重い処理にThreadPoolExecutorを使うとCPU使用率が100%を超える．

※(エラーも出力せず) 動かないのはデッドロックの可能性あり．
　自作クラスのメソッドの中にある関数を，同じファイル直属の関数にするだけで解決することも．
　その他デッドロックは"hang"などで検索．
stackoverflow: Python doctest hangs using ProcessPoolExecutor

参考
Qiita@tag1216: Pythonでconcurrent.futuresを使った並列タスク実行

その他並列処理モジュールとの比較
Heavy Wetal: concurrent.futures — 並行処理 in Python
Qiita@simonritchie: Pythonの並列処理・並行処理をしっかり調べてみた

itertools

Python: itertools

ちなみにイテレータ⊃ジェネレータ

イテレータをn個のサブイテレータに均等に分割する

itertools.isliceを使う．
期待する動作:

• イテレータ -> イテレータ(0), イテレータ(1), ..., イテレータ(n-1)
• 元のイテレータをn個おきにとってくる
  • (先頭からm個ずつ，という分割は全体の長さが必要)

iters = range(12)

n = 3
for s in range(n):
    print(s, list(itertools.islice(iters, s, None, n)))
# 0 [0, 3, 6, 9]
# 1 [1, 4, 7, 10]
# 2 [2, 5, 8, 11]

n = 5
for s in range(n):
    print(s, list(itertools.islice(iters, s, None, n)))
# 0 [0, 5, 10]
# 1 [1, 6, 11]
# 2 [2, 7]
# 3 [3, 8]
# 4 [4, 9]

chain: 複数のイテレータを結合する

n = 3
s0, s1, s2 = 0, n + 1, n + n + 2
iter0 = (i for i in range(s0, s0 + n))
iter1 = (i for i in range(s1, s1 + n))
iter2 = (i for i in range(s2, s2 + n))

for i in itertools.chain(iter0, iter1, iter2):
    print(i) # 0 1 2 4 5 6 8 9 10

stackoverflow: How to join two generators in Python?

cycle: 繰り返しイテレータを回す

vals = ['a', 'b', 'c']
for i, val in enumerate(vals):
    print(i, val)
# 0 a
# 1 b
# 2 c

vals = ['a', 'b', 'c']
#vals = (val for val in vals) # ジェネレータでも同じ
for i, val in enumerate(itertools.cycle(vals)):
    print(i, val)
    if i > 4: break
# 0 a
# 1 b
# 2 c
# 3 a
# 4 b
# 5 c

stackoverflow: How to make a repeating generator in Python

json

インスタンスをjsonで保存する

json.dump(f.__dict__)でOK．
例えば以下の抽象クラスは，インスタンスの情報をjsonファイルで保存・そこから読み込みをするテンプレート．

class AbstractConfig(metaclass=ABCMeta):

    def to_json(self, path):
        '''
        Args:
            path (str)
        '''
        with open(path, 'w') as f:
            json.dump(self.__dict__, f, indent=4)

    @classmethod
    def from_json(cls, path):
        '''
        Args:
            path (str)
        '''
        with open(path, 'r') as f:
            json_dict = json.load(f)
        return cls(**json_dict)

• .__dict__では不十分なケースがあるらしいが，それはその時
• (jsonではなく) jsonsというライブラリもあるらしい

stackoverflow: How to make a class JSON serializable
verilog書く人: 【Python】jsonで自作クラスを含んだデータをシリアライズ/デシリアライズする

JSONDecodeError: Expecting property name enclosed in double quotesになる他の原因

エラーメッセージは「"で囲ってほしい」と言っているが，
そうであってもこのエラーが発生することがある．
}でくくる直前に,がついている場合がその一例．
以下は正常に動作する例．

test.json

{
    "a": "one",
    "b": "two"
}

path = './test.json'
with open(path, 'r') as f:
    adict = json.load(f)
print(adict) # {'a': 'one', 'b': 'two'}

しかし，以下の通り(Pythonでは許される通り)}で閉じる直前に,を付けると上記のエラーが発生する．
(line 4 column 1というので，}でエラーが発生したことになる．)

error.json

{
    "a": "one",
    "b": "two", <- この,がNG
}

mutagen

mp3などのタグ編集

note.nkmk.me: Pythonでmp3などのID3タグを編集するmutagenの使い方

netCDF4

日時の変換

'''nc (netCDF4.Dataset), time_name (str)として'''
from netCDF4 import num2date
time = nc.variables[time_name]
num2date(time[:], units=time.units, calendar=time.calendar)

ただし，これで返ってくるのはcftime.datetimeというクラスのサブクラス．
これをdatetime.datetimeに変換するためにはyearやmonthといった同名の属性にアクセスする．

import datetime as dt
def cftime2datetime(cftime):
    keys = ['year', 'month', 'day', 'hour', 'minute', 'second', 'microsecond']
    dict_kw = {key: getattr(cftime, key) for key in keys}
    return dt.datetime(**dict_kw)

num2dateの使い方は下ページの"7) Dealing with time coordinates."節が分かりやすい．
公式: netCDF4 module
cftime.datetimeの詳細は以下を参照．
公式: cftime API

numpy/scipy

pandas

pathlib

from pathlib import Path
p = Path( 'dir1/dir2/file.txt' )

print( type( p.name ), p.name ) # <class 'str'> file.txt
print( type( p.stem ), p.stem ) # <class 'str'> file
print( type( p.suffix ), p.suffix ) # <class 'str'> .txt
print( type( p.parent ), p.parent ) # <class 'pathlib.WindowsPath'> dir1\dir2
print( type( p.parents ), p.parents ) # <class 'pathlib._PathParents'> <WindowsPath.parents>
print( type( p.parents[0] ), p.parents[0] ) # <class 'pathlib.WindowsPath'> dir1\dir2
print( type( p.parents[1] ), p.parents[1] ) # <class 'pathlib.WindowsPath'> dir1

note.nkmk.me: Python, pathlibでファイル名・拡張子・親ディレクトリを取得
note.nkmk.me: Python, pathlibでファイル一覧を取得（glob, iterdir）

parse

文字列のパース

PyPI: parse
trivial technologies: Pythonicな文字列パーサ parse

pickle

きっと続かんブログ: 【python】リストや辞書を外部ファイルに保存

ジェネレータはpicklableではない

def is_picklable(obj):
    try:
        pickle.dumps(obj)
    except:
        return False
    return True

obj = range(10)
print(type(obj), is_picklable(obj)) # <class 'range'> True
obj = (i for i in range(10))
print(type(obj), is_picklable(obj)) # <class 'generator'> False
obj = iter(range(10))
print(type(obj), is_picklable(obj)) # <class 'range_iterator'> True
obj = iter([1, 2, 3])
print(type(obj), is_picklable(obj)) # <class 'list_iterator'> True

保存するにはリスト等に変換すれば問題ないが，
ProcessPoolExecutorによる並列時に注意．

stackoverflow: Why can't generators be pickled?

re

note.nkmk.me: Pythonの正規表現モジュールreの使い方（match、search、subなど）
note.nkmk.me: Pythonで文字列を抽出（位置・文字数、正規表現）
Qiita@luohao0404: 分かりやすいpythonの正規表現の例

schedule

Qiita@Kai-Suzuki: Python Scheduleライブラリでジョブ実行

sqlite3

Tensorflow

バージョン別のドキュメント (.md内のハイパーリンクは機能しないかも)
Tensorflow: TensorFlowAPIバージョン

(株)クラスキャット セールスインフォメーション: TensorFlow : Get Started : TensorFlow 技法 101

https://vict0rs.ch/2018/05/17/restore-tf-model-dataset/
https://screwandsilver.com/tensorflow_init_variable_error/
https://stackoverflow.com/questions/42322698/tensorflow-keras-multi-threaded-model-fitting

tqdm

基本: イテレータをtqdm.tqdm()で囲む

def myfunc(i):
    time.sleep(0.0001)
    return i

n = int(1e4)
args = range(n)
for i in tqdm.tqdm(args):
    myfunc(i)

Qiita@kuroitu: Pythonで進捗表示したい！

ジェネレータ等lenが無い場合: total=オプションで全体数を指定

- `total=len(list(ジェネレータ))`とするとforが回らなくなるので注意

args = range(n)
args = (i for i in args)
for i in tqdm.tqdm(args, total=n):
    myfunc(i)

Pystyle: tqdm でプログラムの進捗をプログレスバーで表示する方法 > 全体のイテレーション数を指定する

print()との併用: print() -> tqdm.tqdm.write()に置換

陽に文字列に変換する必要あり

def myfunc(i):
    time.sleep(0.0001)
    if i % 100 == 0: tqdm.tqdm.write(str(i))
    return i

ばいばいバイオ: 【Python】 tqdmでプログレスバーを表示してみた

並列計算との複合

tqdm.contrib
rangeやenumerate, Process/ThreadPoolExecutor.mapはtqdmからラッパーが提供されている

from tqdm.contrib.concurrent import process_map#, thread_map
r = process_map(myfunc, args, max_workers=16, chunksize=100)

submitが使いたい場合:

with tqdm.tqdm(total=n) as pbar:
    fs = []
    with futures.ProcessPoolExecutor(max_workers=16) as executor:
        for i in args:
            f = executor.submit(myfunc, i)
            fs += [f]
        for f in futures.as_completed(fs):
            pbar.update(1)

以下はstackoverflowでも指摘されている悪い例．
(最初のタスクが遅いと，プログレスバーが一気に進む)

def myfunc(i):
    time.sleep(10/i)
    #time.sleep(0.0001)
    return i

def main(*args):
    n = int(1e4)
    args = range(1, n+1)

    with futures.ThreadPoolExecutor(max_workers=16) as executor:
        fs = list(tqdm.tqdm(executor.map(myfunc, args), total=n))

stackoverflow: Use tqdm with concurrent.futures?

treelib: 木構造

Welcome to treelib’s documentation!
がドキュメントだが使うには不十分なので，
以下のGitHubのソースコードを見ながら作業する方が無難…
GitHub: caesar0301/treelib

unittest

テンプレート

import unittest

class MyTestCase(unittest.TestCase):

    def test(self):
        # 返り値のテスト
        self.assertEqual(myfunc(args), expected_return)

        # 返り値がTrue/Falseか確認
        self.assertTrue(return_true())
        self.assertTrue(return_false())

        # 返り値がNoneであるかテスト
        self.assertIsNone(return_none(args))

        # 例外のテスト
        with self.assertRaises(ExpectedException):
            myfunc(args)

if __name__ == '__main__':
    # スクリプトとして実行された場合の処理
    unittest.main(verbosity=2)

例外は以下でもテスト可能だが，引数は別に渡す必要がある．
(これ見ると，期待される返り値 -> 関数 -> (引数)の順に書く方が統一感あるような気がするが，
以下のdocs.python.orgでもそうではない: self.assertEqual('foo'.upper(), 'FOO'))

self.assertRaises(ExpectedException, myfunc)
self.assertRaises(ExpectedException, myfunc, arg1, arg2) # 引数があるときはそのまま列挙

docs.python.org: unittest
CUBE SUGAR CONTAINER: Python: ユニットテストを書いてみよう
stackoverflow: How do you test that a Python function throws an exception?

ディレクトリ構成とテスト実行

Qiita@hoto17296: Python 3 標準の unittest でテストを書く際のディレクトリ構成

urllib

urlopenのエラー確認

try:
    res = urllib.request.urlopen(url)
except urllib.error.HTTPError as e:
    print(e.code, e.reason) # ここでエラー取得

Hibiki Programming Notes: [Python] Webサイトのデータを取得する（urllib.requestモジュール）

403: Forbiddenなサイトにアクセス

hdrs = {"User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:47.0) Gecko/20100101 Firefox/47.0"}
req = urllib.request.Request(url=url, headers=hdrs)
res = urllib.request.urlopen(req)

stackoverflow: Python3でwebスクレイピングしたいのですが存在するURLが開けません。