Panel+pyvisで木構造可視化

Panel+pyvisで木構造を可視化する。ほしいのは以下の機能。 - 木構造が表現できること。ノードとエッジが表現でき、矢印で方向を表現できること。 - 分析がしやすいこと。ノードの色や大きさで重要な箇所がわかりやすくできること。あるノードを選択した際に、そこまでのルートが一目でわかること。

起動方法

panel serve dashboard.py --static-dirs assets=./assets

コード

「あるノードを選択した際に、そこまでのルートが一目でわかること」についてはPyvisの元の機能にはなさそうであったため、出力後のHTMLにJSコードを埋め込むことで対応した。

実際にどんなグラフが出てくるかは以下を参照。

wooolwoool.hatenablog.com

import json, re
import panel as pn
from pyvis.network import Network
# from bokeh.models.widgets import DataTable, TableColumn, StringFormatter
import pandas as pd
import os

pn.extension()
old_graph_html_path = None
html_num = 0
result_hist = []


def build_pyvis_graph(tree_data):
    net = Network(height="600px", width="100%", directed=True)
    net.set_options("""
    {
    "nodes": {
        "shape": "dot",
        "font": {
        "size": 14,
        "face": "Tahoma"
        }
    },
    "edges": {
        "arrows": {
        "to": {
            "enabled": true
        }
        },
        "smooth": {
        "enabled": true,
        "type": "cubicBezier"
        }
    },
    "interaction": {
        "navigationButtons": true,
        "keyboard": true
    },
    "physics": {
        "enabled": false
    }
    }
    """)

    # 値の最大・最小を見つける(正規化のため)
    all_vals = []

    def collect_vals(node_data):
        if 'val' in node_data:
            all_vals.append(node_data['val'])
        for child in node_data.get('child', {}).values():
            collect_vals(child)

    for root_data in tree_data.values():
        collect_vals(root_data)

    min_val = min(all_vals) if all_vals else 0
    max_val = max(all_vals) if all_vals else 1
    val_range = max_val - min_val if max_val != min_val else 1

    # RGBグラデーション
    def get_color(val):
        ratio = (val - min_val) / val_range
        # ratio: 0.0 (薄い) → 1.0 (濃い)
        # 濃いCyan: rgb(0, 200, 200)
        # 薄いCyan: rgb(240, 255, 255)

        r = int(240 * (1 - ratio))        # 減らす → 0
        g = int(255 * (1 - 0.2 * ratio))  # 少しだけ暗く → 200
        b = int(255 * (1 - 0.2 * ratio))  # 同上

        return f"rgb({r},{g},{b})"

    def add_nodes(node_name, node_data, path=[]):
        val = node_data.get('val', 0)
        size = 15 + (val - min_val) / val_range * 20  # サイズ調整
        color = get_color(val)

        net.add_node(node_name, label=f"{node_name}\nval={val}", size=size, color=color)
        for child_name, child_data in node_data.get("child", {}).items():
            # child_val = child_data.get('val', 0)
            # width = 1 + (child_val - min_val) / val_range * 5  # エッジ太さ

            # まず子ノードを追加
            add_nodes(child_name, child_data, path + [node_name])

            # ノード追加後にエッジを追加(← ここが重要)
            # net.add_edge(node_name, child_name, width=width)
            net.add_edge(node_name, child_name, width=1)

    for root_name, root_data in tree_data.items():
        add_nodes(root_name, root_data)
    global html_num
    tmp_html_path = f"assets/graph{str(html_num)}.html"
    html_num += 1
    net.save_graph(tmp_html_path)


    # クリック時に親方向のルートを赤くするJS
    custom_js = """
    <script type="text/javascript">
    function highlightPath(network) {
        network.on("click", function (params) {
            const edges = network.body.data.edges;
            edges.update(edges.get().map(e => ({
                ...e,
                color: undefined,
                width: e.originalWidth || e.width
            })));

            if (params.nodes.length === 0) return;
            let target = params.nodes[0];

            while (true) {
                let incoming = edges.get().filter(e => e.to === target);
                if (incoming.length === 0) break;
                let parentEdge = incoming[0];
                parentEdge.originalWidth = parentEdge.width;
                // parentEdge.color = "red";
                parentEdge.width = 5;
                edges.update(parentEdge);
                target = parentEdge.from;
            }
        });
    }

    // ネットワークが準備できたら highlightPath を実行
    if (typeof network !== 'undefined') {
        highlightPath(network);
    } else {
        setTimeout(() => {
            if (typeof network !== 'undefined') highlightPath(network);
        }, 500);
    }
    </script>
    """

    # HTMLファイルにスクリプトを埋め込む
    with open(tmp_html_path, 'r', encoding='utf-8') as f:
        html = f.read()

    html = re.sub(r'</body>', custom_js + '</body>', html, flags=re.IGNORECASE)

    with open(tmp_html_path, 'w', encoding='utf-8') as f:
        f.write(html)
    return tmp_html_path

