カテゴリー: プログラミング

はじめに

ちょっと重い記事が続いてたので今回は小ネタです。

二つの区間データ、要するに数値の区間であれば最小値と最大値のペア、時刻の区間であれば開始時間と終了時間のデータが二つあったときに、それらの区間に重複があるかどうかをサクッと判定するアルゴリズムを紹介します。昔どこかでみた覚えがあるのですが、久々に実装で必要になったときにちょっとだけ悩んだのでそのメモです。

結論

結果だけ知りたい人向けに先に結論だけ書いときます。

区間[s1, e1]と[s2, e2]に重複があるかどうかは次の2つの方法で判定できます。

方法1: $(s1 \leq e2) \land (s2 \leq e1)$ ならばその2区間には重複があります。

方法2: $\max(s1, s2) \leq \max(e1, e2)$ ならばその2区間には重複があります。

両方の方法のメリットデメリットですが、方法1の方が計算速度が速いです。大小比較2個と論理演算ですからmax関数呼び出すより有利ですね。ただ、方法2の方は、応用として、二つの区間に重複があった場合、$\max(e1, e2) – \max(s1, s2)$で重複区間の長さを算出できます。(値が負になったら重複無し。)

(max関数がそんな極端に遅いってこともなく、わざわざ時間測定しない限り体感することもない速度差ではあるのであくまでも参考にどうぞ。)

方法1の導出

これだけで終えてしまうと記事量としてあまりに少ないのと、結論だけ書いてるとすぐ忘れそうなので、真面目に導出を説明しておきます。まずは方法1の方からです。

まず、二つの区間の相対的な位置関係は次の6パターンが存在します。(説明の単純化のため。s1,s2,e1,e2は全て異なる値とします。)

1. 区間1全体が区間2より小さい。つまり、 $s1 < e1 < s2 < e2$.
2. 区間1全体が区間2より大きい。つまり、 $s2 < e2 < s1 < e1$.
3. 区間1が区間2に内包される。つまり、$s2 < s1 < e1 < e2$.
4. 区間1が区間2を内包する。つまり、$s1 < s2 < e2 < e1$.
5. 区間1の後半と区間2の前半が重なる。つまり、$s1 < s2 < e1 < e2$.
6. 区間1の前半と区間2の後半が重なる。つまり、$s2 < s1 < e2 < e1$.

上記の6パターンのうち、1,2 が区間に重複がなく、3,4,5,6は区間に重複があります。

ここで、「そうか！3,4,5,6のどれかを満たすことを確認すればいいんだ！」と判断して4パターンの論理式を書いてorで繋ぐ、としないことがコツです。

区間に重複がないパターンの方が2種類しかなくて判定が単純なんですよね。そして、区間の定義から$s1<e1$と$s2<e2$はわかってるので、あと確認するのはe1とs2, e2とs1の代償関係だけなんですよ。

ということで、 $(e1 < s2) \lor (e2 < s1)$ であれば二つの区間に重複はないと言えます。

これの否定を考えると、ドモルガンの法則から方法1としてあげた式が導かれます。

方法2の導出

続いて方法2の導出です。

これは、先ほど挙げた3,4,5,6の4パターンの不等式を眺めると見つかる法則性なのですが、左の2辺はs1,かs2で、右の2辺はe1, e2なんですよね。

これを素直に数式に落とすと、$\max(s1, s2) \leq \max(e1, e2)$ となります。等号が成立するのはただ1点だけが重複する場合。

そして、区間が重複する場合は、$\max(s1, s2)$は重複区間の開始点であり、$\max(e1, e2)$は重複区間の終了点なので、この2項の差を取ると重複部分の長さも得られます。

まとめ

そんなに難しい計算ではなく、覚えてさえればサクッと実装できる式ではありますが、重複のパターンって4種類あるよなぁと考え始めてしまうと意外に手間取ります。

結果を丸暗記しなくても、区間が重複しないパターンは2個しかなくてそれを否定したら簡単だってことを頭の片隅にでも置いといてもらえると幸いです。

エントロピーや相互情報量の記事が続いていますが、今回の記事で計算の実装方法を紹介して一旦区切りとします。

エントロピーも相互情報量も数式はそこまで難しくないので、numpyで定義通り計算しても良いのですが、エントロピー関係はSciPyに、相互情報量はscikit-learnに用意されているので今回の記事ではそれを使っていきます。

計算対象のデータは、[“a1”, “a2”, “a1”, “a1”, “a2”] みたいにローデータの一覧で保有している場合もあれば、”a1″が3個で”a2″が2個のようにカウント済みのものがある場合もあると思うのでそれぞれ説明していきます。

エントロピーの計算

まず一番基本的なエントロピーの計算からです。これは、scipy.stats.entropy メソッドを使います。
参考: scipy.stats.entropy — SciPy v1.11.3 Manual

基本的な引数はpkなので、確率の一覧を渡すのが想定されていますが、和が1でないなら1になるように正規化してくれるのでサンプルがある場合は個数を渡しても大丈夫です。また、base引数で対数関数の底を指定でき、デフォルトが$e$なので、情報理論で使う場合は$2$を指定しましょう。

やってみます。

import numpy as np  # データ作りに利用
import pandas as pd  # データ作りに利用
from scipy.stats import entropy


pk = np.array([1/2, 1/3, 1/6])  # 確率の一覧が得られた場合。
print(entropy(pk, base=2))
# 1.459147917027245

