25/10/10 進捗報告
Pushbulletでスマホに通知を送る
pushbullet.py をプロジェクトに置いて import すれば、そのまま使えます。
使い方
1) まずはトークンの扱い（おすすめ） セキュリティのため、コードに直書きせず環境変数で渡すのが安全です。
- macOS / Linux:
export PUSHBULLET_TOKEN="o.xxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
- Windows (PowerShell):
setx PUSHBULLET_TOKEN "o.xxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
# 新しいターミナルを開くと反映されます
2) Python から送信（最短）
from pushbullet import notify
# note（通常通知）
notify(title="テスト通知", body="環境変数からトークンを読み込んで送信します。")
# link（URL付き通知）
notify(title="記事", body="あとで読む", url="https://example.com")
3) 明示的にトークンを渡す（直書きも可だが非推奨）
from pushbullet import PushbulletNotifier
pb = PushbulletNotifier(api_token="o.xxxxxxxxxxxxxxxxxxxxxxxxxxxxx") # 直書き例
pb.notify("ビルド完了", "アーティファクトが準備できました。")
4) オプション
# 特定デバイス宛やチャンネル宛にも対応
notify(
title="限定通知",
body="この通知は特定デバイス/チャンネル宛です",
device_iden="ujpah72o0sjAoRtnM0jc", # 任意
channel_tag="my-channel", # 任意
)
# タイムアウトやリトライ回数の調整
pb = PushbulletNotifier(timeout=10.0, max_retries=2)
pb.notify("調整版", "タイムアウトとリトライを設定")
注: あなたが提供してくれた API トークンはそのまま使えますが、セキュリティ上は環境変数で管理する方法を強くおすすめします。リポジトリにコミットしないよう注意してください。
Gas Surface Density
以下のコードを使用して Gas Surface Density を求めた。
import pandas as pd
import numpy as np
import minemine
from minemine import para_conv, to_kpc_array, h
import time
from pushbullet import PushbulletNotifier
pb = PushbulletNotifier(api_token="o.qUD51cNBVnDCegPorisKdlhgXrt3vMUm")
# 複数の半径を指定
radius_list = [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]
def process_one_subhalo(i, subhalo_pos, x, y, z, mass, scale_factor, radius_kpc):
cx, cy, cz = subhalo_pos
dx, dy, dz = para_conv(x, y, z, cx, cy, cz, 50.0)
r = np.sqrt(dx**2 + dy**2 + dz**2)
mask_r01 = r < 0.1
if not np.any(mask_r01):
return 0.0
r = r[mask_r01]
_mass = mass[mask_r01] * 1e10 / h
r_kpc = to_kpc_array(r, scale_factor)
mask_kpc = r_kpc < radius_kpc
if not np.any(mask_kpc):
return 0.0
total_mass = _mass[mask_kpc].sum()
return total_mass / (np.pi * radius_kpc**2)
for mask in [4, 3, 2, 1]:
print(f"\n=== mask {mask} ===")
for snap in range(20, 5, -1):
print(f" --- snap {snap} ---")
t0 = time.time()
scale_factor = minemine.scale_factor[snap]
print(f" loading star coordinates ...")
x, y, z = np.load(f"./dataset_v2Cvisc100pi/raw/fiducial/all.fiducial.{snap}.PartType0.Coordinates.npy").T
mass = np.load(f"./dataset_v2Cvisc100pi/raw/fiducial/all.fiducial.{snap}.PartType0.Masses.npy")
data = pd.read_feather(f"./VirialRadius/catalog_history_mask{mask}/fiducial/mask{mask}_catalog.fiducial.subhalo.{snap}.feather")
subhalo_positions = np.stack(data["SubhaloPos"].to_numpy())
print(f" {len(subhalo_positions)} subhalos to process")
for radius_kpc in radius_list:
surface_density_list = []
for i, subhalo_pos in enumerate(subhalo_positions):
if i % 100 == 0 or i == len(subhalo_positions) - 1:
print(f" processing subhalo {i+1}/{len(subhalo_positions)} (radius={radius_kpc} kpc) ...")
try:
sd = process_one_subhalo(i, subhalo_pos, x, y, z, mass, scale_factor, radius_kpc)
except Exception as e:
print(f" [ERROR] subhalo {i}: {e}")
sd = np.nan
surface_density_list.append(sd)
# 半径ごとに別の列を追加
col_name = f"GasSurfaceDensityIn{int(radius_kpc):1d}kpc"
data[col_name] = surface_density_list
save_path = f"./VirialRadius/catalog_history_mask{mask}/fiducial/mask{mask}_catalog.fiducial.subhalo.{snap}.feather"
print(f" saving to {save_path}")
data.to_feather(save_path)
t1 = time.time()
print(f" done (elapsed: {t1-t0:.1f} s)")
pb.notify("042_surdace_denisty_in_Xkpc_high_gas.py", f"Mask:{mask} / Snap:{snap}")
print("==== ALL DONE ====")
pushbulletから完了するごとに通知がきた。
結構時間がかかっていることがわかる。2,3日ぐらいかかった。以下がログである。