# グラフとテーブルの初期要素
graph_pane = pn.pane.HTML("<b>Submitしてグラフを表示</b>", height=420, sizing_mode="stretch_width")
result_table = pn.widgets.DataFrame(pd.DataFrame(columns=["Step", "Node", "Success"]), height=200, disabled=True)

# 入力フォーム
json_input = pn.widgets.TextAreaInput(placeholder="JSONを入力", height=150, sizing_mode="stretch_width")
submit_button = pn.widgets.Button(name="Submit", button_type="primary")
sample_submit_button = pn.widgets.Button(name="Sample", button_type="primary")

# Submitボタンクリック時の動作
def process_input(event):
    # try:
        print("Processing input...")
        graph_pane.object = "<b>Processing input...</b>"
        pn.state.cache.clear()
        os.makedirs("assets", exist_ok=True)
        global old_graph_html_path
        if old_graph_html_path and os.path.exists(old_graph_html_path):
            os.remove(old_graph_html_path)
        data = json.loads(json_input.value)
        tree = data["tree"]
        result = data["result"]

        # ノードグラフ更新
        graph_html_path = build_pyvis_graph(tree)
        graph_pane.object = f"""
            <iframe src="/assets/{os.path.basename(graph_html_path)}" width="100%" height="400px" frameborder="0"></iframe>
        """

        # テーブル更新
        result_hist.append(result)
        if len(result_hist) > 10:
            result_hist.pop(0)
        result_df = pd.DataFrame(result_hist, columns=["Step", "Node", "Success"])
        result_table.value = result_df
        old_graph_html_path = graph_html_path

    # except Exception as e:
    #     graph_pane.object = f"<pre>Error: {e}</pre>"

def process_sample_input(event):
    # ランダムなJSONを生成して入力欄にセット
    sample_data = {
        "tree": {
            "root": {
                "val": 10,
                "child": {
                    "child1": {"val": 5, "child": {
                        "grandchild1": {"val": 3, "child": {
                            "greatgrandchild1": {"val": 8, "child": {
                                "greatgreatgrandchild2": {"val": 6, "child": {}}
                            }},
                            "greatgrandchild": {"val": 4, "child": {}}
                        }},
                        "grandchild2": {"val": 2, "child": {
                            "greatgrandchild2": {"val": 1, "child": {}}
                        }}
                    }},
                    "child2": {"val": 15, "child": {
                        "grandchild3": {"val": 8, "child": {
                            "greatgrandchild4": {"val": 7, "child": {}},
                            "greatgrandchild5": {"val": 9, "child": {}}
                        }},
                        "grandchild4": {"val": 6, "child": {}}
                    }}
                }
            }
        },
        "result": [
            {"Step": 1, "Node": "root", "Success": True},
            {"Step": 2, "Node": "child1", "Success": False},
            {"Step": 3, "Node": "child2", "Success": True}
        ]
    }
    json_input.value = json.dumps(sample_data, indent=4)
    process_input(event)

submit_button.on_click(process_input)
sample_submit_button.on_click(process_sample_input)