count_list = np.array([3, 2, 1])  # データの個数の場合
print(entropy(count_list, base=2))
# 1.4591479170272446

# カウント前のデータの一覧がある場合
data_sr = pd.Series(["a1", "a1", "a1", "a2", "a2", "a3"])
# value_counts()で数えあげたものをentropyに渡す
print(entropy(data_sr.value_counts(), base=2))
# 1.4591479170272446

結合エントロピーの計算

次は結合エントロピーです。エントロピーを単純に2次元に拡張したやつですね。(条件付きエントロピーではないので注意してください、

例えば次のような例を考えましょうか。

	b1	b2
a1	4	1
a2	2	3

結合エントロピーの場合はですね、元のカウントデータを2次元から1次元に並び替えて渡します。

matrix_data = np.array([[4, 1], [2, 3]])
print(matrix_data)
"""
[[4 1]
 [2 3]]
"""

# ravel か flattenで1次元化して計算する
print(entropy(matrix_data.ravel(), base=2))
# 1.8464393446710157

# 標本データがある場合
df = pd.DataFrame({
    "A": ["a1", "a1", "a1", "a1", "a1", "a2", "a2", "a2", "a2", "a2"],
    "B": ["b1", "b1", "b1", "b1", "b2", "b1", "b1", "b2", "b2", "b2"],
})

# カウントしたデータを使う
print(df.groupby(["A", "B"]).size())
"""
A   B
a1  b1    4
    b2    1
a2  b1    2
    b2    3
dtype: int64
"""

print(entropy(df.groupby(['A', 'B']).size(), base=2))
# 1.8464393446710157

条件付きエントロピー

次は条件付きエントロピーです。残念なことなのですが、メジャーなライブラリでは条件付きエントロピー専用の関数は提供されていません。

そこで、$H(A|B) = H(A, B) – H(B)$などのエントロピー間の関係式を使って計算することになります。相互情報量も含めて、$H(A|B) = H(A) – I(A; B)$などで計算してもいいのですが、SciPyで完結できるので最初の式のほうが良いでしょう。

先ほどの表データをサンプルとします。$H(B)$については、表データを縦に足し合わせてBだけのカウントデータを作って計算します。

data_B = matrix_data.sum(axis=0)
print(data_B)
# [6 4]

# H(B)の計算
entropy_B = entropy(data_B, base=2)
print(entropy_B)
# 0.9709505944546688

# H(A, B)の計算
joint_entropy = entropy(matrix_data.ravel(), base=2)
print(joint_entropy)
# 1.8464393446710157

# H(A|B) = H(A, B) - H(B)
conditional_entropy_A_given_B = joint_entropy - entropy_B
print(conditional_entropy_A_given_B)
# 0.8754887502163469

# 標本データがある場合
entropy_B = entropy(df["B"].value_counts(), base=2)
joint_entropy = entropy(df.groupby(["A", "B"]).size(), base=2)

conditional_entropy_A_given_B = joint_entropy - entropy_B
print(conditional_entropy_A_given_B)
# 0.8754887502163469

以上で、3種類のエントロピーが計算できました。

相互情報量

最後に相互情報量の計算方法です。

$I(A; B) =H(A)-H(A|B)$など複数の表現方法があるので、ここまでに計算してきた値から算出することもできます。

entropy_A = entropy(df["A"].value_counts(), base=2)
print(entropy(df["A"].value_counts(), base=2) - conditional_entropy_A_given_B)
# 0.12451124978365313

ただ、scikit-learnに専用のメソッドがあるのでこちらの使い方も見ておきましょう。
参考: sklearn.metrics.mutual_info_score — scikit-learn 0.18.2 documentation

引数は、mutual_info_score(labels_true, labels_pred, contingency=None)
となっており、標本データを受け取るのが標準的な使い方で、その第一,第二引数をNoneにしてcontingency引数にカウントデータを渡すこともできます。(contingencyがNoneでない場合はこれが優先されて、先の二つの引数が無視されます。)

1点注意しないといけないのは、entropyと違って対数の底が指定できず、自然対数に固定されてしまうことです。底を$2$で考えたい場合は、$\ln{x}/\ln{2} = \log_2{x}$を使って変換が必要です。

from sklearn.metrics import mutual_info_score


# np.log(2)で割ることを忘れない
# カウントした表データがある場合
print(mutual_info_score(None, None, contingency=matrix_data)/np.log(2))
# 0.12451124978365345

# 標本データがある場合
print(mutual_info_score(df["A"], df["B"])/np.log(2))
# 0.12451124978365345

計算の都合上超軽微な誤差がありますが、それ以外は想定通りの値が得られていますね。

以上で、相互情報量も計算できるようになりました。

自分はあまり使ってこなかったのですが、構造化したデータを保存するのに利用されるYAMLというデータ形式が存在します。いかにもPythonと相性が良さそうなフォーマットです。
参考: YAML – Wikipedia

最近、そこそこ大きいYAMLファイルの総チェックを行う必要があったので、Pythonでスクリプトを書いてチャチャっとやりました。その時のYAMLの読み込み方法のメモと、ついでに逆にデータをYAMLに書き出す方法を残しておきます。

サンプルとしては、次の内容が保存した sample.yaml というファイルを使います。

name: John Doe
age: 30
address:
  city: Tokyo
  country: Japan

必要なライブラリはいくつかありますがPyYAMLが代表的です。インストール時と利用時で名前が違うので注意してください。
参考: PyYAML · PyPI

# インストールコマンド
$ pip install PyYAML

# Pythonでインポートする時
>>> import yaml

それではやっていきましょう。まずはYAMLファイルの読み込みからです。

YAMLを読み込むメソッドは、load(), load_all(), safe_load(), safe_load_all() と4つも用意されています。普通はsafe_load()を使えば良いです。safeがついてない2メソッドにはセキュリティ面のリスクもあるので使用を避けることを推奨します。

import yaml


with open("sample.yaml", "r") as f:
    data = yaml.safe_load(f)

print(data)
# {'name': 'John Doe', 'age': 30, 'address': {'city': 'Tokyo', 'country': 'Japan'}}

簡単ですね。気をつけないといけないのは、safe_load()にファイルパスを渡しても読み込んではくれず、openでファイルを開いて、それを渡すって点です。

YAMLファイルは、”—“というセパレーターを使うことで単一ファイルの中に複数のYAMLオブジェクトを記述できますが、そのようなファイルを読み込むときはsafe_load_all()の方を使うので注意してください。読み込んだ結果はジェネレーターで返ってきます。

ファイルにするが手間だったのでセパレーターで区切られたYAML文字列を直接作ってやって見ました。

yaml_data = yaml.safe_load_all("""name: John Doe
age: 30
---
name: Jane Smith
age: 25
---
name: Bob Johnson
age: 40""".strip())

for y in yaml_data:
    print(y)
"""
{'name': 'John Doe', 'age': 30}
{'name': 'Jane Smith', 'age': 25}
{'name': 'Bob Johnson', 'age': 40}
"""

以上が、YAMLの読み込みです。

次は書き出しです。最初のコード例の、data変数をYAML形式ファイルに保存してみます。

保存には yaml.safe_dump() というメソッドを使います。これにPythonのオブジェクトを渡すとYAMLのテキストに変換してくれます。

print(yaml.safe_dump(data))
"""
address:
  city: Tokyo
  country: Japan
age: 30
name: John Doe
"""

ただし、普通はファイルに保存することになると思いますので、次のように使うことになると思います。

with open("output.yaml", "w") as f:
    yaml.safe_dump(data, f)

これで先ほどのテキストが、”output.yaml”ファイルに保存されます。

YAMLには一般的なブロックスタイルと、JSONに近い見た目のフロースタイルという形式がありますが、defalut_flow_style という引数をTrueにすると、フロースタイルで出力できます。ただ、これは使うことあまり無いかな。

また、インデントの数はindentという引数で指定できます。デフォルトは2ですね。

以上がPythonを用いたYAMLファイルの読み書き方法でした。

Jupyterには便利なマジックコマンド(%や%%を付けて呼び出すアレです)がたくさんありますが、あれを自作する方法を紹介します。

ドキュメントは IPythonのドキュメントのこちらを参照します。
参考: Defining custom magics — IPython 8.14.0 documentation

簡単な方法は、register_line_magic, register_cell_magic, register_line_cell_magic の3種のデコレーターをマジックコマンドとして使いたい関数につけることです。

register_line_magicはその行の文字列を格納する引数を1個だけ、register_cell_magicとregister_line_cell_magicは、マジックコマンドと同じ行の文字列を格納する引数と、セル内の文字列を格納する引数の2個をもちます。

ざっと、受け取った文字列をprintするだけのコマンドを作ってみましょう。3種類それぞれのサンプルです。

from IPython.core.magic import register_line_magic
from IPython.core.magic import register_cell_magic
from IPython.core.magic import register_line_cell_magic


@register_line_magic
def line_magic(line):
    print(line)


@register_cell_magic
def cell_magic(line, cell):
    print(f"line: {line}")
    print(f"cell:\n{cell}")


@register_line_cell_magic
def line_cell_magic(line, cell=None):
    print(f"line: {line}")
    if cell:
        print(f"cell:\n{cell}")

順番に使ってみます。

%line_magic ラインマジックテスト
print()
# 以下出力
# ラインマジックテスト

%%cell_magic セルマジックと同じ行のテキスト
セルマジック内のテキスト
その２行目

＃ 以下出力。
"""
line: セルマジックと同じ行のテキスト
cell:
セルマジック内のテキスト
その２行目
"""

%line_cell_magic ラインマジックとして動作させた場合

# 以下出力
# line: ラインマジックとして動作させた場合

%%line_cell_magic セルマジックとして動作させた場合。
セルの中身

# 以下出力
"""
line: セルマジックとして動作させた場合。
cell:
セルの中身
"""

めっちゃ簡単ですね。

最初のマジックコマンドを定義したコードをPythonファイルとして保存して、import可能なディレクトリに置いておくと、インポートして使うこともできる様になります。例えば、 my_magic.py というファイル名で保存しておけば次の様に使えます。

import my_magic


%line_magic 読み込んだモジュールのマジックコマンドが使える
# 読み込んだモジュールのマジックコマンドが使える

my_magic.line_magic("普通の関数としても呼び出せる")
# 普通の関数としても呼び出せる

さて、通常マジックコマンドをライブラリ等から読み込んで使う場合、この様にimport するのではなく、%load_ext して使うことが多いと思います。これは、先ほどあげたドキュメントのページでベストプラクティスとされているのがその方式だからです。@register_* のデコレーターで直接登録する上記の方法は推奨されてないんですね。

その代わりにどうするかというと、 load_ipython_extension というメソッドを持つpythonファイルを作り、このメソッドの中で定義した関数たちを register_magic_function でマジックコマンドへ登録していきます。

引数は順に、登録する関数本体、コマンドの種類(省略したら’line’)、マジックコマンドとして呼び出す時の名前(省略したら元の関数名)です。

例えば、 my_ext.py というファイルを作りその中を次の様にします。

def load_ipython_extension(ipython):
    ipython.register_magic_function(
        line,
        magic_kind='line',
        magic_name='line_magic'
    )

    ipython.register_magic_function(
        cell,
        magic_kind='cell',
        magic_name='cell_magic'
    )

    ipython.register_magic_function(
        line_cell,
        magic_kind='line_cell',
        magic_name='line_cell_magic'
    )


def line(line):
    print(line)


def cell(line, cell):
    print(f"line: {line}")
    print(f"cell:\n{cell}")


def line_cell(line, cell=None):
    print(f"line: {line}")
    if cell:
        print(f"cell:\n{cell}")

各メソッドそれぞれにはデコレーターはつきません。

この様なファイルを用意すると、load_ext で読み込んだ時に、load_ipython_extension が実行されて、その中でマジックコマンドの登録が行われます。結果、次の様に使えます。

%load_ext my_ext


%line_magic ロードしたマジックコマンドが使えた
# ロードしたマジックコマンドが使えた

先ほどのimportした場合との挙動の違いとしては、これは明示的にマジックコマンドの読み込みだけを行っているので、各メソッドはインポートはされておらず、個々のメソッドの、line, cell, line_cell は名前空間に登録されてないということです。(マジックコマンドとして登録された、line_magic, cell_magic, line_cell_magic の名前でなら通常のメソッドと同じ様に使うことも可能です)

以上が簡単なマジックコマンドの作り方になります。

もう結構前なのですが、メモ化というテクニックを紹介しました。
参考: pythonの関数をメモ化する

これは@functools.lru_cacheというデコレーターを使って、関数の戻り値を記録しておいて何度も同じ関数を実行するコストを削減するのでしたね。計算コストが削減される代わりに、結果を保存しておく分メモリを消費します。

僕はこれを結構使ってたのですが、最近、これをクラスのメソッドに対して利用しているとメモリリークを引き起こすことがあるという気になる情報を得ました。ブログで紹介しちゃった責任もあるので、今回はその問題について調べました。

この問題は何箇所かで指摘されていて、一例を挙げるとこのissueなどがあります。
参考: functools.lru_cache keeps objects alive forever · Issue #64058 · python/cpython

こっちのYoutubeでも話されていますね。
参考: don’t lru_cache methods! (intermediate) anthony explains #382 – YouTube

具体的に説明していくために超単純なクラスを作って実験していきましょう。
まず、そのまま返すメソッドを持ってるだけのシンプルクラスを作ります。そして、このクラスがメモリから解放されたことを確認できる様に、デストラクターが呼び出されたらメッセージを表示する様にしておきます。これをインスタンス化して関数を1回使って、delで破棄します。

class sample1:
    def __init__(self, name):
        self.name = name

    def __del__(self):
        print(f"インスタンス: {self.name} を破棄しました。")

    def identity(self, x):
        return x


a = sample1("a")
print(a.identity(5))
# 5
del a
# インスタンス: a を破棄しました。

デストラクターがちゃんと呼び出されていますね。

これが、メソッドがメモ化されていたらどうなるのかやってみます。

from functools import lru_cache


class sample2:
    def __init__(self, name):
        self.name = name

    def __del__(self):
        print(f"インスタンス: {self.name} を破棄しました。")

    @lru_cache(maxsize=None)
    def identity(self, x):
        return x


b = sample2("b")
print(b.identity(5))
# 5
del b
# 何も表示されない。

今度はガベージコレクターが動きませんでしたね。これは、変数bを削除したことによって変数bからの参照は消えたのですが、メソッドのidentity の一つ目の引数がそのインスタンス自体をとっていて、これを含めてキャッシュしているので、キャッシュがインスタンス自身への参照を保存しているためガベージコレクションの対象にならなかったのです。そのため、インスタンスbが確保していたメモリは解放されず、占拠されたままになります。

ちなみに、循環参照の状態なので、明示的にガベージコレクターを動かすと消えます。

import gc


gc.collect()
# インスタンス: b を破棄しました。

ちなみに、最初のキャッシュが発生した時に循環参照が生まれているのでメモ化したメソッドを一回も使わなかったら普通に消えます。

c = sample2("c")
del c
# インスタンス: c を破棄しました。

以上の様な問題があるので、クラスメソッドで lru_casheを使う時は気をつけて使うことをお勧めします。

とはいえ、最近のMacBookくらいのメモリ量であれば、インスタンスが何個か過剰に残ったとしてそれでメモリが枯渇する様なことはないんじゃないかなとも思います。仮にメモリがピンチになる様な使い方をしていたとしても、maxsizeを適切に設定してメモリサイズを押さえておくとか、明示的にgc/collect()動かすとかの対応が取れるかと。

僕としては、メモリが解放されないことよりも、デストラクターが動かなくてそこに仕込んだ後始末形の処理が動かないのが気になりましたね。

これもずっと前に書いたつもりでいたら書いてなかったので書きます。プログラムを開発していると、ログを実装することがあります。Jupyterでインタラクティブにやる時はprintで十分なのですが、機械学習モデルや何かしらのロジックを実装して本番環境で稼働させるなら何かしらログが残る仕組みは必須です。そして僕は横着してJupyterのファイルをそのままバッチとして使い始めたりもするのでJupyterでもロギングを使うことがあります。

その様な時、Pythonにはloggingモジュールという大変便利なモジュールが標準で用意されています。
参考: logging — Python 用ロギング機能 — Python 3.11.4 ドキュメント

このloggingライブラについては必ずドキュメントを読んでから使うことをお勧めします。というのも、このモジュールにはアンチパターンが多く、例えばルートロガーをそのまま使うなどの、適当に書いたらそう書いちゃう実装がリスクが大きいからです。

最近はドキュメントでも上の方で、logging.getLogger(name)を使えってちゃんと書かれる様になりましたね。それでは、使い方書いていきます。

getLoggerを用いたロガーの取得

loggingモジュールでは最初にロガーを取得します。これをやることでそのモジュールごとに独立したロガーが生成され、設定をいじってもシステム全体の他のロガーの設定に影響が出ない様にできます。ロガーには名前をつけるのですが、__name__という変数を使うと自動的にモジュール名が入るので良いでしょう。スクリプト本体の場合は__name__の値は__main__になります。

import logging


logger = logging.getLogger(__name__)

これでloggerが生成されましたが、初めて使った人がそのまま利用しようとすると、デバッグログやinfoのログが出ないことに気づくと思います。

# debugと info は何も出力しない。
logger.debug("デバック")
logger.info("インフォ")

# warning/error/critical は標準エラー出力に出力
logger.warning("ワーニング")
# WARNING:__main__:ワーニング
logger.error("エラー")
# ERROR:__main__:エラー
logger.critical("クリティカル")
# CRITICAL:__main__:クリティカル

ざっくり言うと初期設定ではこういう動きに設定されているということです。
もっと詳しく説明すると、この時点ではloggerにはログの出力先(これをhandlerという)が設定されておらず、この様な場合は警告以上の重要度のメッセージを「lastResort」っていう最終手段のハンドラが出力してくれているという状況になっています。ハンドラが設定されてなくても重要なメッセージはユーザーに伝えようと言うモジュールの思いやりです。

さて、上記のままではせっかくのデバッグメッセージ等が揉み消されるし出力も簡素すぎるのでここから色々設定していきます。

ハンドラーの作成と追加

まず、先ほどの例では出力先を制御しているハンドラが設定されていないので、こちらの追加方法を説明します。

これはファイルに書き出したい場合や標準エラー出力に出力したい場合などに備えて複数のクラスがあるのでそれをインスタンス化し、 addHandlerでロガーに設定します。複数設定も可能です。

ハンドラたちのドキュメントはこちら。次の様なイメージで設定します。

参考: logging.handlers — ロギングハンドラ — Python 3.11.4 ドキュメント

# ファイルに書き出すハンドラ
file_handler = logging.FileHandler('{ファイルパス}')
logger.addHandler(file_handler)

# 標準エラー出力に書き出すハンドラ
stream_handler = logging.StreamHandler()
logger.addHandler(stream_handler)

ログレベルの設定

ログにはログレベルという概念があって、ロガーと、出力先であるハンドラそれぞれがどのレベル以上のログを出力するかという持っています。両方を満たさなければ出力されません。例えば、ロガーはINFO以上、ハンドラがERROR以上を出力する設定なら、ERROR以上のログだけが出力されるという様な仕組みです。

ハンドラが個別に設定を持っていることで、標準エラー出力とログファイルに別々の設定を提供したりできますし、ロガー自体が設定を持っているので、開発中と本番運用中で一括して設定を変えることなどもできます。

ログレベルはこちらの通り で、
NOTSET(0) → DEBUG(10) → INFO(20) → WARNING(30) → ERROR(40) → CRITICAL(50)
の順です。

それぞれ setLevelメソッドを持っているのでそれで設定します。上記のレベルは数値でもいいし、モジュールが定数を持ってます。

次の様なイメージで設定します。

# ロガーのログレベル設定
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
# ハンドラのログレベル設定
handler.setLevel(logging.INFO)
logger.addHandler(stream_handler)

ハンドラごとに設定を変えれば、DEBUGはファイルだけ、INFO以上はファイルと標準出力両方といった設定ができます。

フォーマッターの使い方

ここまでで、ログの出力有無の制御ができたので、最後に出力内容を設定します。これはフォーマッターという仕組みで実現します。

設定によって詳細な時刻や、ログレベル、モジュール名、行番号などを情報に加えることができます。例によってこれもハンドラごとに設定できるので、標準エラー出力には時刻はいらないけど、ファイルには時刻も残したいって出しわけなどができますね。

フォーマッター中で使える変数はこちらの表にまとまっています。
参考: LogRecord 属性

実際に使うのはこの辺かな。

• asctime ・・・ 時刻を人間が読める書式にしたもの。
• levelname・・・DEBUGとかERRORなどのログレベル。
• message・・・ロギングの時に渡されたメッセージ本文。
• name・・・ロガーの名前。
• funcName・・・ロギングの呼び出しを含む関数の名前。
• lineno・・・ソース行番号

使い方のイメージとしては、次の様に %スタイルでフォーマットを作り、setFormatterでセットします。

formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)

