はじめに

私が愛するスラッシュメタル（Thrash Metal）、その中でもSLAYERは一番のお気に入りだ。

長年活躍してきた彼らだが、メンバーの死を乗り越えながらも、ついにファイナルワールドツアーを迎えた。
そして2019年11月30日、LAでの最終講演をもって有終の美を飾った。

https://www.youtube.com/watch?v=OwsdbuxRc_s

これを記念して、彼らが伝えたかったことを確認しておきたいと思う。

環境

• Windows10
• Python 3.7.5

可視化結果

もはや何も言うことはない。

俺は満足だ！！素晴らしい音楽とメッセージをありがとう！！！！！

ソースコード

import requests
from bs4 import BeautifulSoup
import pandas as pd
import time

#アルバムごとのURLリスト
urls = ['http://www.darklyrics.com/lyrics/slayer/shownomercy.html',
        'http://www.darklyrics.com/lyrics/slayer/hauntingthechapel.html',
        'http://www.darklyrics.com/lyrics/slayer/hellawaits.html',
        'http://www.darklyrics.com/lyrics/slayer/reigninblood.html',
        'http://www.darklyrics.com/lyrics/slayer/southofheaven.html',
        'http://www.darklyrics.com/lyrics/slayer/seasonsintheabyss.html',
        'http://www.darklyrics.com/lyrics/slayer/divineintervention.html',
        'http://www.darklyrics.com/lyrics/slayer/undisputedattitude.html',
        'http://www.darklyrics.com/lyrics/slayer/diabolusinmusica.html',
        'http://www.darklyrics.com/lyrics/slayer/godhatesusall.html',
        'http://www.darklyrics.com/lyrics/slayer/christillusion.html',
        'http://www.darklyrics.com/lyrics/slayer/worldpaintedblood.html',
        'http://www.darklyrics.com/lyrics/slayer/repentless.html']

#リリックのデータフレーム準備
list_df = pd.DataFrame(columns=['lyrics'])

for url in urls:
    #取得
    response = requests.get(url)
    soup = BeautifulSoup(response.text, 'lxml')
    song_lyrics = soup.find('div', class_='lyrics')
    song_lyric = song_lyrics.text
    song_lyric = song_lyric.replace('\n','')

    #1秒待機（サーバー負荷の考慮）
    time.sleep(1)

    #取得した歌詞を表に追加
    tmp_se = pd.DataFrame([song_lyric], index=list_df.columns).T
    list_df = list_df.append(tmp_se)

print(list_df)

#csv保存
list_df.to_csv('list.csv', mode = 'a', encoding='utf-8')

from janome.tokenizer import Tokenizer
import pandas as pd
import re

#list.csvファイルを読み込み
df_file = pd.read_csv('list.csv')

song_lyrics = df_file['lyrics'].tolist()

t = Tokenizer()

results = []

for s in song_lyrics:
    tokens = t.tokenize(s)

    r = []

    for tok in tokens:
        if tok.base_form == '*':
            word = tok.surface
        else:
            word = tok.base_form

        ps = tok.part_of_speech

        hinshi = ps.split(',')[0]

        if hinshi in ['名詞', '形容詞', '動詞', '副詞']:
            r.append(word)

    rl = (' '.join(r)).strip()
    results.append(rl)
    #余計な文字コードの置き換え
    result = [i.replace('\u3000','') for i in results]
    print(result)

text_file = 'wakati_list.txt'
with open(text_file, 'w', encoding='utf-8') as fp:
    fp.write("\n".join(result))

from wordcloud import WordCloud

text_file = open('wakati_list.txt', encoding='utf-8')
text = text_file.read()
print(text)

#無意味そうな単語除去
stop_words = ['the', 'of', 'to', 'is', 'in', 'for', 'with', 'that', 'my', 'all', 'will', 'from', 'can', 'your',  
              'on', 'me', 'it', 'and', 'this', 'be', 'are', '=', 're', 'll', 'am', 'their', 'lyrics', 
             'king', 'hanneman', 'chorus', 'music', 'Lead', 'bonus', 'Araya', 'thanks', 'darklyrics', 'track', 'do', 'there',
              'submits', 'comments', 'Richard Hung Himself', 'at', 'webmaster', 'corrections', 'welcomed', 'or', 'its', 'don',
             'correcting', 'before', 'without', 'an', 'they', 'one', 'you', 'com', 'slayer']

wordcloud = WordCloud(background_color='black', colormap='autumn',
    font_path=fpath, width=800, height=600, stopwords=set(stop_words)).generate(text)

#画像はwordcloud.pyファイルと同じディレクトリにpng保存
wordcloud.to_file('./wordcloud.png')

※参考ページから雑に拝借しているので、そのうち見直します・・・。（とにかく作りたかった・・・）

おわりに

1983年「Show No Mercy」から、1986年の伝説的アルバム「Reign in Blood」。
そして、最後のオリジナルアルバムとなる2015年「Repentless」。

生涯をかけて、これだけ音楽性を貫き通したバンドは他にいないだろう。
彼らはとても希有な存在だった。

彼らの音を、そして鋼鉄魂を、俺は一生この胸に抱き続ける。

あ・・・ここブログじゃなかった。

参考

https://qiita.com/yuuuusuke1997/items/122ca7597c909e73aad5
https://qiita.com/itkr/items/513318a9b5b92bd56185
https://qiita.com/TkrUdagawa/items/aa483630b5ec7d337c9e
https://amueller.github.io/word_cloud/generated/wordcloud.WordCloud.html#wordcloud.WordCloud

はじめに

これは、前回(optunaに数独を解かせてみた)の続編です。

前回は数独のルールすら知らない状態で、optunaを使って最適化することで正解にたどり着けるか試してみました。
今回は、縦・横・3x3のブロックで数字の重複がないというルールを知ったうえでoptunaに最適化させてみました。

もっと早くに投稿するつもりだったのですが、前回の投稿後に始めたリングフィット アドベンチャーがハードすぎて気力・体力がともになくなって、遅くなってしまいました。

方法

今回は、焼き鈍し法で解きます。

焼き鈍し法は、現在の値を少しランダムに変化させて次の値を作ります。(初回は元の値がないので、完全にランダムに作ります。)
作った新しい値から計算されるコストが減少するなら新しい値に移動します。
減少しない場合でも確率で移動したり、しなかったりします。
徐々にその確率を小さくしていくことで、局所解に収束します。

実装

optunaで焼き鈍しをする方法は公式ドキュメントにあります。
新しい値をサンプリングする部分を変更しただけで、後は公式ドキュメントのとおりです。
サンプラーを変更した以外は前回のコードのままです。

ここでは、新しい値をサンプリングする部分のみを示します。
コード全体は最後に示します。

params = {}
# サンプリングの順序をランダムにする
name_list = shuffle(list(search_space))
for param_name in name_list:
    # 2次元の座標(i, j)
    i, j = int(param_name[1]), int(param_name[2])
    # 予め(i, j)がある縦・横・3x3のブロックの要素を取り出すマスクを作っておいた
    # ただし、(i, j)の要素はマスクでは取り出さない
    mask = self._mask_array[i, j]
    tmp = self._state * mask
    # 1~9の各要素がそれぞれいくつあるかを数える
    cost = np.asarray([np.count_nonzero(tmp == v) for v in range(1, 10)])
    probability = softmax(-cost * 5)

    # 新しい値をサンプリング   
    new_value = np.random.choice(9, p=probability) + 1
    # 新しい値を記録
    params[param_name] = new_value
    self._state[i, j] = new_value

return params

新しい値は、縦・横・3x3のブロックで数字の重複がないというルールに基づいてサンプリングされる確率が高くなるようにしています。

ここでルールについての知識を利用しています。

実験結果

左に*がある数字は元からヒントとして与えられているものです。

*5*3 2 4*7 6 9 1 8
*6 7 4*1*9*5 3 5 2
 1*9*8 2 3 8 4*6 7
*8 1 9 5*6 4 7 2*3
*4 2 6*8 7*3 5 9*1
*7 5 3 9*2 1 8 4*6
 9*6 1 3 5 7*2*8 4
 2 8 7*4*1*9 6 3*5
 3 4 5 6*8 2 1*7*9
loss: 4.0

惜しい。

トライアル数が115回目にこの答えに辿り着きました。
この後も、200回まで行いましたがダメでした。

一目ではどこが間違っているかわからないですが、中央で縦方向に7が重複しています。
中央の7が縦方向では間違っていますが、横方向では条件を満たしています。
そのため、複数の要素が同時に変わらないと正解に辿り着けないので、かなり深い局所解になっています。(コストがかなり高い状態を経由しないと正解に辿り着かない)

まとめ

局所解になってしまいました。
正解にたどり着くためには、複数の要素が同時に変わらないといけないので、現在の方法では難しいようです。
初期値を変えて何度もやり直せば、正解できるはずです。

今後の予定

今回は中央で縦方向に7が重複していましたが、7は元から与えられているものがあるので、サンプリングされるべきでないはずです。
(ルールに関する知識をどこまで利用するかという条件に関係してくる)
また、右から3列目、上から3行目の様に元から与えられている数字から値が確定するものもあります。

これらが今回の方法では全く考慮されていなかったので、次回はこれを解決したいと思います。
(もうoputunaは関係ない気がする)

コード

from itertools import product

import numpy as np
import optuna
import click
from scipy.special import softmax
from sklearn.utils import shuffle


# https://optuna.readthedocs.io/en/stable/tutorial/sampler.html
class SimulatedAnnealingSampler(optuna.samplers.BaseSampler):
    def __init__(self, temperature=100):
        self._rng = np.random.RandomState()
        self._temperature = temperature
        self._current_trial = None

        self._state = None
        self._mask_array = make_mask_array()

    def sample_relative(self, study, trial, search_space):
        if search_space == {}:
            return {}

        # 現在のtrialはstudy.trials[-1]
        previous_trial = study.trials[-2]
        if self._current_trial is None or previous_trial.value <= self._current_trial.value:
            probability = 1.0
        else:
            probability = np.exp((self._current_trial.value - previous_trial.value) / self._temperature)
        self._temperature *= 0.99

        if self._rng.uniform(0, 1) < probability:
            self._current_trial = previous_trial

        if self._state is None:
            self._state = np.empty([9, 9], dtype=np.int32)
            for i, j in product(range(9), repeat=2):
                name = 'p{}{}'.format(i, j)
                if name in preset:
                    self._state[i, j] = preset[name]

        for i, j in product(range(9), repeat=2):
            name = 'p{}{}'.format(i, j)
            if name in self._current_trial.params:
                self._state[i, j] = self._current_trial.params[name]

        params = {}
        name_list = shuffle(list(search_space))
        for param_name in name_list:
            i, j = int(param_name[1]), int(param_name[2])
            mask = self._mask_array[i, j]
            tmp = self._state * mask
            cost = np.asarray([np.count_nonzero(tmp == v) for v in range(1, 10)])
            probability = softmax(-cost * 5)

            new_value = np.random.choice(9, p=probability) + 1
            params[param_name] = new_value
            self._state[i, j] = new_value

        return params

    def infer_relative_search_space(self, study, trial):
        return optuna.samplers.intersection_search_space(study)

    def sample_independent(self, study, trial, param_name, param_distribution):
        independent_sampler = optuna.samplers.RandomSampler()
        return independent_sampler.sample_independent(study, trial, param_name, param_distribution)


def make_mask_array():
    mask_array = np.zeros([9, 9, 9, 9])
    for i, j in product(range(9), repeat=2):
        mask = mask_array[i, j]

        mask[i] = 1
        mask[:, j] = 1
        s, t = i // 3 * 3, j // 3 * 3
        mask[s:s + 3, t:t + 3] = 1
        # 自分自身を取り除く
        mask[i, j] = 0
    return mask_array