# レイアウト
layout = pn.Row(
    pn.Column(
        pn.pane.Markdown("## ゲーム検証結果可視化"),
        json_input,
        pn.Row(
            submit_button,
            sample_submit_button,
        )
    ),
    pn.Column(
        pn.pane.Markdown("### ノードグラフ"),
        graph_pane,
        pn.pane.Markdown("### 結果テーブル"),
        result_table,
        sizing_mode="stretch_width"
    )
)

layout.servable()

入力するデータ

{
  "tree": {
    "Start": {
      "child": {
        "AA": {
          "child": {
                "AA12": {
                  "child": {},
                  "val": 13
                },
                "AA22": {
                  "child": {},
                  "val": 15
                }
              },
          "val": 10
        },
        "BB": {
          "child": {},
          "val": 20
        }
      },
      "val": 11
    }
  },
  "result": ["Start", "AA", false]
}

PythonでWindowsのGUIアプリを操作するコード

操作記録用。YAML形式で記録。

from pynput import mouse
import pygetwindow as gw
import yaml
import time

recorded = []
last_time = None  # 前回クリック時刻を保存

def on_click(x, y, button, pressed):
    global last_time

    if not pressed:
        return  # 押した瞬間だけ記録

    if button.name == 'right':
        print("右クリックで終了します")
        return False

    active_window = gw.getActiveWindow()
    if not active_window:
        print("アクティブウィンドウが取得できませんでした")
        return

    window_title = active_window.title
    rel_x = x - active_window.left
    rel_y = y - active_window.top

    current_time = time.time()
    if last_time is None:
        delay = 0.0
    else:
        delay = round(current_time - last_time, 3)
    last_time = current_time

    print(f"記録: {window_title} 相対({rel_x},{rel_y}) delay={delay}s")

    recorded.append({
        'window': window_title,
        'action': 'click',
        'pos': [rel_x, rel_y],
        'delay': delay
    })

def record_mouse_clicks():
    print("マウスクリックを記録します。右クリックで終了。")
    with mouse.Listener(on_click=on_click) as listener:
        listener.join()

    print("記録完了。YAMLに保存中...")
    with open('recorded.yaml', 'w', encoding='utf-8') as f:
        yaml.dump({'操作名': recorded}, f, allow_unicode=True)
    print("保存しました → recorded.yaml")

if __name__ == '__main__':
    record_mouse_clicks()

Replay用

import yaml
import time
import pyautogui
import pygetwindow as gw
import pywinauto


def restore_window(title: str):
    """指定タイトルのウィンドウを復元(最小化解除)"""
    try:
        app = pywinauto.Application().connect(title=title, timeout=3)
        dlg = app.window(title=title)
        if dlg.is_minimized():
            dlg.restore()
            time.sleep(1.0)  # 復元待ち
        return True
    except Exception as e:
        print(f"[!] ウィンドウ '{title}' の復元失敗: {e}")
        return False


def click_relative(window_title, rel_x, rel_y):
    """ウィンドウ左上からの相対座標でクリック"""
    windows = gw.getWindowsWithTitle(window_title)
    if not windows:
        print(f"[!] ウィンドウ '{window_title}' が見つかりません")
        return False

    win = windows[0]

    if win.isMinimized:
        print(f"[*] 最小化されたウィンドウ '{window_title}' を復元します")
        if not restore_window(window_title):
            return False
        windows = gw.getWindowsWithTitle(window_title)
        win = windows[0]

    abs_x = win.left + rel_x
    abs_y = win.top + rel_y
    print(f"[*] クリック: {window_title} の相対({rel_x}, {rel_y}) -> 絶対({abs_x}, {abs_y})")
    pyautogui.click(abs_x, abs_y)
    return True


def execute_operation(operation_name: str, yaml_path='record.yaml'):
    """指定された操作名の一連の処理を実行"""
    try:
        with open(yaml_path, encoding='utf-8') as f:
            data = yaml.safe_load(f)
    except Exception as e:
        print(f"[!] YAMLファイルの読み込み失敗: {e}")
        return False

    steps = data.get(operation_name)
    if not steps:
        print(f"[!] 操作 '{operation_name}' が定義されていません")
        return False

    print(f"[+] 操作 '{operation_name}' を実行開始")

    for i, step in enumerate(steps):
        action = step.get('action')
        window = step.get('window')
        pos = step.get('pos', [0, 0])
        delay = step.get('delay', 0.5)

        print(f"  - Step {i+1}: {action} on '{window}' at {pos}, delay={delay}s")

        if action == 'click':
            if not click_relative(window, pos[0], pos[1]):
                print(f"[!] Step {i+1} のクリックに失敗しました")
                return False

        time.sleep(delay)

    print(f"[+] 操作 '{operation_name}' を完了")
    return True