上記の例は%スタイルでこれはFormatterのコンストラクタのstyle引数がデフォルトの’%’だからこうなってるのですが、style引数に ‘{‘ を渡せば中括弧スタイルでも定義できます。個人的はそちらの方が好きです。

# 以下の二つは同じ
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
formatter = logging.Formatter("{asctime} - {levelname} - {message}", style="{")

一通り動かす

以上をまとめて、書いておきます。これをベースに好みの形でいじっていけば良いロガーがが作れると思います。

import logging


logger = logging.getLogger(__name__)
logger.handlers.clear()
logger.setLevel(logging.DEBUG)

formatter = logging.Formatter("{asctime} - {levelname} - {message}", style="{")

file_handler = logging.FileHandler("sample.log")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)

stream_handler = logging.StreamHandler()
stream_handler.setLevel(logging.INFO)
stream_handler.setFormatter(formatter)
logger.addHandler(stream_handler)

# 使い方。debugはファイルのみ、それ以外はファイルと標準エラー出力に表示。
logger.debug("debug")
logger.info("info")
logger.warning("warning")
logger.error("error")
logger.critical("critical")

何度も書くのは面倒なので自分は上記のコードに色々工夫を入れたものモジュール化して使い回しています。(訳あってこの間、久しぶりに書き直すことになりました。その時loggerの使い方を記事にしてないことに気づいたことがこの記事の発端でした。)