"""
 -----------------        
|5|3| | |7| | | | |
|-+-+-+-+-+-+-+-+-|
|6| | |1|9|5| | | |
|-+-+-+-+-+-+-+-+-|
| |9|8| | | | |6| |
|-+-+-+-+-+-+-+-+-|
|8| | | |6| | | |3|
|-+-+-+-+-+-+-+-+-|
|4| | |8| |3| | |1|
|-+-+-+-+-+-+-+-+-|
|7| | | |2| | | |6|
|-+-+-+-+-+-+-+-+-|
| |6| | | | |2|8| |
|-+-+-+-+-+-+-+-+-|
| | | |4|1|9| | |5|
|-+-+-+-+-+-+-+-+-|
| | | | |8| | |7|9|
 -----------------        
"""
preset = {'p00': 5, 'p01': 3, 'p04': 7,
          'p10': 6, 'p13': 1, 'p14': 9, 'p15': 5,
          'p21': 9, 'p22': 8, 'p27': 6,
          'p30': 8, 'p34': 6, 'p38': 3,
          'p40': 4, 'p43': 8, 'p45': 3, 'p48': 1,
          'p50': 7, 'p54': 2, 'p58': 6,
          'p61': 6, 'p66': 2, 'p67': 8,
          'p73': 4, 'p74': 1, 'p75': 9, 'p78': 5,
          'p84': 8, 'p87': 7, 'p88': 9}


def evaluate(answer):
    tmp = np.reshape(answer, [3, 3, 3, 3])
    loss = np.sum((
        np.sum([np.count_nonzero(np.logical_not(np.any(answer == i, axis=0))) for i in range(1, 10)]),
        np.sum([np.count_nonzero(np.logical_not(np.any(answer == i, axis=1))) for i in range(1, 10)]),
        np.sum([np.count_nonzero(np.logical_not(np.any(tmp == i, axis=(1, 3)))) for i in range(1, 10)]),
    ))
    return loss


def objective(trial):
    candidate = (1, 2, 3, 4, 5, 6, 7, 8, 9)

    answer = np.empty([9, 9], dtype=np.uint8)
    for i, j in product(range(9), repeat=2):
        key = 'p{}{}'.format(i, j)
        if key in preset:
            answer[i, j] = preset[key]
        else:
            answer[i, j] = trial.suggest_categorical(key, candidate)

    return evaluate(answer)


def run(n_trials):
    study_name = 'sudoku'
    sampler = SimulatedAnnealingSampler()
    study = optuna.create_study(study_name=study_name, storage='sqlite:///sudoku.db', load_if_exists=True,
                                 sampler=sampler)
    study.optimize(objective, n_trials=n_trials)

    show_result(study.best_params, study.best_value)

    df = study.trials_dataframe()
    df.to_csv('tpe_result.csv')


def show_result(best_params, best_value):
    for i in range(9):
        for j in range(9):
            key = 'p{}{}'.format(i, j)
            if key in preset:
                print('*{:1d}'.format(preset[key]), end='')
            else:
                print('{:2d}'.format(best_params[key]), end='')
        print('')
    print('loss: {}'.format(best_value))


@click.command()
@click.option('--n-trials', type=int, default=1000)
def cmd(n_trials):
    run(n_trials)


def main():
    cmd()


if __name__ == '__main__':
    main()

TL;DR

• dataclassすごくいいよ
• 手書きで書いたclassと遜色ないよ
• これからdataclassをベースにしたライブラリが出てきそう

dataclassとは?

dataclassはpython 3.7で追加された新しい標準ライブラリ。簡単に説明するとclassに宣言に@dataclassデコレータを付けると、__init__, __repr__, __eq__, __hash__といった所謂dunder(double underscoreの略。日本語だとダンダーと読むのかな)メソッドを生成してくるライブラリ。これを使うと面倒なクラスの定義を大幅に短縮できたり、下手な実装より高速だったりする。ここで紹介した以外にもdataclassには色々な機能があるので、詳しくは公式ドキュメントやPython3.7からは「Data Classes」がクラス定義のスタンダードになるかもしれないを参照してほしい。

python3.7がまだ使えないよ、という人にもPyPIに3.6用のbackportが用意されています。

dataclassの使い方

from dataclasses import dataclass, field
from typing import ClassVar, List, Dict, Tuple
import copy

@dataclass
class Foo:
    i: int
    s: str
    f: float
    t: Tuple[int, str, float, bool]
    d: Dict[int, str]
    b: bool = False  # デフォルト値
    l: List[str] = field(default_factory=list)  # listのデフォルトを[]にする
    c: ClassVar[int] = 10  # クラス変数

# 生成された`__init__`でインスタンス化
f = Foo(i=10, s='hoge', f=100.0, b=True,
        l=['a', 'b', 'c'], d={'a': 10, 'b': 20},
        t=(10, 'hoge', 100.0, False))

# 生成された`__repr__`でhの文字列表現をプリントアウトする
print(f)

# コピーを作って書き換えてみる
ff = copy.deepcopy(f)
ff.l.append('d')

# 生成された`__eq__`で比較する
assert f != ff

パフォーマンス

dataclassを使って作ったDataclassFooと手書きで書いたManualFooの__init__, __repr__, __eq__の実行時間を計測してみた。

• macOS 10.14 Mojave
• Intel 2.3GHz 8-core Intel Core i9
• DDR4 32GB RAM
• Python 3.6.3

計測に使ったソースコード

import timeit
from dataclasses import dataclass

@dataclass
class DataclassFoo:
    i: int
    s: str
    f: float
    b: bool

class ManualFoo:
    def __init__(self, i, s, f, b):
        self.i = i
        self.s = s
        self.f = f
        self.b = b
    def __repr__(self):
        return f'ManualFoo(i={self.i}, s={self.s}, f={self.f}, b={self.b})'
    def __eq__(self, b):
        a = self
        return a.i == b.i and a.s == b.s and a.f == b.f and a.b == b.b

def bench(name, f):
    times = timeit.repeat(f, number=100000, repeat=5)
    print(name + ':\t' +  f'{sum(t)/5:.5f}')

bench('dataclass __init__', lambda: DataclassFoo(10, 'foo', 100.0, True))
bench('manual class __init__', lambda: ManualFoo(10, 'foo', 100.0, True))

df = DataclassFoo(10, 'foo', 100.0, True)
mf = ManualFoo(10, 'foo', 100.0, True)
bench('dataclass __repr__', lambda: str(df))
bench('manual class __repr__', lambda: str(mf))

df2 = DataclassFoo(10, 'foo', 100.0, True)
mf2 = ManualFoo(10, 'foo', 100.0, True)
bench('dataclass __eq__', lambda: df == df2)
bench('manual class __eq__', lambda: mf == mf2)

各10万回を5セット実行した平均

	計測結果(sec)
dataclass __init__	0.04382
手書きclass __init__	0.04003
dataclass __repr__	0.07527
手書きclass __repr__	0.08414
dataclass __eq__	0.04755
手書きclass __eq__	0.04593

50万回実行してこれならほぼ差はないと言っていいでしょう。

また、バイトコードも一致した。

dataclassの__init__

>>> import dis
>>> dis.dis(DataclassFoo.__init__)
  2           0 LOAD_FAST                1 (i)
              2 LOAD_FAST                0 (self)
              4 STORE_ATTR               0 (i)

  3           6 LOAD_FAST                2 (s)
              8 LOAD_FAST                0 (self)
             10 STORE_ATTR               1 (s)

  4          12 LOAD_FAST                3 (f)
             14 LOAD_FAST                0 (self)
             16 STORE_ATTR               2 (f)

  5          18 LOAD_FAST                4 (b)
             20 LOAD_FAST                0 (self)
             22 STORE_ATTR               3 (b)
             24 LOAD_CONST               0 (None)
             26 RETURN_VALUE

手書きclassの__init__

>>> dis.dis(ManualFoo.__init__)
 13           0 LOAD_FAST                1 (i)
              2 LOAD_FAST                0 (self)
              4 STORE_ATTR               0 (i)

 14           6 LOAD_FAST                2 (s)
              8 LOAD_FAST                0 (self)
             10 STORE_ATTR               1 (s)

 15          12 LOAD_FAST                3 (f)
             14 LOAD_FAST                0 (self)
             16 STORE_ATTR               2 (f)

 16          18 LOAD_FAST                4 (b)
             20 LOAD_FAST                0 (self)
             22 STORE_ATTR               3 (b)
             24 LOAD_CONST               0 (None)
             26 RETURN_VALUE

dataclassの内部の解説に入る前に

dataclassを説明するにあたり重要なパーツを説明しておきたい。

PEP526: Syntax for Variable Annotations

PEP526は型宣言の方法を記述してあるんだけど、この仕様追加によってclassに宣言された変数の型情報をプログラム実行時に取得することが可能になった。

from typing import Dict
class Player:
    players: Dict[str, Player]
    __points: int

print(Player.__annotations__)
# {'players': typing.Dict[str, __main__.Player],
#  '_Player__points': <class 'int'>}

組み込みexec関数

evalは知ってる人が多いと思う。ざっくりevalとの違いをいうと、

eval: 引数の文字列を式として評価する
exec: 引数の文字列を文として評価する

これだけじゃ意味不明なので次の例を見てみよう。

これを実行すると"typing rocks!"を出力されるのは簡単に想像できる。

>>> exec('print("typing rocks!")')
"typing rocks!"

ではこれは？

exec('''
def func():
    print("typing rocks!")
''')

次にこれを実行してみる

>>> func()
"typing rocks!"

そう。実はexecは文字列を式として評価するので、pythonの関数でさえも動的に定義することができる。すげぇ。

で、dataclassは内部で何を行っているのか？

dataclassデコレータをつけたclassがimportされると、上で説明したtype annotationsやexecを使ってコード生成を行っている。超ざっくりだが、以下のような流れになる。詳しく知りたい人はcpythonのソースのこの辺を読んでみよう。

1. dataclassデコレータがクラスに対して呼ばれる
2. 各フィールドの型情報（型名、型クラス、デフォルト値等）をtype annotationsから取得する
3. 型情報を使って__init__ 関数定義の文字列を作る
4. 文字列をexecに渡して動的に関数を生成する
5. クラスに__init__関数をセットする

3, 4, 5を単純化したコードはこんな感じ。

nl = '\n'  # f-string内でエスケープ使えないので外で定義する

# 関数定義の文字列作成
s = f"""
def func(self, {', '.join([f.name for f in fields(Hoge)])}):
{nl.join('  self.'+f.name+'='+f.name for f in fields(Hoge))}
"""

# 関数定義の文字列をコンソール出力してみる
print(s)
# def func(self, i, s, f, t, d, b, l):
#   self.i=i
#   self.s=s
#   self.f=f
#   self.t=t
#   self.d=d
#   self.b=b
#   self.l=l

# execでコード生成。`func`関数がスコープ内に定義された
exec(s)

setattr(Foo, 'func', func)  # クラスに生成した関数をクラスにセットする

以上は単純化された例だけど、実際には

• フィールドに設定されたデフォルト値
• List等に使うデフォルトファクトリ関数
• クラス変数(ClassVar)
• プログラマが定義済みだったら生成しない
• 他のdunder関数の生成
• dataclassのclassの継承

等を全て考慮して、どんな場合も正しく動作するように丁寧に丁寧に関数定義文字列作成、コード生成が行われているのです。

さらに、もう一つ押さえておきたいことが、このコード生成が行われるのはモジュールがロードされた瞬間のみということ。一度classがimportされたら、手書きで書いたclassと何の変わりもなく使えるということだ。

Rustの#[derive]