if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("使い方: python replay_engine.py 操作名")
    else:
        execute_operation(sys.argv[1])

PythonでExcelを2回目から高速に読む方法

以下のようなExcelデータを使用する際、毎回Excelを読みこんでいるとめちゃ時間がかかるので高速化した。

方法としては、読みこんだデータを.npyで保存しておくようにした。今回はデータが更新されるのでチェックサムで更新を確認して、更新された場合は.npyも更新するようにしている。

import pandas as pd
import numpy as np
import os
import hashlib
import json

def compute_checksum(file_path: str) -> str:
    """Excelファイルのチェックサム(SHA-256)を計算する"""
    hash_sha256 = hashlib.sha256()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_sha256.update(chunk)
    return hash_sha256.hexdigest()

def read_prices_from_sheets(file_path: str, sheet_names: list, step: int = 1, use_cache: bool = False) -> list:
    # キャッシュファイルとチェックサムファイルのパス
    cache_file = file_path.replace('.xlsx', '_cache.npy')
    checksum_file = file_path.replace('.xlsx', '_checksum.json')
    
    # チェックサムの読み込みとExcelファイルの更新確認
    current_checksum = compute_checksum(file_path)
    is_cache_valid = False
    
    if os.path.exists(checksum_file):
        with open(checksum_file, 'r') as f:
            cached_data = json.load(f)
            if cached_data.get("checksum") == current_checksum:
                is_cache_valid = True

    # キャッシュを使用する場合で、有効なキャッシュが存在する場合
    if use_cache and is_cache_valid and os.path.exists(cache_file):
        print(f"Loading data from cache: {cache_file}")
        all_data = np.load(cache_file, allow_pickle=True).item()
    else:
        print(f"Reading data from Excel: {file_path}")
        all_data = {}

        # Excelファイルから各シートの価格データを取得
        for sheet_name in pd.ExcelFile(file_path).sheet_names:
            df = pd.read_excel(file_path, sheet_name=sheet_name)
            all_data[sheet_name] = df.iloc[:, 1].tolist()  # 2列目が価格データ

        # キャッシュとして保存
        np.save(cache_file, all_data)
        with open(checksum_file, 'w') as f:
            json.dump({"checksum": current_checksum}, f)
        print(f"Data cached to: {cache_file}")

    # 指定したシートのデータのみ取得
    all_prices = []
    for sheet_name in sheet_names:
        if sheet_name in all_data:
            all_prices.extend(all_data[sheet_name][::step])

    return all_prices

ビットコインの年間利益計算スクリプト

ビットコインの年間利益計算スクリプト作成した。BitFlyerの取引レポートをそのまま読みこめるようにしてある。

間違っている点あればコメントで教えてください。

import pandas as pd