今まで知らなかったメソッド(expanduser)を見つけたのでその紹介を兼ねて小ネタの紹介です。

※注: Windowsでも動作する様に書いていますが、手元にWindowsのPython環境を持っていないのでこの記事の内容はWindowsでは未検証です。

今回の記事でやるのはタイトルの通り、ユーザーのホームディレクトリの取得です。

ホームディレクトリの下に作業ディレクトリやログディレクトリを作ってプログラムで使うと言う状況はそこそこある事だと思っています。書き捨てのスクリプトであればホームディレクトリのパスを直書きしてしまえばいいのですが、どこかに公開する場合や自分で使いまわしたい場合はそれでは不便が起きます。僕の場合でもAWS環境と、複数台のMacを使っているので、ホームディレクトリを直書きしてしまうと使い回しに修正が必要です。そのため、ホームディレクトリは自動的に取得することが望ましく、この記事のテクニックが必要になります。

いくつか方法があるので順番に紹介していきます。ちなみに、サンプルコードではユーザー名がyutaroだと仮定します。そのため結果として得られるホームディレクトリは ‘/Users/yutaro’ です。

環境変数を使う方法

一番簡単なのは環境変数から取得してしまうことですね。Mac/LinuxではHOME、WindowsではHOMEPATHやUSERPROFILEという環境変数に格納されています。次の様にするとどの環境でも動く様に書くことができます。orで繋いで最初に見つかったものを採用しているだけです。なお、絶対Windowsでは使わないよ、って場合はもうHOMEだけ見たら良いです。

