プログラミング/python の履歴(No.4)

前処理結果を dump する†

jupyter 上で計算した結果を引き継ぎつつ jupyter 外で処理を回すには、前処理で得られたデータ（プロファイリングしたい処理への入力）を pickle.dump しておき、外部で動かす処理ではそれを呼んで動作を行う

Jupyter セル内で実行

LANG:python
import pickle
payload = {
    "param1": param1,
    "param2": param2,
    "param3": param3,
}
with open("payload.pkl", "wb") as f:
    pickle.dump(payload, f, protocol=pickle.HIGHEST?PROTOCOL)

で（通常の python データであれば）書き出せる

読み込んで動かす .py ファイル†

do_profile.py

LANG:python
import pickle
import cProfile
import pstats

from my_module import func_to_profile

def main():
    with open("payload.pkl", "rb") as f:
        payload = pickle.load(f)

    prof = cProfile.Profile()
    prof.enable()
    refined = func_to_profile(**payload)
    prof.disable()

    prof.dump_stats("prof.out")

if __name__ == "__main__":
    main()

実行する†

LANG:console
$ python do_profile.py

解析する†

LANG:console
$ python -m pstats prof.out
Welcome to the profile statistics browser.
prof.out% strip        # strip directory names
prof.out% sort cumtime # sort by cumtime
prof.out% stats 40     # show first 40 items

これで、

my_module.py:1234(my_function)

みたいな関数ごとに

LANG:console
ncalls   tottime  percall  cumtime  percall filename:lineno(function)
呼び出し 自前処理 呼び出し 子も含む 呼び出し ファイル名:行番号(関数名)
回数     時間     あたり   全時間   あたり

のリストが出ます

特定の関数内から呼ばれている分だけを調べたければ、

LANG:console
prof.out% sort time
prof.out% callees my_module.py:1234(my_function)

などとします。

必要なモジュールを確実に見つけられるように検索パスを広げる†

LANG:python
from pathlib import Path
import sys

# 必要なモジュールを確実に見つけられるように検索パスを広げる
#
# from some_module.some_file import xxx は実質的に
# from root.**.some_module.some_file import xxx のように働くので、
# root の配下のどこかに some_module があれば確実に import できる
# しかし some_module と同列に置かれた ipynb では root が
# some_module になって、その下で some_module を探してエラーになる
# ここは root を十分上に持っていくための処理

# どれかが見つかった場所をルート扱いにする
MARKERS = ("pyproject.toml", ".git", ".vscode")
p = Path.cwd().resolve()
root = None
for parent in (p, *p.parents):
    if any((parent / m).exists() for m in MARKERS):
        root = parent
        break
if root is None:
    # 見つからなければさしあたり親ディレクトリにしておく
    root = Path.cwd().resolve().parents[0]

# 先頭がrootでなければ、既存のrootエントリを全部消してから先頭に入れる
print(f"Adding this folder to sys.path: {root}")
root_s = str(root)
if not sys.path or sys.path[0] != root_s:
    sys.path[:] = [p for p in sys.path if p != root_s]
    sys.path.insert(0, root_s)

オートリロード†

ソースファイルが日本語を含む可能性があるならこれをまず実行

LANG: python
import builtins

# デフォルトでファイルは UTF-8 で読むように open を置き換える
# 日本語コメント入りの .py ファイルを autoreload しようとして落ちるのを避けるため
if "_original_open" not in globals():
    _original_open = builtins.open

    def open_utf8_default(path, mode="r", *args, **kwargs):
        # binary モードや明示指定がある場合は素通し
        if "b" not in mode and "encoding" not in kwargs:
            kwargs["encoding"] = "utf-8"
        return _original_open(path, mode, *args, **kwargs)

    builtins.open = open_utf8_default

で、これが自動リロードの設定。

LANG: python
# 自動リロードを設定
%reload_ext autoreload
%autoreload 2

必要に応じて依存モジュールを pip install する†

LANG:python
ensure_package("trimesh")

みたいにしてモジュールがない場合に限って pip install できるようになる。

LANG:python
# 必要に応じて必要なモジュールをインストールする
import sys
import subprocess
import importlib.util

print(sys.executable)
print(f"sys.version: {sys.version}")

def _get_installed_version(dist_name: str) -> str | None:
    """
    Return installed distribution version (e.g., 'numpy' -> '1.26.4').
    Prefer importlib.metadata; fallback to pkg_resources for older Pythons.
    """
    try:
        # Python 3.8+
        from importlib.metadata import version, PackageNotFoundError  # type: ignore
        try:
            return version(dist_name)
        except PackageNotFoundError:
            return None
    except Exception:
        # Older Python or environment without importlib.metadata
        try:
            import pkg_resources  # type: ignore
            try:
                return pkg_resources.get_distribution(dist_name).version
            except Exception:
                return None
        except Exception:
            return None

def _get_module_dunder_version(import_name: str) -> str | None:
    """Try module.__version__ as a last resort."""
    try:
        import importlib
        m = importlib.import_module(import_name)
        v = getattr(m, "__version__", None)
        return v if isinstance(v, str) and v else None
    except Exception:
        return None

def ensure_package(import_name: str, dist_name: str | None = None, pip_args: list[str] | None = None) -> None:
    """
    Ensure a module is importable. If not, install it via pip.
    - import_name: the name used in `import ...`
    - dist_name  : the name used in `pip install ...` / metadata lookup (often same)
    """
    dist = dist_name or import_name

    if importlib.util.find_spec(import_name) is None:
        print(f"{import_name} not found. Installing {dist}...")
        args = [sys.executable, "-m", "pip", "install", dist]
        if pip_args:
            args.extend(pip_args)
        subprocess.check_call(args)
        # install後の表示（取れなければunknown）
        ver = _get_installed_version(dist) or _get_module_dunder_version(import_name) or "unknown"
        print(f"{import_name} installed. version={ver}")
    else:
        ver = _get_installed_version(dist) or _get_module_dunder_version(import_name) or "unknown"
        print(f"{import_name} already importable. version={ver}")

# ここに必要なパッケージを並べる
ensure_package("scipy")
ensure_package("trimesh")
ensure_package("embreex") # for raycast
ensure_package("fast_simplification", dist_name = "fast-simplification")