Rustにはstructを定義する時につけるDerive attribute(#[derive])というものがある。これ、dataclassとほぼ同等かそれ以上のことができる。例えば以下をみてもらえると、

#[derive(Debug, Clone, Eq, PartialEq, Hash)]
struct Foo {
    i: i32,
    s: String,
    b: bool,
}

#[derive(Debug, Clone, Eq, PartialEq, Hash)]をつけるだけで、これだけのメソッドを生成してくれる。

• Debug用文字列生成のメソッド生成 (Pythonでいう__repr__)
• オブジェクトをクローンするメソッド生成
• 比較メソッド生成(Pythonでいう__eq__や__gt__)
• ハッシャーメソッド生成(Pythonでいう__hash__)

またRustはさらにすごくて、自分のCustom driveを実装する機能が公式でサポートされていて、割とカジュアルに型ベースのメタプログラミングができる。

こういったプログラマを楽にする機能がRustには他にもたくさんあるので、型制約や所有権が難しくてもRustが生産性が高い理由だと筆者は思っている。Rustは本当に素晴らしい言語なのでPythonistaの方々もぜひぜひ触ってみてほしい。

メタプログラミングとしてのdataclassの可能性

dataclassは型ベースのメタプログラミングの有用性と可能性を示したいい例だと個人的には思っている。

筆者も二つほどdataclassをベースにしたライブラリを作ってみたので、興味がある人はみてみてほしい。

• envclasses
  • 環境変数の値をdataclassのフィールドにマッピングするライブラリ。コンテナ使ってて、Pythonのコンフィグclassを環境変数でオーバーライドしたい時とかに便利
• pyserde
  • dataclassベースのシリアライズライブラリ

RustでそうであるようにPythonでもこの分野が盛り上がって良いライブラリがたくさん出てきてほしい。

動機

（要約）ただ僕が知りたくなったからです。それ以上の理由はありません。

もっと詳しく

文字化けには何種類ものパターンがあります。
UTF8のファイルをSJISで表示した時に起こる文字化け、
UTF8をEUCで表示した時に起こる文字化け、
EUCをUTF8で表示した時に起こる文字化け……。
それぞれがどんな見た目になるかはこのページなどで確認できます。
正直最近はUTF8で作業をするため、文字化けはあまり見ていません。
けれども私が新卒で入った職場ではテキストファイルの作成はSJISが基本でした。そして、UTF8のファイルをSJISで開いて文字化けを起こすことが、わりと頻繁に起きていました。
UTF8→SJISへの文字化けだとよく縺ｫ蜈育函縺ｨ蜻ｼ繧薙〒縺�縺溘€ゅ□縺……のような文字化けが発生するのですが、これを見ていると、同じ漢字が何度も何度も登場していることが分かると思います。
文字化けで出てくる漢字には偏りがある。
そのよく出てくる漢字とその意味を、ただ知りたかった……。
でも意外とそういう内容の記事って見つからないし、しょうがないから自分で書くか……。
という、私の特に何の役にも立たない些細な好奇心がこの記事を書く原動力でした。

同じ漢字ばかり出てくる理由

同じ文字ばかり出てくる理由については、去年に素晴らしい記事が書かれていたため、ここでは割愛させて頂きます。
UTF-8からSJISに文字化けすると糸偏の漢字がよく出てくる

集計方法

適当な長文をもとに、元UTF8 → SJIS表示での文字化けを発生させてファイルに保存しておき、その漢字をPythonで数え上げます。

どんなに輝かしい名文でも、一旦SJISで表示させてからUTF8として保存し直せば、見るも無残な姿に早変わりできます。この世の無常を感じずにはいられません。

長文には何をチョイスするかですが、ここではおそらくみんなが知っているであろうものを題材にしてみます。
高校の教科書に載っている夏目漱石の『こころ』と、中学校の教科書に載っている太宰治の『走れメロス』でそれぞれカウントしてみましょう。メインは『こころ』の方で、短編の『走れメロス』のランキングはオマケ程度に紹介します。

文章は青空文庫に載っているため、そこからコピペしてデータを作らせて頂きました。

夏目漱石『こころ』
太宰治『走れメロス』

プログラムは、特に何の面白みもないpythonのコードを書きました。

import re


with open('./source.txt', encoding="utf-8") as f:
    s: str = f.read()

    # 出てくる文字を数え上げて、その結果を辞書にする
    # このとき、漢字以外の文字を除外する。つまり「ｮ」とか「ｹ」とかをはじく。
    count_dic = {}
    for char in s:
        result = re.search('[一-鿐]', char)
        if result is None:
            # 漢字じゃないので飛ばす
            continue
        if char in count_dic:
            count_dic[char] += 1
        else:
            count_dic[char] = 1

    # 少ない順に出力
    for k, v in sorted(count_dic.items(), key=lambda x: x[1]):
        print(str(k) + ": " + str(v))

結果発表『こころ』部門

※漢検の級判定はgoo辞書で出てきたものをそのまま使っています。

第20位

出現数1049回
$\huge{遘}$
漢検1級

トップバッターからいきなり心を折ってくる難しい漢字が出ました。『こころ』のランキングの自覚はあるのでしょうか。
あう。であう。めぐりあう、という漢字です。
めぐりあうは「めぐり遘う」とも書くことができ、邂逅は邂遘とも書けるようです。
現代でこんな字を書いたら多分嫌われてしまいますね。

第19位

出現数1112回
$\huge{代}$
漢検8級

「田代まさし」の「代」。覚醒剤はだめだぞ。

第18位

出現数1190回
$\huge{荳}$
漢検1級

植物のマメ、を表す字です。確かに豆が難しくなったような見た目です。
荳角皇女(ささげのおうじょ)という継体天皇の皇女様が6世紀頃にいらっしゃったようです。
それと関係あるかは不明ですが、大角豆（ささげ）という豆は日本で古くから食べられているササゲ属の一年草です。お祝い用の赤飯には、ササゲが使われます。時は江戸時代、小豆（あずき）は煮ると皮が破れやすく、「腹切れする豆は切腹に通じる」として、武士の間で嫌われていました。そのため、赤飯にはササゲを使ったんですって。……これはプログラミングの記事です。大丈夫です。上にPythonのコードをちゃんと書いておきました。なので消されないはずです。

第17位

出現数1201回
$\huge{莠}$
漢検1級

訓読みで「はぐさ」と読みます。稲に似ているけれども、葉ばかり伸びて実らない雑草のことだそうです｡例えば、猫じゃらしとして有名なエノコログサがこれに当たります。稲っぽいけど、実らない。転じて、悪いものの例えとしても使われるようです。なので莠言（ゆうげん）というのは、有害で醜悪な言葉を指します。まぁ使いませんよねこんな熟語……。

第16位

出現数1401回
$\huge{昴}$
漢検1級

紛らわしいですが「激昂」の「昂」ではありません。「昴」は「すばる」と読みます。星です。
平安時代の清少納言は、「星はすばる。ひこぼし。ゆふづつ。よばひ星、すこしをかし」とすばるを讃えました。
風の中のすばる　砂の中の銀河　みんな何処へ行った　見送られることもなく……

第15位

出現数1493回
$\huge{峨}$
漢検準1級

文系で日本史選択だった場合は、後嵯峨天皇（ごさがてんのう）という文字を見たことがあるはずです。そうでなくても、この漢字は人名で目にするかもしれません。峨という字は、山が高くけわしいさまを表します。

第14位

出現数1512回
$\huge{翫}$
漢検準1級

翫ぶ（もてあそぶ）、翫る（むさぼる）、翫る（あなどる）と読めます。
芝翫縞（しかんじま）というのは、江戸時代に流行した着物の柄です。四本の縦縞と鐶（かん。金属の輪）をつないだ形を合わせた柄のことで、「四鐶縞」と書くのが意味的には正しかったのですが、歌舞伎役者の3代目中村歌右衛門(うたえもん)の俳号「芝翫」にちなんでこの漢字があてられたようです。

第13位

出現数1553回
$\huge{医}$
漢検8級

お客様の中にお医者様はいませんか！？

第12位

出現数1555回
$\huge{上}$
漢検10級

領収書の宛名、上様でいいよって会社ってあるんでしょうか……？

第11位

出現数1625回
$\huge{輔}$
漢検準1級

人名でよく出てくるけど意外と漢検準1級扱い。
漢検準1級の読み問題で出てくる「輔弼」は、「ほひつ」と読みます。
輔弼は天皇の行為としてなされるべき、あるいは、なされざるべきことについて進言することです。

第10位

出現数1794回
$\huge{阪}$
漢検2級

サカには「坂」と「阪」の2種類があります。理由は諸説あるようですが、文化5年(1808)刊行の『摂陽落穂集』によると、坂は分解すると「土」と「反」に分けられ「土に返る」と読めることから、縁起がよくないと忌み嫌って「阪」を用いる人がいたとされています。せやから大坂って書いたらアカンで。知らんけど。

第9位

出現数2215回
$\huge{吶}$
漢検1級

たまに小説で出てくる漢字です。吶（ども）る、と書いたりします。
彼は訥々（とつとつ）と語りだした――。訥々と語るというのは、口ごもりながら語るということです。
飾りけがなく無口なことを朴訥（ぼくとつ）と言います。
言偏の「訥」と口偏の「吶」がありますが、同じ意味のようです。

第8位

出現数2282回
$\huge{薙}$
漢検準1級

漢検準1級扱いですが、説明は不要ですね。
三種の神器「八咫鏡」「天叢雲剣（別名：草薙剣）」「八尺瓊勾玉」はオタクの義務教育です。

第7位

出現数3147回
$\huge{後}$
漢検9級

ここから後ろは出現数が跳ね上がっていきます。

第6位

出現数4078回
$\huge{溘}$
漢検1級

「溘」という字はたちまち、にわかにという意味です。
「溘焉（こうえん）として逝く」というのは、雑に言うと「突然の死！！！」です。

第5位

出現数4718回
$\huge{励}$
漢検3級

Qiitaのいいねって励みになりますよね。押してください。

第4位

出現数5831回
$\huge{九}$
漢検10級

九なのに4位。
次からは文字化けで本当によく目にする糸偏三銃士です。

第3位

出現数6656回
$\huge{縲}$
漢検1級

螺旋丸！！！　ではありません。虫偏ではなく糸偏です。
「縲」は罪人をしばる縄を表す、かなりニッチな漢字です。
「縲絏（るいせつ）の辱を受けんより、寧ろ只今潔く自殺せん」〈竜渓・経国美談〉とあるように、主に「絏（せつ）」とセットで使われるようです。絏は、しばるという意味があるようです。

昭和よりも前の時代には金属の手錠なんてものはありませんから、縄で罪人を縛るわけです。江戸時代には捕手術（とりてじゅつ。素手で敵を殺さずに捕り抑える武術）の一環として捕縄術（ほじょうじゅつ/とりなわじゅつ）が盛んに用いられていました。取り押さえた敵を素早く拘束する『早縄』、形式・儀式的に用いる『本縄』、緊縛による拷問を加えるための『拷問縄』などがあり、縛る相手の身分や職業、性別、用途によってそれぞれ異なる縛り方が用意されていたらしいです。捕縄術はれっきとした武術であり、江戸時代には150もの流派があったとか……。すごい。

第2位

出現数12928回
$\huge{繧}$
漢検1級

出現回数を3位から大きく引き離して倍近くを稼ぎ出しました。
エンジニアでなくとも文字化けで何度も何度も目にしたであろうこの憎き漢字……。

繧繝（うんげん。暈繝）という言葉があります。
繧繝というのは簡単にいうと、古のグラデーションです。中国西域から伝わり、奈良・平安時代の仏画、寺院の装飾や染織などに用いられました。
そして繧繝彩色（うんげんさいしき）という色彩用語があります。具体的な絵を見た方が分かると思います。色彩検定などで出てくるらしいので、もしかしたらWebデザイナーの方の中には知っている方もいるかもしれません。参考：暈繝彩色の意味とは
正倉院の宝物、漆金薄絵盤（うるしきんぱくえのばん）というものがあるのですが、ここにわかりやすい繧繝彩色があります。

（出典：宮内庁のHP http://shosoin.kunaicho.go.jp/ja-JP/Treasure?id=0000014245 ）
よく見ると、ぼかして色を重ねるのではなく、層をなすように同系統の色が重なっているでしょう？　これが繧繝彩色です。
身近な例でいうと、Vueのロゴも繧繝彩色であると言えるかもしれませんね。

第1位

出現数60693回
$\huge{縺}$
漢検1級

6万回という圧倒的な出現数で他の追随を許さないスコアを稼ぎ出しました。
UTF8→SJISの文字化け界の王者に輝いたのは、糸偏三銃士最強の漢字「縺」です！

よく聞くのは「痴情の縺（もつ）れ」というように、「もつれ」を表す漢字です。
もつれは、絡み合うことです。「糸の縺れを解く」というふうに言います。
紛らわしいことに、「ほつれ」と「もつれ」の2つの言葉があります。
「解れ（ほつれ）」と「縺れ（もつれ）」というそれぞれの漢字が指すように、
解れはほどけること、縺れは絡まることなので、意味的には真逆です。注意しましょう。

舌縺れ（したもつれ）というのは、舌がからんですらすらと物が言えないことです。
具体的に言うと初対面の方と会話する時に僕のような引きこもりエンジニアはよく舌縺れします。
縺れ髪（もつれがみ）というのは、もつれた髪。乱れ髪のことです。
具体的に言うと一般的エンジニアの髪型のことです。
エンジニアにふさわしい漢字が第1位に輝きましたね！！！（やかましいわ）

結果発表『走れメロス』部門

『こころ』だけで終わらせてしまうと、上の結果は『こころ』でしか通用しないんじゃないの？？？ 他の文章だと全然違う結果になるんじゃないの？？？ と思われがちなので、念のため、他の文章でもランキングをつけてみました。ちょっと順位は変動していますが、一部を除いて概ね似たような結果になっています。
メロスは短編小説なので、テキストの量は少なめです。

41位　遘: 20回（ランク外）
32位　昴: 24回（ランク外）
22位　莠: 48回（ランク外）
―――――――――――――――
20位　譚: 54回 new!（冒険譚の譚）
19位　吶: 56回
18位　偵: 57回 new!（偵察の偵）
17位　輔: 59回
16位　荳: 63回
15位　代: 76回
14位　峨: 80回
13位　翫: 92回
12位　上: 98回
11位　阪: 98回
10位　医: 111回
9位　薙: 117回
8位　励: 149回
7位　繝: 156回 new!（繧繝（うんげん）の繝の方）
6位　溘: 222回
5位　後: 224回
4位　九: 290回
3位　縲: 753回
2位　繧: 933回
1位　縺: 2944回

2位「繧」のパートナーである「繝」が7位にランクインしているところが注目の点ですね。
実は「繝」は『こころ』では172回出現の78位という不完全燃焼な結果となっていましたが、メロスではその実力を遺憾なく発揮してくれました。パートナーの飛躍に「繧」も喜びの表情を浮かべています。

実は「繝」の字は上にも貼った別の方の記事「UTF-8からSJISに文字化けすると糸偏の漢字がよく出てくる」でも紹介されているのですが、「ダチヂッツヅテデトドナニヌネノハバパヒビピフブプヘベペホボポマミムメモャヤュユョヨラリルレロヮワヰヱヲンヴヵヶ」の文字が文字化けすると出てきます。なので、「メロス」を文字化けさせると「繝｡繝ｭ繧ｹ」みたいに「繝」の字が2回も出てきてくれるのですが、『こころ』はカタカナの割合が現在の文章に比べるとかなり少ないので、「繝」選手にとってはやや不利な戦いを強いられてしまった感は否めません。糸偏三銃士がもし糸偏四天王になるとしたら、その最弱ポジションは間違いなく「繝」選手になるでしょう。

おわりに

こ、これはぷろぐらみんぐのきじです……。ぱいそんかいてるもんね……。

文字化けした結果出てくる、なんだか恐ろしげな漢字も、調べてみたら意外と面白かったよということが伝わったら幸いです。憎き謎の漢字も、文学やら歴史やらの背景があって生きている（いた）漢字なので、あまり繧繝ちゃんのことをいじめないであげてください。

はじめに

今年も年末が近づいてきたので無駄に Python C/API を使ってみます。目的と手段の逆転。今回は Awaitable を、次の coroutine function spam を実行して得られる Coroutine のようなものをめざします。

import asyncio


async def spam():
    print('do something')
    ret = await asyncio.sleep(1, 'RETURN VALUE')
    return ret.lower()


async def main():
     ret = await spam()
     print(ret)


if __name__ == '__main__':
    asyncio.run(main())

なにか print して、 await sleep して結果を得て、それの lower メソッドを呼んで結果を返します。 3 動作。 Awaitable はもっていないが Coroutine や Generator ならばもっている send, throw, close メソッドは今回は再現しません。

await と yield from

async def, await と書かれたコードを模倣する前に。そもそもこれはなんなのかを復習します。これらの言語機能が Python に追加提案されたときの文章が PEP 492 です。
これを読んで await とは実質 yield from だったな、と雑に思い出しました。これが用意された経緯は次のようなものだったのではないでしょうか。「Generator の仕様はもともと Coroutine として使われることを視野に入れて設計されていました。一時停止・再開ができるという特性は Iterator そのものでしたし、値を送り込むことができるという特性も PEP 342 にて追加しました。そして Python 3.4 以前まで順調に実績を積んできました。しかし Generator や Iterator と Coroutine の区別がつかないことによる問題が報告され始めたので区別がつくように Python 3.5 の言語仕様に手を入れます。 __iter__ ではなく __await__ という別のメソッドを新設します。 Generator Function 用である yield from 文も流用はせず、 Coroutine Function の中で使うための別の文 await 文を新設して区別します。」 、と。 __await__ と __iter__ という別の名前が用意されてはいますが動作としては異なるものではないのでした。 CPython の PyTypeObject 構造体上でも tp_iter と tp_as_sync.am_await が用意され別のメンバ扱いとなっています。

__await__

Awaitable に対する理解が進んだので。さっそく coroutine function spam を class 文で書き下します。 Iteratable との違いは __iter__ の代わりに __await__ を持っているところですね。とはいえ名前が違うだけで、ここから Iterator を返すという点はかわりません。

class Spam:
    def __await__(self):
        ...  # TODO: どのような Itarator を返せば spam コルーチンを模倣できる？

__iter__ と __next__

分解をつづけます。 Iterator とは __iter__ で自身を返し、 __next__ で処理を再開し次の値を作り処理を中断しつつ値を返すものです。

class _Spam:
    def __iter__(self):
        return self

    def __next__(self):
        ...  # TODO: どうすれば spam コルーチンを模倣できる？

class Spam:
    def __await__(self):
        return _Spam()

さて、「coroutine function spam を実行して得られる Coroutine と似た Awaitable」を実装するためには何が必要でしょうか。「なにか print して、 await sleep して結果を得て、それの lower メソッドを呼んで結果を返す」の 3 動作のうちどこまで処理したかの状態の保持ですね。オブジェクトに _state 属性を持たせるようにします。今回はとりあえず 0, 1, 2 の int で表現しましたが、きれいに書くのであれば Enum をもちいるのがよいでしょう。
そして状態別に処理を行う __next__ を実装します。ここで問題となるのが yield from 。 yield from とは別の Iterator に処理を委譲するものです。別の Iterator が停止するまで繰り返し処理をしつづけて得られた値をそのまま返し続けます。これと同等の処理を実装するために動作中の別の Iterator を保持する必要があります。このため _it 属性をもたせました。この Iterator が停止したら StopIteration 例外が送られてくるので value 属性をみます。これが coroutine function や generator function の return 文に、 yield from の値に対応しています。
逆に値を返すときには StopIteration 例外に値を持たせます。
一度停止した Iterator は StopIteration 例外を返し続けないといけないという制約がありますのでその対策をします。 __next__ の末尾は raise StopIteration にします。状態を保持する属性は外部から書き換えて欲しくないものであることを示すため名前を _ で始めるようにします。この _state の初期化は、 __init__ の複数回の呼び出しへの耐性をもたせるため __new__ でおこないます。

class _Spam:
    def __new__(cls):
        obj = super().__new__(cls)
        obj._state = 0
        obj._it = None
        return obj

    def __iter__(self):
        return self

    def __next__(self):
        if self._state == 0:
            print('do something')
            self._it = asyncio.sleep(1, 'RETURN VALUE').__await__()
            self._state = 1
        if self._state == 1:
            try:
                v = next(self._it)
            except StopIteration as e:
                ret = e.value
                self._it = None
                self._state = 2
                raise StopIteration(ret.lower())
            else:
                return v
        raise StopIteration


class Spam:
    def __await__(self):
        return _Spam()

Python C/API でクラスを書こう

分解が終わったので。これを Python C/API でかきます。ここから C 言語。クラスをつくるには PyTypeObject 構造体を直接書くか PyType_Spec を定義して PyType_FromSpec を呼びます。今回は PyType_FromSpec を使う方向で。
まずは _Spam よりは簡単な Spam から。 __await__ をつくるには tp_as_sync.am_await に関数を登録します。このメソッドからは _Spam のインスタンスを返す必要があるので Spam クラスの属性に _Spam クラスを持たせるようにすることにします。クラス属性追加処理は Py_mod_exec として登録するモジュールの初期化処理にて PyType_FromSpec でのクラス作成後に行います。

typedef struct {
    PyObject_HEAD
} SpamObject;


static PyObject *
advent2019_Spam_await(SpamObject *self)
{
    PyObject *_Spam_Type = PyObject_GetAttrString((PyObject *)Py_TYPE(self), "_Spam");
    if (_Spam_Type == NULL) { return NULL; }
    PyObject *it = PyObject_CallFunction(_Spam_Type, "");
    Py_DECREF(_Spam_Type);

    return it;
}


static PyType_Slot advent2019_Spam_slots[] = {
    {Py_am_await, (unaryfunc)advent2019_Spam_await},
    {0, 0},
};


static PyType_Spec advent2019_Spam_spec = {
    .name = "advent2019.Spam",
    .basicsize = sizeof(SpamObject),
    .flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE,
    .slots = advent2019_Spam_slots,
};

static int advent2019_exec(PyObject *module) {
    int ret = -1;
    PyObject *_Spam_Type = NULL;  // TODO
    PyObject *Spam_Type = NULL;

    if (!(Spam_Type = PyType_FromSpec(&advent2019_Spam_spec))) { goto cleanup; }
    // Spam._Spam = _Spam
    if (PyObject_SetAttrString(Spam_Type, "_Spam", _Spam_Type)) { goto cleanup; }

    if (PyObject_SetAttrString(module, "Spam", Spam_Type)) { goto cleanup; }
    if (PyObject_SetAttrString(module, "_Spam", _Spam_Type)) { goto cleanup; }

    ret = 0;
cleanup:
    Py_XDECREF(_Spam_Type);
    Py_XDECREF(Spam_Type);

    if (ret) { Py_XDECREF(module); }
    return ret;
}

C/API で書く Iterator

さて、後回しにした Iterator _Spam の実装ですね。まずは _SpamObject 構造体と __new__ から。状態を保持するための何らかの値 state と asyncio.sleep().__await__ イテレータを保持する it を用意。他の PyObject* を保持することになるのでガベージコレクション機構に対応する PyObject_GC_New でメモリを確保するようにします。また、初期化が終わったら PyObject_GC_Track を呼び、ガベージコレクタに自身を登録します。

typedef struct {
    PyObject_HEAD
    unsigned char state;
    PyObject *it;
} _SpamObject;


static PyObject *
advent2019__Spam_new(PyTypeObject *type, PyObject *args, PyObject *kwargs)
{
    static char *kwlist[] = {NULL};

    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "", kwlist)) {
        return NULL;
    }

    _SpamObject *obj = PyObject_GC_New(_SpamObject, type);
    if (!obj) { return NULL; }
    obj->state = 0;
    obj->it = NULL;

    PyObject_GC_Track(obj);

    return (PyObject *)obj;
}