import os


home_directory = (
    os.environ.get('HOME') or
    os.environ.get('HOMEPATH') or
    os.environ.get('USERPROFILE')
)
print(home_directory)
# /Users/yutaro

上記の方法は特別なメソッドの知識とかいらないのですがご覧の通り記述量が多いので他の方法をつづいて紹介していきます。以降の方法の方がお勧めです。

osモジュールのメソッドを使う方法

(環境変数の取得もosモジュールでやってるので表題困りましたが、) osモジュールにはexpanduserという専用のメソッドを持っています。これは与えられたパスの先頭の ~ (チルダ) や ~user という文字列をホームディレクトリのパスに置換してくれるものです。

参考: os.path — 共通のパス名操作 — Python 3.11.4 ドキュメント

いくつか実験したのでコードを載せておきます。 ‘~user’って文字列でもいい様なことがドキュメントに書いてますが、これはuserじゃなくてOSのユーザー名を入れないといけない様です。環境ごとの記述の差異を吸収してほしいという今回の記事の主題的にはちょっとダメですね。素直に~だけ使いましょう。

# チルダをHOMEディレクトリに書き換えてくれる。
print(os.path.expanduser("~"))
# /Users/yutaro

# ドキュメントの ~user は ~userという文字列を指してるものではないらしく、~userはそのまま。
print(os.path.expanduser("~user"))
# ~user

