月: 2023年8月

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の使い方を記事にしてないことに気づいたことがこの記事の発端でした。)