def calculate_annual_profit_average_method(csv_file):
    # CSVファイルの読み込み
    df = pd.read_csv(csv_file, parse_dates=['取引日時'])

    # 必要な列のみ取得
    df = df[['取引日時', '取引種別', '取引価格', '通貨1数量', '手数料']]
    df = df.sort_values('取引日時')  # 取引日時でソート

    # 年ごとのデータに分ける
    df['年'] = df['取引日時'].dt.year

    annual_profit = {}

    # 年ごとに計算
    total_quantity = 0  # 総購入数量
    total_cost = 0      # 総購入金額
    for year, group in df.groupby('年'):
        profit = 0          # 売却利益

        # まず、年内の購入取引の合計を計算
        for _, row in group.iterrows():
            if row['取引種別'] == '買い':
                price = float(row['取引価格'].replace(",", ""))
                quantity = abs(float(row['通貨1数量'].replace(",", "")))
                # 購入にかかったコスト
                total_cost += price * quantity
                # 購入数量
                total_quantity += quantity - abs(float(row['手数料']))

            if row['取引種別'] == '受取':
                quantity = abs(float(row['通貨1数量'].replace(",", "")))
                total_quantity += quantity

        # 年末の平均取得原価を計算
        if total_quantity > 0:
            avg_cost_price = total_cost / total_quantity
        else:
            avg_cost_price = 0

        # 次に売却取引を確認して利益を計算
        for _, row in group.iterrows():
            if row['取引種別'] == '売り':
                price = float(row['取引価格'].replace(",", ""))
                quantity = abs(float(row['通貨1数量'].replace(",", "")))
                # 売却利益 = (売却価格 - 平均取得原価) * 売却数量
                profit += (price - avg_cost_price) * (quantity - abs(float(row['手数料'])))
                total_quantity -= quantity
                total_quantity -= abs(float(row['手数料']))

        # 年ごとの利益を辞書に追加
        annual_profit[year] = int(profit)

        total_cost = avg_cost_price * total_quantity

    return annual_profit

def calculate_annual_profit_moving_average(csv_file):
    # CSVファイルの読み込み
    df = pd.read_csv(csv_file, parse_dates=['取引日時'])

    # 必要な列のみ取得
    df = df[['取引日時', '取引種別', '取引価格', '通貨1数量', '手数料']]
    df = df.sort_values('取引日時')  # 取引日時でソート

    # 年ごとのデータに分ける
    df['年'] = df['取引日時'].dt.year

    annual_profit = {}

    # 年ごとに計算
    total_quantity = 0  # 保有残高
    total_cost = 0      # 総購入金額

    for year, group in df.groupby('年'):
        profit = 0          # 売却利益
        
        for _, row in group.iterrows():
            if row['取引種別'] == '買い' or row['取引種別'] == '売り' or row['取引種別'] == '受取':
                price = float(row['取引価格'].replace(",", ""))
                quantity = abs(float(row['通貨1数量'].replace(",", "")))

                if row['取引種別'] == '買い':
                    # 購入時は総購入金額と保有数量を更新
                    total_cost += price * quantity
                    total_quantity += quantity
                    total_quantity -= abs(float(row['手数料']))
                    # 平均取得原価の更新
                    avg_cost_price = total_cost / total_quantity

                elif row['取引種別'] == '売り':
                    # 売却利益 = (売却価格 - 平均取得原価) * 売却数量
                    profit += (price - avg_cost_price) * quantity

                    # 保有数量と総購入金額の更新
                    total_cost -= (avg_cost_price * quantity)
                    total_quantity -= quantity
                    total_quantity -= abs(float(row['手数料']))
                    
                elif row['取引種別'] == '受取':
                    total_quantity += quantity
                    avg_cost_price = total_cost / total_quantity

        # 年ごとの利益を辞書に追加
        annual_profit[year] = int(profit)

    return annual_profit

# 使用例
csv_file = 'TradeHistory.csv'

annual_profit = calculate_annual_profit_average_method(csv_file)
print("総平均法")
print(" 年間利益:", annual_profit)

annual_profit = calculate_annual_profit_moving_average(csv_file)
print("移動平均法")
print(" 年間利益:", annual_profit)

CでSEGVなどで異常終了した際にバックトレースログを残す方法

開発初期あたりのとりあえず正常系のみ書いて動かしてみるときなどで、SEGVが発生した場合にバックトレースログを出力したいときのコード。

シグナルハンドラでSEGVなど異常終了のシグナルをハンドリングしてログ出力する。

#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#ifdef DEBUG
#include <signal.h>
#include <execinfo.h>
#endif

// $ gcc -DDEBUG -g -o out main.c
// $ ./out > bt.log 2>&1

// DEBUGが定義されている場合のみシグナルハンドラを設定
#ifdef DEBUG
// グローバル変数に実行ファイルのパスを保存
char exe_path[1024];