# ~ユーザー名、僕の場合は ~yutaro は置換される。
print(os.path.expanduser("~yutaro"))
# /Users/yutaro

# ~の後ろにそのまま文字列が続いていたら置換されない。
print(os.path.expanduser("~work"))
# ~work

# ~の後ろに小ディレクトリを書いておくことができる。
print(os.path.expanduser("~/folder/subfolder"))
# /Users/yutaro/folder/subfolder

# ~が先頭でない場合は置換されない。
print(os.path.expanduser("/~/subfolder"))
# /~/subfolder

最後にpathlibって別のモジュールを使った方法もあるのでそれも紹介します。

pathlibモジュールを使った方法

こちらは、home()っていうズバリなメソッドを持ってます。
参考: pathlib — オブジェクト指向のファイルシステムパス — Python 3.11.4 ドキュメント

これが一番いいかな。(ネックは、pathlibの存在を忘れがちなことくらいか。)

from pathlib import Path


print(Path.home())
# /Users/yutaro

# 実はデータ型が違う
print(type(Path.home()))
# <class 'pathlib.PosixPath'>

print(type(os.path.expanduser("~")))
# <class 'str'>

内部的には、実質的に os.path.expanduser(“~”) と同等の処理をやってるみたいですね。ただし、上のサンプルコードの後半で見ている通り、Path.home()はPathオブジェクトを返してくれているので、その後のパス操作がより直感的になります。

話のついでにもう一つ紹介しておくと、 pathlibもexpanduserを持ってます。ただ、これは文字列を受け付けてくれないのでosのそれより使い勝手が悪いです。(個人の感想)