ガベージコレクションが動いた時用の関数が必要になります。抱えているオブジェクトを手繰れるようにする traverse と、参照を破棄する clear 、自身を破棄する dealloc の 3 つです。 Py_VISIT マクロ は traverse を書くのに便利。 PyObject_GC_Track と対になる PyObject_GC_UnTrack は dealloc の開始時に呼び出すようにします。

static int
advent2019__Spam_traverse(_SpamObject *self, visitproc visit, void *arg)
{
    Py_VISIT(self->it);
    return 0;
}

static int
advent2019__Spam_clear(_SpamObject *self)
{
    Py_CLEAR(self->it);
    return 0;
}

static void
advent2019__Spam_dealloc(_SpamObject *self)
{
    PyObject_GC_UnTrack(self);
    advent2019__Spam_clear(self);
    PyObject_GC_Del(self);
}

__iter__ は自身を返すだけなので簡単ですね。参照カウントの操作を忘れないようにしつつ。

static PyObject *
advent2019__Spam_iter(_SpamObject *self)
{
    Py_INCREF(self);
    return (PyObject *)self;
}

最後に __next__ 。 Python C/API 上では iternext という名前になります。組み込み関数 print と asyncio モジュールの sleep はクラス作成時に _print と _sleep という名前でクラス属性に持たせるようにしておきます、 Spam._Spam と同じ要領で。
あとは Python で書いたコードを PyObject_GetAttrString, PyObject_CallFunction らを用いて地道に移植していく作業です。呼ぶたびに戻り値が NULL になっていないかを確認します。不必要となったオブジェクトの参照カウントを減らす処理がしんどい。
try 文の移植には PyErr_Fetch, PyErr_GivenExceptionMatches, PyErr_Restore を使います。