void signal_handler(int sig, siginfo_t *info, void *ucontext) {
    void *buffer[30];
    int nptrs;
    char **symbols;
    char msg[256];
    int len;

    // バックトレースを取得
    nptrs = backtrace(buffer, sizeof(buffer) / sizeof(void *));
    symbols = backtrace_symbols(buffer, nptrs);
    if (symbols == NULL) {
        const char *msg_fail = "Failed to get backtrace symbols.\n";
        write(STDERR_FILENO, msg_fail, strlen(msg_fail));
        _exit(EXIT_FAILURE);
    }

    // シグナル情報を標準エラー出力に出力
    len = snprintf(msg, sizeof(msg), "Received signal %d (%s)\n", sig, strsignal(sig));
    write(STDERR_FILENO, msg, len);

    if (sig == SIGSEGV || sig == SIGBUS) {
        len = snprintf(msg, sizeof(msg), "Fault address: %p\n", info->si_addr);
        write(STDERR_FILENO, msg, len);
    }

    // 実行ファイルのパスを出力
    len = snprintf(msg, sizeof(msg), "Executable Path: %s\n", exe_path);
    write(STDERR_FILENO, msg, len);

    // バックトレースを標準エラー出力に出力
    write(STDERR_FILENO, "Backtrace:\n", strlen("Backtrace:\n"));
    for (int i = 0; i < nptrs; i++) {
        // シンボル情報の形式: ./myprogram(main+0x15b) [0x4006d3]
        // ここからアドレスを抽出するために、[] 内のアドレス部分を取得します。
        char *start = strchr(symbols[i], '[');
        char *end = strchr(symbols[i], ']');
        if (start && end && end > start + 1) {
            *end = '\0'; // ']' を終端に
            start++; // '[' をスキップ
            len = snprintf(msg, sizeof(msg), "  [%d] %s - %s:%s\n", i, symbols[i], exe_path, start);
            write(STDERR_FILENO, msg, len);
        } else {
            // アドレスが取得できない場合はそのまま出力
            len = snprintf(msg, sizeof(msg), "  [%d] %s\n", i, symbols[i]);
            write(STDERR_FILENO, msg, len);
        }
    }

    free(symbols);

    // デフォルトハンドラに戻す
    signal(sig, SIG_DFL);
    raise(sig);
}

void setup_signal_handler() {
    struct sigaction sa;
    memset(&sa, 0, sizeof(sa));
    sa.sa_sigaction = signal_handler;
    sa.sa_flags = SA_SIGINFO | SA_RESTART;

    // 捕捉するシグナルのリスト
    int signals[] = {SIGSEGV, SIGABRT, SIGFPE, SIGILL, SIGBUS, SIGTERM, SIGINT, SIGQUIT, SIGHUP};
    int num_signals = sizeof(signals) / sizeof(signals[0]);

    char msg[256];
    int len;
    for(int i = 0; i < num_signals; i++) {
        if (sigaction(signals[i], &sa, NULL) == -1) {
            perror("sigaction");
            exit(EXIT_FAILURE);
        } else {
            // シグナルハンドラが設定されたことを標準エラー出力に出力
            len = snprintf(msg, sizeof(msg), "Signal handler set for signal %d (%s)\n", signals[i], strsignal(signals[i]));
            write(STDERR_FILENO, msg, len);
        }
    }
}

#endif // DEBUG

int main(int argc, char *argv[]) {
#ifdef DEBUG
    // 実行ファイルのパスを取得
    ssize_t len = readlink("/proc/self/exe", exe_path, sizeof(exe_path) - 1);
    if (len != -1) {
        exe_path[len] = '\0';
    } else {
        strcpy(exe_path, "unknown");
    }

    // シグナルハンドラの設定(DEBUGが定義されている場合のみ有効)
    setup_signal_handler();
#endif
    // プログラムのメイン処理
    // ここではテスト用に意図的にセグメンテーションフォルトを発生させます
    // 本番コードでは削除してください
    fprintf(stderr, "DEBUGモード: セグメンテーションフォルトを発生させます。\n");
    fflush(stderr); // 出力をフラッシュ
    int *p = NULL;
    *p = 42;  // ここでSIGSEGVが発生

    fprintf(stdout, "プログラムが正常に終了しました。\n");
    return 0;
}