# Path.expanduser は文字列を受け付けないので例外が発生する。
try:
    print(Path.expanduser("~"))
except Exception as e:
    print(e)
# 'str' object has no attribute '_drv'


from pathlib import PosixPath


# PosixPath型に変換して使う
print(Path.expanduser(PosixPath(("~"))))
# /Users/yutaro

まとめ

以上で、Pythonでホームディレクトリを取得する方法をまとめてきました。ハードコーディングをやめてこれらの方法を使うとコードの使い回しがよりやりやすくなると思います。

個人的にはイチオシは pathlibの Path.home() ですかね。ただ自分でも忘れてosで環境変数取りに行くことが多そうですが。

データ分析を行う際、データセットに欠損値（NaNやNoneなど）が含まれていることはよくあります。これらの欠損値をどのように取り扱うかは、分析結果に大きな影響を及ぼすため、重要なステップとなります。

Pandasを使う場合、これらの欠損値に対応する一番簡易的な方法はfillna()を使った定数による補完です。もしくはdropna()を使ってそのデータを消す事もあるかもしれませんね。

しかし、状況によってはすべての欠損値を単一の値で補完するのは、データの分布や傾向を歪める可能性があります。また、時系列データなどでは欠損値が発生したレコードをdropできない事もあるかもしれません。周期がずれたりしますからね。

そこで使えるのが、掲題のinterpolate()メソッドです。これを使うとここの欠損値の前後の値を使った補完など多様な補完ができます。特に引数を指定しなければ線型補完です。

参考: pandas.Series.interpolate — pandas 2.0.3 documentation

まず基本的な使い方を見ていきましょう。欠損値を含む単純なSeriesデータを用意してやってみます。

import pandas as pd
import numpy as np


s = pd.Series([0, 2, np.nan, np.nan, np.nan, 10])
print(s)
"""
0     0.0
1     2.0
2     NaN
3     NaN
4     NaN
5    10.0
dtype: float64
"""

print(s.interpolate())
"""
0     0.0
1     2.0
2     4.0
3     6.0
4     8.0
5    10.0
dtype: float64
"""

はい、等差数列で補完してくれていますね。単純な例なのでとても自然な結果になっています。補完の方法はmethod引数で指定でき、デフォルトは”linear”です。
他には、次の様な値が使えます。
– linear ・・・ 線型補完。これがデフォルト値。
– ffill または pad ・・・ 前の値。
– bfill, backfill ・・・ 後ろの値。
– nearest ・・・ 最も近い値。
– polynomial ・・・多項式補完 (orderで次数を指定する)。
– spline ・・・スプライン補完 (orderで次数を指定する)。

他にも indexの値を考慮してくれるindexやvalue、時系列で使いやすそうなtimeなどもありますね。(実際に指定できる文字列は他にもあり、その種類はかなり多いです。公式ドキュメントの参照をお勧めします。)

いくつかやってみます。

s = pd.Series([0, 2, np.nan, np.nan, np.nan, 10, 9, np.nan, np.nan, 6])
print(s.values)
# [ 0.  2. nan nan nan 10.  9. nan nan  6.]

# 線型補完
print(s.interpolate(method='linear').values)
# [ 0.  2.  4.  6.  8. 10.  9.  8.  7.  6.]

# Pad。ffillも同じ結果。前の値を使う。
print(s.interpolate(method='pad').values)
# [ 0.  2.  2.  2.  2. 10.  9.  9.  9.  6.]

# bfill。backfill。後ろの値を使う。
print(s.interpolate(method='bfill').values)
# [ 0.  2. 10. 10. 10. 10.  9.  6.  6.  6.]

# 最も近い値。
print(s.interpolate(method='nearest').values)
# [ 0.  2.  2.  2. 10. 10.  9.  9.  6.  6.]

# 多項式補完
print(s.interpolate(method='polynomial', order=2).values)
# [ 0.          2.          4.43062201  7.29186603  9.50717703 10.
#  9.          7.88516746  6.88516746  6.        ]

# スプライン補完
print(s.interpolate(method='spline', order=2).values)
# [ 0.          2.          5.30198447  7.26402071  8.60569456 10.
#  9.          8.90854185  7.76876618  6.        ]

polynomial と spline の違いがわかりにくいと思いますが、ざっくり説明すると次の様になります。

polynomial：多項式補間では、欠損値を補完するために多項式関数が使用されます。指定した次数の多項式がデータにフィットされ、その多項式関数に基づいて欠損値が補完されます。しかし、データの点が多い場合や次数が高い場合、多項式補間はデータに過剰にフィットする（オーバーフィッティングする）傾向があります。

spline：スプライン補間では、データセット全体を通じて一つの関数が使用されるのではなく、各データ点の間に別々の多項式（通常は3次）がフィットされます。これらの多項式は、データ点において連続性と滑らかさを保つように選ばれます。スプライン補間は、より滑らかな曲線を生成し、オーバーフィッティングを避けるために通常は低次の多項式（たとえば3次）が使用されます。

この二つの選択に限った話ではありませんが、適切な補間方法を選択する際には、データの性質と分析の目的を考慮することが重要です。

何となくpandasのドキュメントを眺めていたら見つけた小ネタの紹介です。
この記事を読むと、pandasのSeriesをもっと手軽に値で絞り込める様になります。
参考: pandas.Series.loc — pandas 2.0.3 documentation