static PyObject *
advent2019__Spam_iternext(_SpamObject *self)
{
    if (self->state == 0) {
        // print('do something')
        PyObject *printfunc = PyObject_GetAttrString((PyObject *)Py_TYPE(self), "_print");
        if (!printfunc) { return NULL; }
        PyObject *ret = PyObject_CallFunction(printfunc, "s", "do something");
        Py_DECREF(printfunc);
        if (!ret) { return NULL; }
        Py_DECREF(ret);

        // self._it = asyncio.sleep(1, 'RETURN VALUE').__await__()
        PyObject *sleep_cofunc = PyObject_GetAttrString((PyObject *)Py_TYPE(self), "_sleep");
        if (!sleep_cofunc) { return NULL; }
        PyObject *sleep_co = PyObject_CallFunction(sleep_cofunc, "is", 1, "RETURN VALUE");
        Py_DECREF(sleep_cofunc);
        if (!sleep_co) { return NULL; }
        if (!(Py_TYPE(sleep_co)->tp_as_async)) { Py_DECREF(sleep_co);  return NULL; }
        if (!(Py_TYPE(sleep_co)->tp_as_async->am_await)) { Py_DECREF(sleep_co);  return NULL; }
        PyObject *temp = self->it;
        self->it = Py_TYPE(sleep_co)->tp_as_async->am_await(sleep_co);
        Py_DECREF(sleep_co);
        Py_XDECREF(temp);
        if (self->it == NULL) { return NULL; }

        self->state = 1;
    }
    if (self->state == 1) {
        // next(self.it)
        if (Py_TYPE(self->it)->tp_iternext == NULL) { PyErr_SetString(PyExc_TypeError, "no iternext"); return NULL; }
        PyObject *ret = Py_TYPE(self->it)->tp_iternext(self->it);
        if (!ret) {
            // except StopIteration as e
            PyObject *type, *value, *traceback;
            PyErr_Fetch(&type, &value, &traceback);
            if (PyErr_GivenExceptionMatches(type, PyExc_StopIteration)) {
                Py_XDECREF(type);
                Py_XDECREF(traceback);
                if (!value) { PyErr_SetString(PyExc_ValueError, "no StopIteration value"); return NULL; }
                // ret = e.value.lower()
                PyObject *value2 = PyObject_CallMethod(value, "lower", NULL);
                Py_DECREF(value);
                if (!value2) { return NULL; }
                // raise StopIteration(ret)
                PyErr_SetObject(PyExc_StopIteration, value2);
                Py_DECREF(value2);

                Py_CLEAR(self->it);
                self->state = 2;
            } else {
                // except:
                //     raise
                PyErr_Restore(type, value, traceback);
            }
        }
        return ret;
    }

    // raise StopIteration(None)
    PyErr_SetNone(PyExc_StopIteration);
    return NULL;
}

これで _Spam クラスのメソッドがそろったので PyType_Spec を定義します。ガベージコレクションで管理されるべきクラスであることををしめすフラグ Py_TPFLAGS_HAVE_GC を設定するようにします。

static PyType_Slot advent2019__Spam_slots[] = {
    {Py_tp_new, advent2019__Spam_new},
    {Py_tp_iter, advent2019__Spam_iter},
    {Py_tp_iternext, advent2019__Spam_iternext},
    {Py_tp_traverse, advent2019__Spam_traverse},
    {Py_tp_clear, advent2019__Spam_clear},
    {Py_tp_dealloc, advent2019__Spam_dealloc},
    {0, 0},
};


static PyType_Spec advent2019__Spam_spec = {
    .name = "advent2019._Spam",
    .basicsize = sizeof(_SpamObject),
    .flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_GC,
    .slots = advent2019__Spam_slots,
};

おわりに

Python C/API で Awaitable を実装してみました。また、これをおこなうにあたって必要な情報がまとまっている公式ドキュメントへのリンクを集めることができました。
ところで。これは役に立つのでしょうか？ たったこれだけのことのためのコードがずいぶんと長く、 async def, await 構文の便利さを思い知っただけのような、いや CPython への理解を深めるのには役立つかもしれない………。

setup.cfg

[metadata]
name = advent2019
version = 0.0.0

[options]
python_requires = >=3.5.0

setup.py

from setuptools import Extension, setup

extensions = [Extension('advent2019', sources=['advent2019.c'])]

setup(ext_modules=extensions)

advent2019.c

#define PY_SSIZE_T_CLEAN
#include <Python.h>


typedef struct {
    PyObject_HEAD
    unsigned char state;
    PyObject *it;
} _SpamObject;


static PyObject *
advent2019__Spam_new(PyTypeObject *type, PyObject *args, PyObject *kwargs)
{
    static char *kwlist[] = {NULL};

    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "", kwlist)) {
        return NULL;
    }

    _SpamObject *obj = PyObject_GC_New(_SpamObject, type);
    if (!obj) { return NULL; }
    obj->state = 0;
    obj->it = NULL;

    PyObject_GC_Track(obj);

    return (PyObject *)obj;
}


static PyObject *
advent2019__Spam_iter(_SpamObject *self)
{
    Py_INCREF(self);
    return (PyObject *)self;
}


static PyObject *
advent2019__Spam_iternext(_SpamObject *self)
{
    if (self->state == 0) {
        // print('do something')
        PyObject *printfunc = PyObject_GetAttrString((PyObject *)Py_TYPE(self), "_print");
        if (!printfunc) { return NULL; }
        PyObject *ret = PyObject_CallFunction(printfunc, "s", "do something");
        Py_DECREF(printfunc);
        if (!ret) { return NULL; }
        Py_DECREF(ret);

        // self._it = asyncio.sleep(1, 'RETURN VALUE').__await__()
        PyObject *sleep_cofunc = PyObject_GetAttrString((PyObject *)Py_TYPE(self), "_sleep");
        if (!sleep_cofunc) { return NULL; }
        PyObject *sleep_co = PyObject_CallFunction(sleep_cofunc, "is", 1, "RETURN VALUE");
        Py_DECREF(sleep_cofunc);
        if (!sleep_co) { return NULL; }
        if (!(Py_TYPE(sleep_co)->tp_as_async)) { Py_DECREF(sleep_co);  return NULL; }
        if (!(Py_TYPE(sleep_co)->tp_as_async->am_await)) { Py_DECREF(sleep_co);  return NULL; }
        PyObject *temp = self->it;
        self->it = Py_TYPE(sleep_co)->tp_as_async->am_await(sleep_co);
        Py_DECREF(sleep_co);
        Py_XDECREF(temp);
        if (self->it == NULL) { return NULL; }

        self->state = 1;
    }
    if (self->state == 1) {
        // next(self.it)
        if (Py_TYPE(self->it)->tp_iternext == NULL) { PyErr_SetString(PyExc_TypeError, "no iternext"); return NULL; }
        PyObject *ret = Py_TYPE(self->it)->tp_iternext(self->it);
        if (!ret) {
            // except StopIteration as e
            PyObject *type, *value, *traceback;
            PyErr_Fetch(&type, &value, &traceback);
            if (PyErr_GivenExceptionMatches(type, PyExc_StopIteration)) {
                Py_XDECREF(type);
                Py_XDECREF(traceback);
                if (!value) { PyErr_SetString(PyExc_ValueError, "no StopIteration value"); return NULL; }
                // ret = e.value.lower()
                PyObject *value2 = PyObject_CallMethod(value, "lower", NULL);
                Py_DECREF(value);
                if (!value2) { return NULL; }
                // raise StopIteration(ret)
                PyErr_SetObject(PyExc_StopIteration, value2);
                Py_DECREF(value2);

                Py_CLEAR(self->it);
                self->state = 2;
            } else {
                // except:
                //     raise
                PyErr_Restore(type, value, traceback);
            }
        }
        return ret;
    }

    // raise StopIteration(None)
    PyErr_SetNone(PyExc_StopIteration);
    return NULL;
}


static int
advent2019__Spam_traverse(_SpamObject *self, visitproc visit, void *arg)
{
    Py_VISIT(self->it);
    return 0;
}

static int
advent2019__Spam_clear(_SpamObject *self)
{
    Py_CLEAR(self->it);
    return 0;
}

static void
advent2019__Spam_dealloc(_SpamObject *self)
{
    PyObject_GC_UnTrack(self);
    advent2019__Spam_clear(self);
    PyObject_GC_Del(self);
}


static PyType_Slot advent2019__Spam_slots[] = {
    {Py_tp_new, advent2019__Spam_new},
    {Py_tp_iter, advent2019__Spam_iter},
    {Py_tp_iternext, advent2019__Spam_iternext},
    {Py_tp_traverse, advent2019__Spam_traverse},
    {Py_tp_clear, advent2019__Spam_clear},
    {Py_tp_dealloc, advent2019__Spam_dealloc},
    {0, 0},
};


static PyType_Spec advent2019__Spam_spec = {
    .name = "advent2019._Spam",
    .basicsize = sizeof(_SpamObject),
    .flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_GC,
    .slots = advent2019__Spam_slots,
};


typedef struct {
    PyObject_HEAD
} SpamObject;


static PyObject *
advent2019_Spam_await(SpamObject *self)
{
    PyObject *_Spam_Type = PyObject_GetAttrString((PyObject *)Py_TYPE(self), "_Spam");
    if (_Spam_Type == NULL) { return NULL; }
    PyObject *it = PyObject_CallFunction(_Spam_Type, "");
    Py_DECREF(_Spam_Type);

    return it;
}


static PyType_Slot advent2019_Spam_slots[] = {
    {Py_am_await, (unaryfunc)advent2019_Spam_await},
    {0, 0},
};


static PyType_Spec advent2019_Spam_spec = {
    .name = "advent2019.Spam",
    .basicsize = sizeof(SpamObject),
    .flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE,
    .slots = advent2019_Spam_slots,
};


static int advent2019_exec(PyObject *module) {
    int ret = -1;
    PyObject *builtins = NULL;
    PyObject *printfunc = NULL;
    PyObject *asyncio_module = NULL;
    PyObject *sleep = NULL;
    PyObject *_Spam_Type = NULL;
    PyObject *Spam_Type = NULL;

    if (!(builtins = PyEval_GetBuiltins())) { goto cleanup; }  /* borrowed */
    // fetch the builtin function print
    if (!(printfunc = PyMapping_GetItemString(builtins, "print"))) { goto cleanup; }

    // import asyncio
    if (!(asyncio_module = PyImport_ImportModule("asyncio"))) { goto cleanup; }
    if (!(sleep = PyObject_GetAttrString(asyncio_module, "sleep"))) { goto cleanup; };

    if (!(_Spam_Type = PyType_FromSpec(&advent2019__Spam_spec))) { goto cleanup; }
    // _Spam._print = print
    if (PyObject_SetAttrString(_Spam_Type, "_print", printfunc)) { goto cleanup; }
    // _Spam._sleep = asyncio.sleep
    if (PyObject_SetAttrString(_Spam_Type, "_sleep", sleep)) { goto cleanup; }

    if (!(Spam_Type = PyType_FromSpec(&advent2019_Spam_spec))) { goto cleanup; }
    // Spam._Spam = _Spam
    if (PyObject_SetAttrString(Spam_Type, "_Spam", _Spam_Type)) { goto cleanup; }

    if (PyObject_SetAttrString(module, "Spam", Spam_Type)) { goto cleanup; }
    if (PyObject_SetAttrString(module, "_Spam", _Spam_Type)) { goto cleanup; }

    ret = 0;
cleanup:
    Py_XDECREF(printfunc);
    Py_XDECREF(asyncio_module);
    Py_XDECREF(sleep);
    Py_XDECREF(_Spam_Type);
    Py_XDECREF(Spam_Type);

    if (ret) { Py_XDECREF(module); }
    return ret;
}


static PyModuleDef_Slot advent2019_slots[] = {
    {Py_mod_exec, advent2019_exec},
    {0, NULL}
};


static struct PyModuleDef advent2019_moduledef = {
    PyModuleDef_HEAD_INIT,
    .m_name = "advent2019",
    .m_slots = advent2019_slots,
};


PyMODINIT_FUNC PyInit_advent2019(void) {
    return PyModuleDef_Init(&advent2019_moduledef);
}

これをつかってみるコードの例

import sys
import asyncio

import advent2019


async def main():
    v = await advent2019.Spam()
    print(v)


if __name__ == '__main__':
    if sys.version_info < (3, 7):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(main())
        loop.close()
    else:
        asyncio.run(main())

はじめに

最近、無謀にもDDDの勉強を始めたのですが、Pythonの例が少ない・・・！
ということで、
・DDDに関する自分の考えを整理する
・Pythonによる記述方法の一例を提案する
という目的で記事を書いてみることにしました。
とりあえず今回は一番とっつきやすそうなValueObjectについて書いています。

ValueObjectと完全コンストラクタとは・・・

ちゃんとした説明は他の方々のとても分かりやすい記事がたくさんあるのでそちらに譲るとして、簡単に書くと、
ValueObjectは「値をintなどのプリミティブ型ではなく、期待している振る舞い、在り方をクラスを使って表現したもの」、
完全コンストラクタは「期待している振る舞い」や、逆に「期待しない振る舞いをさせない」を実現するための手法と言えます。
(私はこう理解しています・・・合ってる・・・？)

例えば、「金額」を表現するのであれば「負の値にはならない」など、表現する対象物に対して
本来あるべき振る舞いなどをクラスで定義したものがValueObjectです。
上記の「負の値にはならない」というのをコンストラクタの中で記述して、
「存在してはならない値を存在させない」ということを実現するのが完全コンストラクタです。

dataclassesでValueObjectを表現してみる

前節で金額の話をしたので、金額を例に簡単に書いてみます。

import dataclasses

@dataclasses.dataclass
class Money:
    # 詳細は省きますが、ここに挙げた変数が通常のクラスの __init__ に記述するインスタンス変数となります
    amount: int

    # __post_init__ は __init__ の後に実行される処理です。
    # 完全コンストラクタの表現をここに記述します。
    def __post__init__(self):
        if self.amount < 0:
            raise ValueError()

これで負の値を持ったMoneyオブジェクトは存在できず、
正しい値を持っていることが保証されたMoneyオブジェクトだけが存在することになります。
(他にも課すべき制限はあるかと思いますが、今回はこの程度にして話を先に進めます)

イミュータブルにする

dataclassesを使うと、クラスをイミュータブルにすることもできます。
イミュータブルにすることで再代入が不可となり、コードの安全性が向上します。
(ValueObjectは基本的にイミュータブルにするべきもののようです)
Pythonはインスタンス変数への再代入を防ぐ方法が乏しいため、重要な機能です。

import dataclasses

# frozen=Trueとするとイミュータブルになる
@dataclasses.dataclass(frozen=True)
class Money:
    amount: int

    def __post__init__(self):
        if amount < 0:
            raise ValueError()

金額の変化をどう表現するか

イミュータブルにしたことで一度インスタンス化すると、その金額の増減が発生しても値を変更できなくなりました。
では、金額の増減はどう表現するか、増減後の金額の値を持った新しいインスタンスを生成します。
メソッドは通常のクラスと同様に記述できます。

import dataclasses

@dataclasses.dataclass(frozen=True)
class Money:
    amount: int

    def __post__init__(self):
        if amount < 0:
            raise ValueError()

    def lost(self, loss):
        return Money(self.amount - loss.amount)

# 例えばこんな感じに書きます
money1 = Money(1000)
money2 = Money(100)
left_money = money1.lost(money2)

dataclassesを使うべきか

従来の記述方法でも同様のことは可能なので必須ではありません。
ただ、
・イミュータブルにできる
・コンストラクタの記述量が減らせる
・インスタンス変数の記述と完全コンストラクタにするためのロジックの記述を分離でき、見やすくなる
という点でdataclassesを使ったほうがいいかなぁという印象です。
dataclassesは他にも様々な機能があるのでValueObjectに限らず使ってみたいですね。

最後に

このように書けばPythonでもValueObjectを表現できる、ということを今回紹介させていただきました。
まだ勉強し始めたばかりのためValueObjectそのものについての説明は薄いですが、理解が深まれば
補足の記事を書いたり、エンティティなどについても記事にできればと思っています。

参考

dataclassesの用法はこちらを参考にしています。
https://docs.python.org/ja/3.7/library/dataclasses.html

はじめに

OpenCVやPILの話。
Qiitaに限らずウェブ上にいくらでもある情報ではあるが、記事が細切れになっているものが多く比較するには適していないのでその点に注意してまとめてみた。
その過程で先人が触れていない事項にたどり着いた（ように思える）ので、お読みいただけると幸いです。

OpenCV

CVはComputer Visionの略で、単に画像を加工するだけでなくさまざまな機能を持つ。
まあ、今回は画像加工どころか画像を表示するだけの話なのだが。

画像を読み込む　imread()

cv2.imread(filename, flags)という使い方をする。
filenameはファイル名。これが適切でない場合でもエラーにならずNoneを返す。
flagsについては後述する。

画像読み込み

import cv2

filename = "hoge.png"
imgCV = cv2.imread(filename)  # flagsは省略（デフォ値＝1）

画像を表示する　imshow()

cv2.imshow(winname, mat) という使い方をする。
winnameはウィンドウの名前。ヌルストリングでもいいが指定は必須。
matはマトリックス、すなわち行列。要するに画像データ。
デフォではサイズ変更ができないが、imshowする前にnamedWindowで「これこれこういう名前でサイズ変更可能なウィンドウを作るよ」としてやればサイズ変更できるようになる。
imreadを失敗してNoneが返ってきた場合、ここでエラーになる。

サイズ変更不可のウィンドウで画像表示

cv2.namedWindow("image", cv2.WINDOW_AUTOSIZE)  # この一文、なくてもよい
cv2.imshow("image", imgCV)

サイズ変更可のウィンドウで画像表示

cv2.namedWindow("image", cv2.WINDOW_NORMAL)  # cv2.WINDOW_NORMALの値は0なので0を指定してもよい
cv2.imshow("image", imgCV)

OpenCVの画像データについて

画像データはnumpy.ndarrayの型で構成されている。シェイプで次元を調べることができるし内容を確認するのも容易だ。
カラーはBGRの順で格納されていることに注意。

画像のシェイプ

flagsを指定して画像を読み込んだとき、画像データがどのようなカタチになっているか確認してみよう。

flags	cv2.IMREAD_COLOR	cv2.IMREAD_GRAYSCALE	cv2.IMREAD_UNCHANGED
値	1 （デフォルト値）	0	-1
処理	カラー画像として読み込む	グレースケールで読み込む	そのままの仕様で読み込む
例1 RGBA画像 実際の背景は市松模様ではなく透明	shape=(200, 182, 3)	shape=(200, 182)	shape=(200, 182, 4)
例2 グレースケール画像	shape=(192, 144, 3)	shape=(192, 144)	shape=(192, 144)

カラーチャンネル数は、cv2.IMREAD_COLORでは透明度があってもグレースケール画像であっても一律で3チャンネルの配列になる。cv2.IMREAD_GRAYSCALEの場合はチャンネル数が1になるのではなく、チャンネル数の指定のない(h, w)という二次元配列になる。
cv2.IMREAD_UNCHANGEDは、OpenCV-Pythonチュートリアル「画像を扱う」には「アルファチャンネルも含めた画像として読み込む」とあるが、一律4チャンネルになるわけではなく、元の画像から変更なく読み込むというのが正しい。RGBA画像なら4チャンネル、RGB画像なら3チャンネル。グレースケール画像なら1チャンネルなので(h, w, 1)ではなく(h, w)を返す。
ややこしい？ いや、最初からそう言ってるでしょ、cv2.IMREAD_UNCHANGEDって。

サイズを取得する

元画像のタイプによって配列のカタチが異なるOpenCVの画像データ。高さや幅を取得するにはif文を使って場合分けしなくてはいけない…ということはない。
(高さ, 幅)で得られるグレースケール画像のシェイプも、(高さ, 幅, チャンネル数)となるカラー画像も、0番目に高さがあって1番目に幅があることは同じ。だから0や1で決め打ちしてやればいいのだ。

# どちらでも可

def getSize1(imgCV):
    h = imgCV.shape[0]
    w = imgCV.shape[1]
    return h, w

def getSize2(imgCV):
    h, w = imgCV.shape[:2]
    return h, w

Google Colabの場合

Google Colabではcv2.imshow()は使えないようになっている。Jupyterのセッションがクラッシュしてしまうらしい。
代わりにcv2_imshow()というGoogle Colab独自のメソッドを使うよう代替案を提示してくる。
cv2_imshow()にウィンドウ名は不要で画像データのみを指定する。これによりColabのセル内に画像が表示される。

Google Colab

# 1セル1文でも全部まとめても可
import cv2
from google.colab.patches import cv2_imshow

filename = "hoge.png"
imgCV = cv2.imread(filename)

cv2_imshow(imgCV)

Jupyter Notebookの場合

Jupyter Notebookではcv2.imshow()は禁止されているわけではないが、クラッシュすることがあるのは変わらない。

実は、Jupyter Notebookでは、正しい処理をすることで正しく表示させることができる。
先程のサイトにあるようにcv2.imshow()で表示した後、キー入力を待ってウィンドウを破棄するようにすればいいのだ。
この場合でも右上バッテンで画像ウィンドウを閉じようとしてはいけない。やはりクラッシュしてしまう。
正しく表示できるといってもミスったら死だなんて嫌すぎる。そんなのはゲームの世界だけで十分だ。

Jupyter Notebook

import cv2

filename = "hoge.png"
imgCV = cv2.imread(filename)

# 以下を一つのセルで実行する
cv2.imshow("image",imgCV)
cv2.waitKey(0)
cv2.destroyAllWindows()

正しい処理とはどういうことか

画像表示するにはcv2.imshow()だぞと書いたが、この段階ですでにつまづいている人もいるかもしれない。
Pythonをインストールしたときに一緒に付いてくる開発環境IDLE上で実行するとうまくいく。だがpyファイルをダブルクリックすると期待通りの動きにならない。VS Codeでもダメ。
これは、IDLEではプログラムが終了してもシェルが生き続けているのに対し、python.exeを実行する場合はそれが終了した瞬間に画像のウィンドウも閉じてしまうから。だと思われる。
cv2で作成したウィンドウはcv2で破棄する。これがこの世界のジャスティスなのだ。

PIL(Pillow)

PIL(Python Image Library)という画像処理ライブラリがあって、その後継がPillow。OpenCVとの得手不得手の差はまた今度。
Pillowをインストールしても、実際に使う際にインポートするのはPIL。

画像を読み込む　Image.open()

Image.open(filename)という使い方をする。
filenameが適切でない場合はエラーになる。
厳密にはmodeという引数もありデフォ値が"r"なのだが、これが何を意味しそれ以外にどんな値が使えるのかよくわからない。

画像を表示する　show()

読み込んだ画像データをshow()することで画像が表示される。
画像はOSごとに異なる画像ビューアが起動して表示される。ちょいと不便だな。
引数にはtitleやcommandがあるがいずれも省略可能。言うまでもないことだが、引数がなくてもカッコは必須。

ソース

from PIL import Image

filename = "hoge.png"
imgPIL = Image.open(filename)  # 画像読み込み

imgPIL.show()  # 画像表示

PILの画像データについて

画像データは、たとえばpng画像ならばPIL.PngImagePlugin.PngImageFileという形式になっており、中身を確認するのは容易ではない。
その代わり、これは画像データであると自覚しているのでさまざまな属性を持っている。