pandasのlocといえば、自分としてはDataFrameで使うことが多いプロパティですが、もちろんSeriesにも実装されています。そして、これを使うとindexの値に従って要素を絞り込むことができます。

今回見つけたのは、このlocにcallable、要するにメソッドが渡せるってことです。渡したメソッドにSeriesの値が渡され、その結果がTrueのものに絞り込まれます。

こんな感じで使えます。例えば値が3以上の要素だけに絞り込む例です。

import pandas as pd

# Seriesを作成
s = pd.Series([1, 2, 3, 4, 5], index=['a', 'b', 'c', 'd', 'e'])

# 値が3以上の要素だけを絞り込む
s[lambda x: x>=3]
"""
c    3
d    4
e    5
dtype: int64
"""

上記の例だとメリットがわかりにくいかと思います。と言うのもpandasに慣れてる人だったら次の様に書けることがわかってると思いますし、タイプ数も少なく可読性も高いからです。

s[s>=3]
"""
c    3
d    4
e    5
dtype: int64
"""

単純な大小比較や一致不一致ではなく、もっと複雑な判定を行うメソッドを適用するときなどは、このlocにメソッドを渡すやり方が便利に使えますね。

僕が個人的に気に入ったのは、この絞り込みが特定の変数に格納されていないSeriesについても使えると言うことです。
「特定の変数に格納されていないSeries」ってのは、例えばDataFrameのvalues_counts()メソッドなどを実行した結果として得られる値などの形で取得されます。

例えば、dfというDataFrameがあるとして、そのcolumn_name列の値の出現回数を数え、そのうち出現回数が10以上のものだけを取り出すとしましょう。

通常であれば、value_counts()の結果を何かの変数に格納して実行するか、もしくは2回value_counts()を実行する非効率に我慢するかして次の様に実装します。

# カウント結果を一度変数に格納する場合
count_sr = df["column_name"].value_counts()
count_sr[count_sr >= 10]

# value_countsを2回実行する場合
df["column_name"].value_counts()[df["column_name"].value_counts() >= 10]

これが、locにメソッドが渡せることを知っていると次の様に書けます。

# 無駄な変数も定義しないし、value_counts()の実行も1回でいい書き方
df["column_name"].value_counts()[lambda x: x>=10]

個人的に、「ある列の要素ごとに数を数えて、一定件数以上データがあったものだけ残す処理」ってのをやることが頻繁にあり、Seriesのfiterメソッドが値ではなくindexにしか使えないことを日々残念に思っていた自分にとってはめっちゃ嬉しいテクニックだったので紹介しました。

ちなみに、DataFrameのlocも同じ様にcallableを渡せます。こっちはあまり使い道が思いつかないですね。
参考: pandas.DataFrame.loc — pandas 2.0.3 documentation

タイトルの通りで、Jupyter notebookファイル (.ipynb ファイル)をモジュールとしてインポートする方法を紹介します。

僕は普段のコーディングをJupyterでやっているので、自分で使う汎用的なモジュールを作るときも一度Jupyterで作って.pyファイルに移行するという手順で作っていました。一回完成させて仕舞えばそれでいいのですが、作りかけのものを別のプログラムで使いたいとか、それに限らず、あるnotebookで定義した関数を別のnotebookで使いたい、ってときにこれまでセルの中身をコピペしてたのですが、実はnotebookファイルのままimportできるという噂を聞き試しました。

機械学習の前処理とか何度も同じコードを書いてるので使いまわせる様になると便利そうです。

使うのはこちらの importnb というライブラリです。
参考: importnb · PyPI

とりあえず試してみましょう。
importされる側のnotebookファイルを以下の内容で作ります。importできることの確認だけなので、適当なメソッドと定数が定義されているだけのファイルです。

ファイル名は sub_file.ipynb としました。

def foo():
    return "bar"


hoge = "hoge"

では、このファイルをimportしてみましょう。

importnb sub_file とかで済むと簡単なのですが、やや独特な記法でimportします。先ほどのsub_file.ipynb は閉じて新しいnotebookファイルで以下の様に書きます。

from importnb import imports


with imports("ipynb"):
    import sub_file

これでimportできました。
メソッド foo や 変数 hoge が使えます。

print(sub_file.foo())
# bar

print(sub_file.hoge)
# hoge

少し検証してみたのですが、このライブラリはimportされるnotebookの中身によっては注意して使う必要があります。というのも、notebookを import するときにimport されるnotebookの全てのセルが実行されるのです。なので、何かファイルを書き出す処理があればimportした時点でファイルを書き出ししますし、重い処理があれば時間かかりますし、外部APIを叩く処理が入っていたら外部APIを叩きます。

宣言されているメソッドやクラス、変数だけを持ってきて使えると言うわけでは無いのでimportするnotebookは慎重に選びましょう。

というか、その確認作業をするのであれば .py ファイルに書き出すとか必要なセルだけコピペして持ってくるといった対応をする方が早いことが多く、この importnb を使う場面ってかなり限られるなぁと言うのが自分の所感です。

頻繁に使っうMeCabを使った前処理とかワードクラウドクラウド作成とか汎用的なSQLとか自分が頻繁に使うメソッドや定数をまとめた神notebook集を用意したりするとまた活用の幅も出てくるかもしれませんね。

もう一点、検証時に気付いた注意点があります。これ、notebookのセルを全て実行するので、その中に一つでもエラーになるセルがあったらimportを失敗します。

そのエラーになるセルより先に実行されたセルの中身だけimportされるのではなく、何もimportされない結果になります。これも注意しましょう。