サイズを取得する

print (imgPIL.mode)
# RGBA  # ほかに RGB L（グレースケール）などがある。
# Image.open()のmodeとの関係は不明。

print (imgPIL.size)
# (182, 200)  # タプルで、幅,高さ の順

print (imgPIL.width)
# 182

print (imgPIL.height)
# 200

matplotlibのグラフとして表示

画像をmatplotlibのグラフとして表示することも多い。
matplotlib.pyplotの詳しい使い方はここでは説明しない。

python上で実行すると、インタラクティブなmatplotlibのグラフがあらわれる。拡大したり表示エリアを変えたりできる。

Google ColabやJupyter Notebook上では単なる画像としてグラフが表示される。
Jupyter Notebookでは%matplotlib inlineというおまじないを唱えるといいらしい。

え？ 普通に画像を表示する場合との違いがわからない？
では、こんな画像ならばどうする？
　← ここにいる
これは6×8の画像。matplotlibのグラフはこんな小さい画像もいい感じに拡大してくれるのがありがたい。

PIL画像の場合

PIL画像はそのままではグラフ化できないのでnumpy.asarray()でnumpy.ndarrayにしてやる必要がある。

ソース

import numpy as np
from PIL import Image
from matplotlib import pyplot as plt
%matplotlib inline  # Jupyter Notebookでインライン表示する

filename = "hoge.png"
imgPIL = Image.open(filename)
arrPIL = np.asarray(imgPIL)
plt.imshow(arrPIL)
plt.show()

OpenCVの場合

ここのみ諸事情によりサンプル画像がいらすとやでなくスキマナースになっています。

これが元画像。

OpenCの画像データはもとよりnumpy.ndarray。ならばそのままplt.imshowすればいい？
とやってみると。

はいダメー。OpenCVの画像はBGRだって言ったでしょ。matplotlib.pyplotは普通にRGBなのでOpenCVの画像をmatplotlibでグラフ表示する際は色を変換してやる必要があるのだ。

カラーをコンバートするにはcv2.cvtColorを使う。cv2.cvtColor(src, code)という使い方をする。
srcはソース。元の画像データ。
codeは色変換の組み込み定数。BGRをRGBにするとか、その逆とか、RGBをグレーにするとか、RGBをRGBAにするとか、いろいろある。BGRをRGBにコンバートするのはcv2.COLOR_BGR2RGB。
このひと手間によってOpenCVの画像もmatplotlibのグラフとして表示できる。

BGR→RGBは、(高さ, 幅, BGR値)というシェイプの配列の2番目の並びをRGBと逆順にすることにほかならない。
前回学んだスライスを活用することもできる。

ソース

import numpy as np
import cv2
from matplotlib import pyplot as plt

filename = "nurse.jpg"
imgCV = cv2.imread(filename)

# cv2.cvtColorを使う方法
imgCV_RGB = cv2.cvtColor(imgCV,cv2.COLOR_BGR2RGB)

# スライスを使う方法
# imgCV_RGB = imgCV[:, :, ::-1]

plt.imshow(imgCV_RGB)
plt.show()

ここで透過を持つpng画像を使わなかったのは、今後詳しく調べていくため。
で、できなかったからじゃないんだからね、勘違いしないでよね。
証拠として、グラフ画像として透過が正しく表現できた例を示す。

次回予告

画像と画像を合成する、その際に手前の画像の透明部分は背後の画像が透けて見えるようにする。そんな80年代のゲーマーなら「ああ、スプライトね」と言いたくなるようなマスク処理を手作業でおこなっていきます。

最後に、もう一度透過画像をさまざまなflagsで表示する表をもう一度見てみよう。
がんばって作ったので予習として。

元画像	cv2.IMREAD_COLOR	cv2.IMREAD_GRAYSCALE	cv2.IMREAD_UNCHANGED

目的

• 時系列データときたらとりあえずRNN！、LSTM！！っていう感じだったのですが、ARIMAとかSARIMAなるものを知り自分の引き出しに加えたいということで勉強。
• ライブラリ使ってそれにぶち込むのは簡単だろうけど、一応理論も知っておきたいということで、本を購入した。
• 沖本竜義先生の『経済・ファイナンスデータの計量時系列分析』という本。2010年の本です。

データ

読みながら手も動かしていこうということでとりあえずTOPIXのヒストリカルデータを使うことにした。
https://quotes.wsj.com/index/JP/TOKYO%20EXCHANGE%20(TOPIX)/180460/historical-prices
WSJのHPからダウンロードできた。

とりあえず欠損値はない模様。
期間は2008年12月30日から2019年11月29日まで。
配当落ち調整なしの４本値のヒストリカルデータ。
終値をプロットすると下図のような感じ。
ちょうどリーマンショック直後の1,000pt割れの時期からアベノミクスを経てというような期間です。

日次リターンは以下の通り。日次なので大した差は出ないだろうが対数リターン（$\Delta \log{y_t} = \log{(y_t)} - \log{(y_{t-1})}$）を使用した。


最大の下げを記録したのは2011年3月15日で、$\Delta \log{y_t} = -0.0995$となっている。前日引け前15分くらいのところで東北地方太平洋沖地震が起きたのが原因。

各種統計量

平均
$\bar{y} = \frac{1}{T}\displaystyle{\sum_{t=1}^{T}y_t}$

tpx_return = np.log(tpx['close'].values)[1:] - np.log(tpx['close'].values)[:-1]
tpx_return.mean()

0.00025531962222667643

自己共分散(auto-covariance)
$\hat{\gamma}_k = \frac{1}{T} \displaystyle{\sum_{t=k+1}^{T}}(y_t-\bar{y})(y_{t-k}-\bar{y}),\quad k = 0,1,2,...$

import statsmodels.api as sm
sm.tsa.stattools.acovf(tpx_return, fft=True, nlag=5)

array([ 1.57113176e-04,  1.16917913e-06,  3.48846296e-06, -4.67502133e-06, -5.31500282e-06, -2.82855375e-06])

statsmodelsのライブラリに慣れていないので一応手でも確認。

# k=0
((tpx_return-tpx_return.mean())**2).sum() / len(tpx_return)

0.00015711317609153836

# k=4
((tpx_return-tpx_return.mean())[4:]*(tpx_return-tpx_return.mean())[:-4]).sum() / len(tpx_return)

-5.315002816332674e-06

自己相関係数(auto-correlation)
$\hat{\rho}_k = \frac{\hat{\gamma}_k}{\hat{\gamma}_0},\quad k=1,2,3,...$

sm.tsa.stattools.acf(tpx_return)[:5]

array([ 1.        ,  0.00744164,  0.0222035 , -0.02975576, -0.03382913])

先ほどの$k=0$と$k=4$の結果を使って確認すると、

-5.315002816332674e-06 / 0.00015711317609153836

-0.03382913482212345

ということでライブラリはイメージした通りの計算をしてくれている様子。

コレログラムも描いてみる。コレログラムとは自己相関係数をグラフにしたもののこと。

autocorr = sm.tsa.stattools.acf(tpx_return)
ci = 1.96 / np.sqrt(len(tpx_return))
plt.bar(np.arange(len(autocorr)), autocorr)
plt.hlines([ci,-ci],0,len(autocorr), linestyle='dashed')
plt.title('Correlogram')
plt.ylim(-0.05,0.05)
plt.show()

CI(confidence interval)とは信頼区間のことで、データが互いに独立で同一の分布に従う時、$\hat{\rho}_k$が漸近的に平均$0$、分散$\frac{1}{T}$の正規分布に従うという性質を利用して両側95%を計算している。

ライブラリを使うとこんな感じ。オシャレ。

sm.graphics.tsa.plot_acf(tpx_return)
plt.ylim(-0.05,0.05)
plt.show()

得られた数字について少し考えてみる。
$\hat{\rho}_{k=11}=-0.0421, \quad \hat{\rho}_{k=16}=0.0415$
の２回　$CI=0.0379$　をわずかながら上回っている。
特に　$\hat{\rho}_{k=12}=-0.0323$　もマイナスの値となっており、１回相場が上昇・下落を始めても一旦そのモメンタムが２週間程度で終わることを示唆していると捉えることができるかもしれない。

かばん検定（portmanteau test）

複数の自己相関係数がすべて0であるという帰無仮説を検定する手法。
$H_0 : \rho_1 = \rho_2 =\quad ... \quad= \rho_m = 0$
ここでは本で紹介されているLjung and Box(1978)の統計量を用いて検定を行ってみる。下の$Q(m)$とカイ２乗分布の95%点を比較するというアプローチである。
$Q(m) = T(T+2)\displaystyle{\sum_{k=1}^{m}}\frac{\hat{\rho}^2_k}{T-k} \sim \chi^2(m)$
また、P値という統計量も定義されており、カイ２乗分布に従う確率変数が$Q(m)$より大きな値をとる確率を示したものである。すなわち、有意水準5%とすれば、P値が0.05より小さいとき$H_0$は棄却される。
$m$の値については、$m \approx \log{(T)}$が目安とされるようだが、複数の$m$に対して検定を行い総合的に判断するのが一般的なようである。

まずはstatsmodelsのライブラリを使うパターンから。
$m \approx \log{(T)} = 7.89$　ということで、とりあえずラグは16までの範囲で考えてみる。

lvalue, pvalue = sm.stats.diagnostic.acorr_ljungbox(tpx_return)

これだけで完了。非常に簡単。

$m$がどの値でもP値は0.05以下にはならず、TOPIXの日次変化率は自己相関を持つとは言えないという結果となった。まあ相場はそんなに単純なものではないってことですね。

最後に自分の理解を深めるためライブラリなしでも試してみる。

from scipy.stats import chi2

def Q_func(data, max_m):
    T = len(data)
    auto_corr = sm.tsa.stattools.acf(data)[1:]
    lvalue = T*(T+2)*((auto_corr**2/(T-np.arange(1,len(auto_corr)+1)))[:max_m]).cumsum()
    pvalue = 1 - chi2.cdf(lvalue, np.arange(1,len(lvalue)+1))
    return lvalue, pvalue

同じ結果が得られていることを確認。

l_Q_func, p_Q_func = Q_func(tpx_return,max_m=16)
l_sm, p_sm = sm.stats.diagnostic.acorr_ljungbox(tpx_return, lags=16)
((l_Q_func-l_sm)**2).mean(), ((p_Q_func-p_sm)**2).mean()

(0.0, 7.824090399073146e-34)

$m=8$のケースを細かく見てみると、

T = len(tpx_return)
auto_corr = sm.tsa.stattools.acf(tpx_return)[1:]
lvalue = T*(T+2)*((auto_corr**2/(T-np.arange(1,len(auto_corr)+1)))[:8]).sum()
print(lvalue)

8.604732778577853

1-chi2.cdf(lvalue,8)

0.37672860496603844

ということで、帰無仮説を棄却することはかなり難しいことが分かった。

初投稿！
本当はサーバレスまで一つの記事に入れたかったけど間に合わなかった・・・。
ということで今回はスクレイピング編になります。

やりたいこと

定期的に情報が更新されるwebページを自動でスクレイピングしたい！

目標

Yahoo!天気（東京）のデータを6時間おきに取得。

方法

Python + Scrapy + AWSlambda + CroudWatchEventsあたりでいけそう・・・？

とりあえずやってみる

まずはスクレイピングから

以下手順でクローリング、スクレイピング部分を作成。

1. Scrapyインストール
2. Scrapy　projectを作成
3. spiderの作成
4. 実行

1. Scrapyインストール

$ python3 -V
Python 3.7.4

$ pip3 install scrapy
...
Successfully installed

$ scrapy version
Scrapy 1.8.0

2. Scrapy projectを作成

コマンドを入力した階層にプロジェクトのフォルダが作成されます。

$ scrapy startproject yahoo_weather_crawl
New Scrapy project 'yahoo_weather_crawl'

$ ls
yahoo_weather_crawl

今回はyahoo天気のこの部分を取得してみます。

発表日時、日付、天気、気温、降水確率を拾ってみます。

Scrapyはコマンドラインシェルがあり、コマンドを入力して取得対象がちゃんと取れているか確認することが可能なので、一旦それで確認しつつ進めてみます。

取得対象をxpathで指定します。
xpathはgoogle chromeのデベロッパーツール(F12押すと出るやつ)から簡単に取得することができます。

今回取得した発表日時のxpathは以下
//*[@id="week"]/p

これをresponceから抜いてみます。

# scrapy shellの起動
$ scrapy shell https://weather.yahoo.co.jp/weather/jp/13/4410.html

>>> announcement_date = response.xpath('//*[@id="week"]/p/text()').extract_first()
>>> announcement_date
'2019年11月29日  18時00分発表'

text()を指定すると、本文のみを取得することが可能です。
詳しくは、参考文献参照。

とりあえず日時はとれたので、他も同様に取得していきましょう。

他の情報はtableタグの中にあるので、一度tableの中身を全て取得します。

>>> table = response.xpath('//*[@id="yjw_week"]/table')

これで、id="yjw_week"のテーブルタグ内の要素が取得できました。
ここから各要素を取得していきます。

# 日付
>>> date = table.xpath('//tr[1]/td[2]/small/text()').extract_first()
>>> date
'12月1日'

# 天気
>>> weather = table.xpath('//tr[2]/td[2]/small/text()').extract_first()
>>> weather
'曇時々晴'

# 気温
>>> temperature = table.xpath('//tr[3]/td[2]/small/font/text()').extract()
>>> temperature
['14', '5']

# 降水確率
>>> rainy_percent = table.xpath('//tr[4]/td[2]/small/text()').extract_first()
>>> rainy_percent
'20'

これでそれぞれの取得方法がわかったので、
Spider(処理のメイン部分）を作成していきます。

3. spiderの作成

先ほど作成したプロジェクトフォルダの構成は以下のようになっています。

.
├── scrapy.cfg
└── yahoo_weather_crawl
    ├── __init__.py
    ├── __pycache__
    ├── items.py
    ├── middlewares.py
    ├── pipelines.py
    ├── settings.py
    └── spiders
        ├── __init__.py
        └── __pycache__

まずは取得するitemsを定義しておきます。

items.py

import scrapy

class YahooWeatherCrawlItem(scrapy.Item):
    announcement_date = scrapy.Field()  # 発表日時
    date = scrapy.Field()               # 日付
    weather = scrapy.Field()            # 天気
    temperature = scrapy.Field()        # 気温
    rainy_percent = scrapy.Field()      # 降水確率

次に、spiderの本体をspidersフォルダ内に作成します。

spider/weather_spider.py

# -*- coding: utf-8 -*-
import scrapy
from yahoo_weather_crawl.items import YahooWeatherCrawlItem

# spider
class YahooWeatherSpider(scrapy.Spider):

    name = "yahoo_weather_crawler"
    allowed_domains = ['weather.yahoo.co.jp']
    start_urls = ["https://weather.yahoo.co.jp/weather/jp/13/4410.html"]

    # レスポンスに対する抽出処理
    def parse(self, response):
        # 発表日時
        yield YahooWeatherCrawlItem(announcement_date = response.xpath('//*[@id="week"]/p/text()').extract_first())
        table = response.xpath('//*[@id="yjw_week"]/table')

        # 日付ループ
        for day in range(2, 7):

            yield YahooWeatherCrawlItem(
                # データ抽出
                date=table.xpath('//tr[1]/td[%d]/small/text()' % day).extract_first(),
                weather=table.xpath('//tr[2]/td[%d]/small/text()' % day).extract_first(),
                temperature=table.xpath('//tr[3]/td[%d]/small/font/text()' % day).extract(),
                rainy_percent=table.xpath('//tr[4]/td[%d]/small/text()' % day).extract_first(),
                )

4. いざ実行！

scrapy crawl yahoo_weather_crawler

2019-12-01 20:17:21 [scrapy.core.scraper] DEBUG: Scraped from <200 https://weather.yahoo.co.jp/weather/jp/13/4410.html>
{'announcement_date': '2019年12月1日  17時00分発表'}
2019-12-01 20:17:21 [scrapy.core.scraper] DEBUG: Scraped from <200 https://weather.yahoo.co.jp/weather/jp/13/4410.html>
{'date': '12月3日',
 'rainy_percent': '10',
 'temperature': ['17', '10'],
 'weather': '晴れ'}
2019-12-01 20:17:21 [scrapy.core.scraper] DEBUG: Scraped from <200 https://weather.yahoo.co.jp/weather/jp/13/4410.html>
{'date': '12月4日',
 'rainy_percent': '0',
 'temperature': ['15', '4'],
 'weather': '晴れ'}
2019-12-01 20:17:21 [scrapy.core.scraper] DEBUG: Scraped from <200 https://weather.yahoo.co.jp/weather/jp/13/4410.html>
{'date': '12月5日',
 'rainy_percent': '0',
 'temperature': ['14', '4'],
 'weather': '晴時々曇'}
2019-12-01 20:17:21 [scrapy.core.scraper] DEBUG: Scraped from <200 https://weather.yahoo.co.jp/weather/jp/13/4410.html>
{'date': '12月6日',
 'rainy_percent': '10',
 'temperature': ['11', '4'],
 'weather': '曇り'}
2019-12-01 20:17:21 [scrapy.core.scraper] DEBUG: Scraped from <200 https://weather.yahoo.co.jp/weather/jp/13/4410.html>
{'date': '12月7日',
 'rainy_percent': '30',
 'temperature': ['9', '3'],
 'weather': '曇り'}

上手く取れてそうですね！
せっかくなので、ファイルに出力してみましょう。

ファイルに出力する際はデフォルトだと日本語が文字化けしてしまうため、
settings.pyにエンコードの設定を加えておきます。

settings.py

FEED_EXPORT_ENCODING='utf-8'

$ scrapy crawl yahoo_weather_crawler -o weather_data.json
...

weather_data.json

[
{"announcement_date": "2019年12月1日  17時00分発表"},
{"date": "12月3日", "weather": "晴れ", "temperature": ["17", "10"], "rainy_percent": "10"},
{"date": "12月4日", "weather": "晴れ", "temperature": ["15", "4"], "rainy_percent": "0"},
{"date": "12月5日", "weather": "晴時々曇", "temperature": ["14", "4"], "rainy_percent": "0"},
{"date": "12月6日", "weather": "曇り", "temperature": ["11", "4"], "rainy_percent": "10"},
{"date": "12月7日", "weather": "曇り", "temperature": ["9", "3"], "rainy_percent": "30"}
]

出力できました！

次回はこの処理とAWSを組み合わせてサーバレスで動かしてみようと思います。

参考文献

Scrapy 1.8 documentation
https://doc.scrapy.org/en/latest/index.html
10分で理解する Scrapy
https://qiita.com/Chanmoro/items/f4df85eb73b18d902739
ScrapyによるWebスクレイピング
https://qiita.com/Amtkxa/items/4c1172c932264ae941b4

とりあえず、pythonでロジスティック回帰を行いたい人向けです。パラメータはいじりません。
使用するデータはこちら（https://gist.github.com/tijptjik/9408623）
＊一番下に、全てをまとめたコードがあります。

ロジスティック回帰用のモジュールをインポート

sklearn.linear_modelの中から、LogisticRegressionだけをインポートします。

from sklearn.linear_model import LogisticRegression

データを分割するモジュールをインポート

sklearn.model_selectionの中から、train_test_splitだけをインポートします。

from sklearn.model_selection import train_test_split

行列を扱うモジュールをインポート

numpyをnpという名前で使用できる状態でインポートします。

import numpy as np

csvを扱うモジュールをインポート

pandasをpdという名前で使用できる状態でインポートします。

import pandas as pd

グラフを描くモジュールをインポート

import matplotlib.pyplot as plt

平均平方二乗誤差を求めるモジュールをインポート

from sklearn.metrics import mean_squared_error

csvを読み込む

df(データフレーム)に、iris.csvを読み込む。

df=pd.read_csv('wine_type.csv')

*('iris.csv')はカレントディレクトリからcsvファイルへのアクセスです。
pythonをデスクトップで実行し、Desktop>Documentsにcsvがある場合は、

df=pd.read_csv('Desktop/Documents/wine.csv')

などとなります。（Linux）

データを訓練用、テスト用に分ける

訓練：学習＝6:4にします。

df_train, df_test = train_test_split(df, test_size=0.4)

データを表示するとこんな感じです。

df_train=
     wine_type  alcohol  malic_acid   ash  alcalinity_of_ash  magnesium  total_phenols  flavanoids  nonflavanoid_phenols  proanthocyanins  color_intensity   hue  OD280/OD315_of_diluted_wines  proline
106          2    12.25        1.73  2.12               19.0         80           1.65        2.03                  0.37             1.63             3.40  1.00                          3.17      510
157          3    12.45        3.03  2.64               27.0         97           1.90        0.58                  0.63             1.14             7.50  0.67                          1.73      880
75           2    11.66        1.88  1.92               16.0         97           1.61        1.57                  0.34             1.15             3.80  1.23                          2.14      428
142          3    13.52        3.17  2.72               23.5         97           1.55        0.52                  0.50             0.55             4.35  0.89                          2.06      520
83           2    13.05        3.86  2.32               22.5         85           1.65        1.59                  0.61             1.62             4.80  0.84                          2.01      515
..         ...      ...         ...   ...                ...        ...            ...         ...                   ...              ...              ...   ...                           ...      ...
117          2    12.42        1.61  2.19               22.5        108           2.00        2.09                  0.34             1.61             2.06  1.06                          2.96      345
129          2    12.04        4.30  2.38               22.0         80           2.10        1.75                  0.42             1.35             2.60  0.79                          2.57      580
60           2    12.33        1.10  2.28               16.0        101           2.05        1.09                  0.63             0.41             3.27  1.25                          1.67      680
25           1    13.05        2.05  3.22               25.0        124           2.63        2.68                  0.47             1.92             3.58  1.13                          3.20      830
41           1    13.41        3.84  2.12               18.8         90           2.45        2.68                  0.27             1.48             4.28  0.91                          3.00     1035

[106 rows x 14 columns]

説明変数と目的変数を分ける

xには分析に使いたい列を挿入します。（説明変数）
yには分析の結果の列を挿入します。（目的変数）
今回は、'color_intensity'と'alcohol'から'proline'を予測します。

x_train = df_train[['color_intensity','alcohol']]
x_test  = df_test[['color_intensity','alcohol']]

y_train = df_train[['proline']]
y_test  = df_test[['proline']]

空モデルを作る

lr = LogisticRegression()

回帰の学習を行う

fit（説明変数、目的変数）

で学習を行い、上で作ったモデルlrに学習した結果が保持されます。

lr.fit(x_train, y_train)

回帰を行う

predict（回帰分析を行うデータ）

で回帰を行い、y_predに代入します。

y_pred = lr.predict(x_test)

グラフに表示してみる

plt.scatter(x軸、y軸)で散布図を作ることができます。
今回は、データのインデックスと予測値、データのインデックスと実際の値をグラフに表示し比べてみます。

plt.scatter(y_test.index, y_pred)
plt.scatter(y_test.index, y_test)
plt.xlabel("index")
plt.ylabel("proline")
plt.show()

青色が実際の値、オレンジが予測値というわけです。
全然予測できてない笑

最後に、平均平方二乗誤差を求めます。

mean_squared_error(y_test,y_pred)　#111299.75ですって笑

パラメータをいじるともっと精度が上がると思います。

以下コピペ用コード

from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.metrics import mean_squared_error
df=pd.read_csv('wine.csv')
df_train, df_test = train_test_split(df, test_size=0.4)
x_train = df_train[['color_intensity','alcohol']]
x_test  = df_test[['color_intensity','alcohol']]

y_train = df_train[['proline']]
y_test  = df_test[['proline']]
print(y_train)
lr = LogisticRegression()
lr.fit(x_train, y_train)
y_pred = lr.predict(x_test)


plt.scatter(y_test.index, y_pred)
plt.scatter(y_test.index, y_test)
plt.xlabel("index")
plt.ylabel("proline")
plt.show()

print(mean_squared_error(y_test, y_pred))