"""sm-remote コマンドラインエントリポイント（研究室サーバでの計算支援）．
手元（Windows の PowerShell / Git Bash）から 1 コマンドで以下を行う．
- push : コードを研究室サーバへ送る（tar over ssh．rsync 不要）
- run : サーバの tmux セッションでコマンドを起動する（接続が切れても継続）
- status : 実行中ジョブ・最近のジョブログ・GPU・負荷を表示する
- logs : ジョブログを表示する（-f で追従）
- attach : ジョブの tmux セッションに接続する
- kill : ジョブ（tmux セッション）を終了する
- pull : output/ のサブディレクトリを持ち帰る
接続先は --host / --remote-dir，または環境変数 SM_REMOTE_HOST / SM_REMOTE_DIR で上書きできる．
"""
from __future__ import annotations
import argparse
import codecs
import contextlib
import os
import socket
import subprocess
import sys
import tarfile
import time
from collections.abc import Callable, Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from sm_tias_lightsim.remote.commands import (
EXIT_NO_SESSION,
EXIT_NOT_FOUND,
EXIT_REMOTE_DIR_MISSING,
EXIT_SESSION_EXISTS,
REMOTE_LOG_DIR,
build_attach_check_command,
build_attach_command,
build_job_script,
build_kill_command,
build_list_files_command,
build_logs_command,
build_pull_command,
build_push_command,
build_remove_files_command,
build_run_command,
build_ssh_argv,
build_status_command,
normalize_output_subpath,
parse_env_assignment,
validate_gpu_spec,
validate_job_name,
)
from sm_tias_lightsim.remote.files import (
EXCLUDED_DIR_NAMES,
PRUNE_DIRS,
PushEntry,
collect_push_entries,
compute_prune_targets,
find_project_root,
find_yaml_with_windows_paths,
)
from sm_tias_lightsim.remote.transfer import extract_tar_stream, write_tar_stream
from sm_tias_lightsim.runtime.overrides import THREADS_ENV_VAR, VARIANT_ENV_VAR
DEFAULT_HOST = "Newton"
DEFAULT_REMOTE_DIR = "~/lab/SmTIAS-LightSim"
# 非ログインの `ssh host 'cmd'` では ~/.profile が読まれず uv が PATH に入らないため絶対パスで呼ぶ
DEFAULT_UV = "~/.local/bin/uv"
# サーバでは GPU が手元の CPU（llvm）より 37〜72 倍速く，結果も一致する（2026-09-13 Newton で実測）
DEFAULT_VARIANT = "cuda_ad_rgb"
ENV_HOST = "SM_REMOTE_HOST"
ENV_REMOTE_DIR = "SM_REMOTE_DIR"
ENV_UV = "SM_REMOTE_UV"
ENV_SSH = "SM_REMOTE_SSH"
ENV_VARIANT = "SM_REMOTE_VARIANT"
# サーバ上での実行検出（detect_running_on_server）を無効にする逃げ道
ENV_ALLOW_SELF = "SM_REMOTE_ALLOW_SELF"
_MAX_LISTED_FILES = 50
_PROGRESS_INTERVAL_S = 0.5
_EPILOG = """\
使用例:
uv run sm-remote push # コードを送る
uv run sm-remote push --include configs/_tmp_offaxis # 除外対象の _tmp も一緒に送る
uv run sm-remote push --only configs/_tmp_offaxis # _tmp だけ送る
uv run sm-remote run --name offaxis --gpu 0 -- python scripts/sweep_offaxis_led.py
uv run sm-remote run --name r100 --gpu 0 -- sm-render --scene configs/x.yaml --output output/x
uv run sm-remote status
uv run sm-remote logs offaxis -f
uv run sm-remote pull offaxis_led_sweep # output/offaxis_led_sweep を持ち帰る
"""
@dataclass(frozen=True)
class RemoteSettings:
"""接続先の設定．
Attributes:
host: SSH ホスト名（~/.ssh/config の Host）．
remote_dir: サーバ上のプロジェクトルート．
uv_path: サーバ上の uv 実行ファイル．
ssh: 手元の ssh 実行ファイル．
"""
host: str
remote_dir: str
uv_path: str
ssh: str
def default_ssh_executable(environ: Mapping[str, str], os_name: str = os.name) -> str:
"""既定の ssh 実行ファイルを返す．
Windows では System32 の OpenSSH を優先する．Git Bash から起動すると PATH 上の
Git 同梱 ssh（MSYS 版）が先に見つかり，PowerShell からの起動と挙動が揃わないため．
Args:
environ: 参照する環境変数．
os_name: `os.name` 相当（テスト用に差し替え可能）．
Returns:
ssh 実行ファイルのパスまたはコマンド名．
"""
if os_name == "nt":
system_root = environ.get("SystemRoot") or environ.get("SYSTEMROOT") or r"C:\Windows"
candidate = Path(system_root) / "System32" / "OpenSSH" / "ssh.exe"
if candidate.is_file():
return str(candidate)
return "ssh"
def resolve_settings(
host: str | None,
remote_dir: str | None,
uv_path: str | None,
environ: Mapping[str, str],
) -> RemoteSettings:
"""引数 > 環境変数 > 既定値 の順で接続先の設定を決める．
Args:
host: --host の値（未指定なら None）．
remote_dir: --remote-dir の値（未指定なら None）．
uv_path: --uv の値（未指定なら None）．
environ: 参照する環境変数．
Returns:
解決済みの設定．
"""
def pick(arg: str | None, key: str, default: str) -> str:
if arg and arg.strip():
return arg.strip()
return environ.get(key, "").strip() or default
return RemoteSettings(
host=pick(host, ENV_HOST, DEFAULT_HOST),
remote_dir=pick(remote_dir, ENV_REMOTE_DIR, DEFAULT_REMOTE_DIR),
uv_path=pick(uv_path, ENV_UV, DEFAULT_UV),
ssh=pick(None, ENV_SSH, default_ssh_executable(environ)),
)
def _short_hostname(name: str) -> str:
"""ホスト名を比較用に正規化する（小文字化してドメイン部を除く．IPv4 はそのまま）．"""
name = name.strip().lower()
if name.replace(".", "").isdigit():
return name
return name.split(".", 1)[0]
def _expand_remote_dir(remote_dir: str, home: Path) -> Path | None:
"""remote_dir を「このマシンがサーバだったら」の前提で展開する（~user 形式は None）．"""
if remote_dir == "~":
return home
if remote_dir.startswith("~/"):
return home / remote_dir[2:]
if remote_dir.startswith("~"):
return None
path = Path(remote_dir)
# ssh 越しの相対パスはホーム基準で解釈される
return path if path.is_absolute() else home / path
def detect_running_on_server(
host: str,
remote_dir: str,
*,
environ: Mapping[str, str],
local_hostname: str,
project_root: Path | None,
home: Path | None,
os_name: str = os.name,
) -> str | None:
"""sm-remote が接続先サーバ自身の上で実行されていないかを判定する．
サーバ上（VS Code Remote の端末等）で実行すると ssh で自分自身へつなぎに行き，
ホスト鍵の確認プロンプト等で混乱するため，ssh を起動する前に止める
（2026-09-13 に Newton で発生）．次のどちらかで「サーバ上」とみなす．
- このマシンのホスト名が接続先と一致する（大文字小文字とドメイン部を無視するので，
ssh config の Host 名 `Newton` と実ホスト名 `newton` / `newton.local` のどれでも一致する）
- POSIX で，いまいるリポジトリのルートが接続先のプロジェクトルート（~ 展開後）と一致する
（Host 名が実ホスト名と異なるエイリアスの場合の保険）
Windows ではプロジェクトルートの比較をしない（手元の PC を誤検出しないため）．
Args:
host: 接続先ホスト（--host / SM_REMOTE_HOST / 既定）．
remote_dir: 接続先のプロジェクトルート．
environ: 参照する環境変数（SM_REMOTE_ALLOW_SELF=1 なら判定しない）．
local_hostname: このマシンのホスト名．
project_root: いまいるリポジトリのルート（リポジトリ外なら None）．
home: このマシンのホームディレクトリ（不明なら None）．
os_name: `os.name` 相当（テスト用に差し替え可能）．
Returns:
サーバ上と判定した理由（日本語）．サーバ上でなければ None．
"""
if environ.get(ENV_ALLOW_SELF, "").strip().lower() in ("1", "true", "yes"):
return None
if local_hostname and _short_hostname(local_hostname) == _short_hostname(host):
return f"このマシンのホスト名 {local_hostname} が接続先 {host} と一致"
if os_name == "nt" or project_root is None or home is None:
return None
remote_path = _expand_remote_dir(remote_dir, home)
if remote_path is not None and remote_path.resolve() == project_root.resolve():
return f"このリポジトリ {project_root} が接続先のプロジェクトルート {remote_dir} と一致"
return None
def _detect_on_this_machine(settings: RemoteSettings) -> str | None:
"""このマシンの情報を集めて detect_running_on_server を呼ぶ．"""
try:
project_root: Path | None = find_project_root(Path.cwd())
except ValueError:
project_root = None
try:
home: Path | None = Path.home()
except RuntimeError:
home = None
return detect_running_on_server(
settings.host,
settings.remote_dir,
environ=os.environ,
local_hostname=socket.gethostname(),
project_root=project_root,
home=home,
)
def build_job_env(
*,
gpu: str | None,
variant: str | None,
threads: int | None,
extra: Sequence[str],
) -> dict[str, str]:
"""ジョブに渡す環境変数を組み立てる．
Args:
gpu: CUDA_VISIBLE_DEVICES に渡す GPU 番号（None なら設定しない）．
variant: SM_RENDER_VARIANT に渡す variant（None なら YAML のまま）．
threads: SM_DRJIT_THREADS に渡すスレッド数（None なら制限しない）．
extra: 追加の `KEY=VALUE`（後勝ち）．
Returns:
環境変数の辞書（挿入順＝ログに表示する順）．
Raises:
ValueError: GPU 指定・スレッド数・KEY=VALUE が不正な場合．
"""
# パイプ越しでも Python の print がすぐログに出るようにする（既定はブロックバッファ）
env = {"PYTHONUNBUFFERED": "1"}
if gpu is not None:
env["CUDA_VISIBLE_DEVICES"] = validate_gpu_spec(gpu)
if variant:
env[VARIANT_ENV_VAR] = variant
if threads is not None:
if threads < 1:
raise ValueError(f"--threads は 1 以上を指定してください（指定 {threads}）．")
env[THREADS_ENV_VAR] = str(threads)
for item in extra:
key, value = parse_env_assignment(item)
env[key] = value
return env
def _default_job_name() -> str:
"""既定のジョブ名（起動時刻）を返す．"""
return datetime.now().strftime("job-%Y%m%d-%H%M%S")
def _format_size(num_bytes: float) -> str:
"""バイト数を読みやすい単位で返す．"""
for unit in ("B", "KB", "MB", "GB"):
if num_bytes < 1024 or unit == "GB":
return f"{num_bytes:.0f} {unit}" if unit == "B" else f"{num_bytes:.1f} {unit}"
num_bytes /= 1024
return f"{num_bytes:.1f} GB"
def _write_stdout(text: str) -> None:
"""標準出力へ書く（出力先の文字コードで表せない文字は置換する）．
リモートの出力は UTF-8 で届くが，リダイレクト時の Windows の標準出力は cp932 のことがある．
"""
try:
sys.stdout.write(text)
except UnicodeEncodeError:
encoding = sys.stdout.encoding or "utf-8"
sys.stdout.write(text.encode(encoding, errors="replace").decode(encoding, errors="replace"))
sys.stdout.flush()
class _ProgressPrinter:
"""転送量を標準エラーの 1 行に上書き表示する（端末でないときは何もしない）．"""
def __init__(self, label: str) -> None:
self._label = label
self._enabled = sys.stderr.isatty()
self._last = 0.0
self._printed = False
def __call__(self, done: int, total: int) -> None:
if not self._enabled:
return
now = time.monotonic()
if now - self._last < _PROGRESS_INTERVAL_S and done != total:
return
self._last = now
if total > 0:
text = f"{self._label} {_format_size(done)} / {_format_size(total)}"
text += f" ({100.0 * done / total:.0f}%)"
else:
text = f"{self._label} {_format_size(done)}"
sys.stderr.write("\r" + text.ljust(60))
sys.stderr.flush()
self._printed = True
def finish(self) -> None:
if self._printed:
sys.stderr.write("\n")
sys.stderr.flush()
self._printed = False
def _ssh_argv(settings: RemoteSettings, command: str, *, tty: bool = False) -> list[str]:
"""接続先の設定で，リモートコマンドを実行する ssh の引数リストを返す．"""
return build_ssh_argv(settings.ssh, settings.host, command, tty=tty)
def _host_option(settings: RemoteSettings) -> str:
"""案内に表示する sm-remote コマンドへ付ける ` --host ...`（既定のホストなら空）を返す．"""
return "" if settings.host == DEFAULT_HOST else f" --host {settings.host}"
def _popen(argv: list[str], **kwargs: object) -> subprocess.Popen[bytes]:
"""ssh を起動する（見つからなければ対処を添えて ValueError にする）．"""
try:
return subprocess.Popen(argv, **kwargs) # noqa: S603
except FileNotFoundError as exc:
msg = f"ssh が見つかりません: {argv[0]}（環境変数 {ENV_SSH} で実行ファイルを指定できます）"
raise ValueError(msg) from exc
def _run_ssh(
argv: list[str], *, capture: bool = False, input_bytes: bytes | None = None
) -> tuple[int, bytes]:
"""ssh を実行して終了を待つ．
標準入力は input_bytes が無ければ /dev/null 相当にする（手元の入力を ssh が
吸い込まないように）．
Returns:
(終了コード, 標準出力．capture=False なら空)．
"""
stdin = subprocess.PIPE if input_bytes is not None else subprocess.DEVNULL
proc = _popen(argv, stdin=stdin, stdout=subprocess.PIPE if capture else None)
out, _ = proc.communicate(input=input_bytes)
return proc.returncode, out or b""
def _report_common_failure(rc: int, settings: RemoteSettings) -> None:
"""ssh・リモートスクリプトに共通の失敗理由を日本語で補足する．"""
if rc == EXIT_REMOTE_DIR_MISSING:
print(
f"エラー: サーバに {settings.remote_dir} がありません．"
"先に sm-remote push してください．",
file=sys.stderr,
)
elif rc == 255:
print(
f"エラー: {settings.host} へ ssh 接続できませんでした"
"（~/.ssh/config・鍵・ネットワークを確認）．",
file=sys.stderr,
)
elif rc != 0:
print(f"エラー: リモートコマンドが失敗しました（終了コード {rc}）．", file=sys.stderr)
def _print_list(title: str, items: Sequence[str]) -> None:
"""一覧を先頭 _MAX_LISTED_FILES 件まで表示する．"""
print(title)
for item in items[:_MAX_LISTED_FILES]:
print(f" {item}")
if len(items) > _MAX_LISTED_FILES:
print(f" ...ほか {len(items) - _MAX_LISTED_FILES} 件")
# ---------------------------------------------------------------------------
# push
# ---------------------------------------------------------------------------
def _stream_push(argv: list[str], entries: Sequence[PushEntry], *, compress: bool) -> int:
"""tar ストリームを ssh の標準入力へ流し，ssh の終了コードを返す．"""
proc = _popen(argv, stdin=subprocess.PIPE)
progress = _ProgressPrinter("送信中")
try:
write_tar_stream(entries, proc.stdin, compress=compress, progress=progress)
except OSError as exc:
progress.finish()
# ssh が先に終了してパイプが閉じている．後始末の close 失敗は原因ではないので無視する
with contextlib.suppress(OSError):
proc.stdin.close()
rc = proc.wait()
print(f"エラー: 送信が中断されました（ssh 終了コード {rc}）— {exc}", file=sys.stderr)
return rc or 1
progress.finish()
proc.stdin.close()
return proc.wait()
def _find_prune_targets(settings: RemoteSettings, local_arcnames: set[str]) -> list[str] | None:
"""サーバの src/ tests/ scripts/ にあって手元に無いファイルを求める（失敗時 None）．"""
command = build_list_files_command(settings.remote_dir, PRUNE_DIRS, sorted(EXCLUDED_DIR_NAMES))
rc, out = _run_ssh(_ssh_argv(settings, command), capture=True)
if rc != 0:
_report_common_failure(rc, settings)
return None
# Linux のファイル名は任意のバイト列なので，往復で失われないよう surrogateescape で扱う
remote_files = [p.decode("utf-8", "surrogateescape") for p in out.split(b"\0") if p]
try:
return compute_prune_targets(remote_files, local_arcnames)
except RuntimeError as exc:
# 不変条件違反（誤削除の危険）なので削除せず，他の失敗経路と同じく None で返す
print(
f"エラー: サーバの一覧に想定外のパスが含まれたため削除を中止しました: {exc}",
file=sys.stderr,
)
return None
def _prune_remote(settings: RemoteSettings, local_arcnames: set[str]) -> int:
"""サーバの src/ tests/ scripts/ にあって手元に無いファイルを削除し，終了コードを返す．"""
targets = _find_prune_targets(settings, local_arcnames)
if targets is None:
return 1
if not targets:
print("--prune: 削除対象はありません．")
return 0
_print_list(f"--prune: サーバから {len(targets)} 件を削除します:", targets)
payload = b"".join(t.encode("utf-8", "surrogateescape") + b"\0" for t in targets)
remove = build_remove_files_command(settings.remote_dir)
rc, _ = _run_ssh(_ssh_argv(settings, remove), input_bytes=payload)
if rc != 0:
_report_common_failure(rc, settings)
return rc
def _cmd_push(args: argparse.Namespace, settings: RemoteSettings) -> int:
"""push サブコマンド: コードをサーバへ送る．"""
if args.include and args.only:
raise ValueError("--include と --only は同時に指定できません．")
if args.prune and args.only:
raise ValueError(
"--prune は --only と同時に指定できません（削除判定に手元の全体が要るため）．"
)
root = args.local_dir.resolve() if args.local_dir else find_project_root(Path.cwd())
entries = collect_push_entries(
root, include_base=not args.only, extra_paths=args.include or args.only
)
if not entries:
raise ValueError("送信するファイルがありません．")
total = sum(entry.size for entry in entries)
print(
f"送信対象: {len(entries)} ファイル / {_format_size(total)}"
f"（{root} → {settings.host}:{settings.remote_dir}）"
)
windows_yaml = find_yaml_with_windows_paths(entries)
if windows_yaml:
_print_list(
"警告: 次の YAML は Windows の絶対パスを含むためサーバでは読めません"
"（生成スクリプトを --prepare-only で再実行すると YAML 相対パスで作り直せます）:",
windows_yaml,
)
if args.dry_run:
for entry in entries:
print(f" {_format_size(entry.size):>10} {entry.arcname}")
if args.prune:
targets = _find_prune_targets(settings, {e.arcname for e in entries})
if targets is None:
return 1
_print_list(f"--prune で削除されるファイル: {len(targets)} 件", targets)
print("（--dry-run のため送信していません）")
return 0
command = build_push_command(settings.remote_dir, compress=args.compress)
start = time.monotonic()
rc = _stream_push(_ssh_argv(settings, command), entries, compress=args.compress)
if rc != 0:
_report_common_failure(rc, settings)
return rc
print(f"送信完了: {time.monotonic() - start:.1f} 秒")
if args.prune:
return _prune_remote(settings, {e.arcname for e in entries})
return 0
# ---------------------------------------------------------------------------
# run / status / logs / attach / kill
# ---------------------------------------------------------------------------
def _cmd_run(args: argparse.Namespace, settings: RemoteSettings) -> int:
"""run サブコマンド: サーバの tmux セッションでジョブを起動する．"""
command = list(args.command)
if command and command[0] == "--":
command = command[1:]
if not command:
raise ValueError(
"実行するコマンドを指定してください（例: sm-remote run -- sm-render --scene ...）．"
)
name = validate_job_name(args.name or _default_job_name())
if args.yaml_variant:
variant = None
else:
variant = args.variant or os.environ.get(ENV_VARIANT, "").strip() or DEFAULT_VARIANT
env = build_job_env(gpu=args.gpu, variant=variant, threads=args.threads, extra=args.env)
script = build_job_script(
settings.remote_dir,
name,
command,
uv_path=settings.uv_path,
use_uv=not args.no_uv,
env=env,
)
remote_command = build_run_command(settings.remote_dir, name, script)
if args.dry_run:
print(remote_command)
return 0
rc, _ = _run_ssh(_ssh_argv(settings, remote_command))
if rc == EXIT_SESSION_EXISTS:
print(
f"エラー: 同名のジョブ（tmux セッション）'{name}' が実行中です．"
"--name で別名を付けるか，終了を待ってください．",
file=sys.stderr,
)
return rc
if rc != 0:
_report_common_failure(rc, settings)
return rc
host_opt = _host_option(settings)
print(f"ジョブを起動しました: {name}（{settings.host}）")
print(f" 環境変数: {' '.join(f'{k}={v}' for k, v in env.items())}")
print(f" ログ : {settings.remote_dir}/{REMOTE_LOG_DIR}/{name}.log")
print(f" 状態確認: uv run sm-remote status{host_opt}")
print(f" ログ追従: uv run sm-remote logs {name} -f{host_opt}")
return 0
def _cmd_status(args: argparse.Namespace, settings: RemoteSettings) -> int:
"""status サブコマンド: 実行中ジョブ・最近のログ・GPU・負荷を表示する．"""
command = build_status_command(settings.remote_dir)
rc, out = _run_ssh(_ssh_argv(settings, command), capture=True)
_write_stdout(out.decode("utf-8", errors="replace"))
if rc != 0:
_report_common_failure(rc, settings)
return rc
def _follow(argv: list[str]) -> int:
"""ssh の出力を逐次表示し続ける（Ctrl+C で終了）．
mitsuba の進捗バーは改行せず \\r で上書きするため，行単位ではなく受け取った分から表示する．
ssh の標準入力はパイプにして開いたままにし，終了時に閉じてリモートの tail を止める．
"""
proc = _popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
interrupted = False
try:
while chunk := proc.stdout.read1(65536):
_write_stdout(decoder.decode(chunk))
except KeyboardInterrupt:
interrupted = True
print("\n（追従を終了しました．ジョブは継続しています）", file=sys.stderr)
finally:
# stdin を閉じるとリモートの cat が EOF を受けて tail を止める．既に ssh が
# 終了していれば close は失敗しうるが，後始末なので無視する
with contextlib.suppress(OSError):
proc.stdin.close()
rc = proc.wait()
# Ctrl+C は追従の正常な終わらせ方（Windows では ssh も同時に止まり非 0 を返す）
return 0 if interrupted else rc
def _cmd_logs(args: argparse.Namespace, settings: RemoteSettings) -> int:
"""logs サブコマンド: ジョブログの末尾を表示する（-f で追従）．"""
if args.lines < 0:
raise ValueError("--lines は 0 以上を指定してください．")
command = build_logs_command(
settings.remote_dir, args.name, lines=args.lines, follow=args.follow
)
argv = _ssh_argv(settings, command)
if args.follow:
rc = _follow(argv)
else:
rc, out = _run_ssh(argv, capture=True)
_write_stdout(out.decode("utf-8", errors="replace"))
if rc == EXIT_NOT_FOUND:
print(f"エラー: ジョブ '{args.name}' のログがサーバにありません．", file=sys.stderr)
elif rc != 0:
_report_common_failure(rc, settings)
return rc
def _cmd_attach(args: argparse.Namespace, settings: RemoteSettings) -> int:
"""attach サブコマンド: ジョブの tmux セッションに接続する．
ジョブが終わっていると tmux の `no sessions` だけが出て分かりにくいので，
先に別の ssh（非対話）でセッションの有無を確かめ，無ければログの最終行を添えて案内する．
"""
check = build_attach_check_command(settings.remote_dir, args.name)
rc, out = _run_ssh(_ssh_argv(settings, check), capture=True)
if rc == EXIT_NO_SESSION:
last = out.decode("utf-8", errors="replace").strip()
detail = f"（{last}）" if last else ""
host_opt = _host_option(settings)
print(
f"ジョブ '{args.name}' は既に終了しています{detail}．"
f"結果は `uv run sm-remote logs {args.name}{host_opt}` で確認できます．"
)
return rc
if rc == EXIT_NOT_FOUND:
print(
f"エラー: ジョブ '{args.name}' はありません（tmux セッションもログも見つかりません）．",
file=sys.stderr,
)
return rc
if rc != 0:
_report_common_failure(rc, settings)
return rc
command = build_attach_command(args.name)
print(
"tmux に接続します．デタッチ（ジョブは継続）は Ctrl+B → D．"
"Ctrl+C はジョブ自体を止めるので注意．",
file=sys.stderr,
)
proc = _popen(_ssh_argv(settings, command, tty=True))
return proc.wait()
def _cmd_kill(args: argparse.Namespace, settings: RemoteSettings) -> int:
"""kill サブコマンド: ジョブの tmux セッションを終了する．"""
command = build_kill_command(settings.remote_dir, args.name)
rc, _ = _run_ssh(_ssh_argv(settings, command))
if rc == EXIT_NO_SESSION:
print(f"エラー: ジョブ '{args.name}' は実行中ではありません．", file=sys.stderr)
elif rc != 0:
_report_common_failure(rc, settings)
else:
print(f"ジョブ '{args.name}' を終了しました．")
return rc
# ---------------------------------------------------------------------------
# pull
# ---------------------------------------------------------------------------
def _cmd_pull(args: argparse.Namespace, settings: RemoteSettings) -> int:
"""pull サブコマンド: output/ のサブディレクトリを持ち帰る．"""
rel_paths = list(dict.fromkeys(normalize_output_subpath(p) for p in args.paths))
dest = args.dest.resolve() if args.dest else find_project_root(Path.cwd())
command = build_pull_command(settings.remote_dir, rel_paths, compress=args.compress)
print(f"取得: {settings.host}:{settings.remote_dir}/{{{', '.join(rel_paths)}}} → {dest}")
proc = _popen(
_ssh_argv(settings, command),
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
)
progress = _ProgressPrinter("受信中")
start = time.monotonic()
try:
files, num_bytes = extract_tar_stream(proc.stdout, dest, rel_paths, progress=progress)
except (tarfile.TarError, OSError) as exc:
progress.finish()
# ssh が既に終わっていれば（リモート側の失敗）その終了コードで原因を示す
ssh_finished = proc.poll() is not None
if not ssh_finished:
proc.kill()
rc = proc.wait()
if ssh_finished and rc == EXIT_NOT_FOUND:
print(
"エラー: 指定したパスがサーバにありません（上のメッセージ参照）．", file=sys.stderr
)
elif ssh_finished and rc in (EXIT_REMOTE_DIR_MISSING, 255):
_report_common_failure(rc, settings)
else:
print(f"エラー: 受信・展開に失敗しました — {exc}", file=sys.stderr)
return rc if ssh_finished and rc > 0 else 1
progress.finish()
# 終端ブロック後のパディングを読み切らないと ssh が書き込みで止まり wait が戻らない
while proc.stdout.read(65536):
pass
rc = proc.wait()
if rc == 1:
# GNU tar の終了コード 1 は「読み取り中にファイルが変化した」等の警告（ジョブ実行中など）
print(
"警告: 一部のファイルが読み取り中に変化しました（ジョブが実行中の可能性）．",
file=sys.stderr,
)
elif rc != 0:
_report_common_failure(rc, settings)
return rc
print(
f"取得完了: {files} ファイル / {_format_size(num_bytes)}"
f"（{time.monotonic() - start:.1f} 秒）"
)
return 0
# ---------------------------------------------------------------------------
# 引数
# ---------------------------------------------------------------------------
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"""コマンドライン引数をパースして返す．
Args:
argv: 引数リスト．None のとき sys.argv[1:] を使用．
Returns:
パース済み Namespace．
"""
parser = argparse.ArgumentParser(
prog="sm-remote",
description="研究室サーバでの計算を支援する（コード送信・ジョブ起動・状態確認・結果取得）．",
epilog=_EPILOG,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
common = argparse.ArgumentParser(add_help=False)
common.add_argument(
"--host",
default=None,
metavar="HOST",
help=(
"SSH ホスト名（~/.ssh/config の Host）．"
f"既定: 環境変数 {ENV_HOST} または {DEFAULT_HOST}．"
),
)
common.add_argument(
"--remote-dir",
default=None,
metavar="DIR",
help=(
f"サーバ上のプロジェクトルート．既定: 環境変数 {ENV_REMOTE_DIR} または "
f"{DEFAULT_REMOTE_DIR}（Git Bash では ~ が手元で展開されるので引用符で囲む）．"
),
)
sub = parser.add_subparsers(dest="subcommand", required=True, metavar="SUBCOMMAND")
p_push = sub.add_parser(
"push",
parents=[common],
help="コードをサーバへ送る",
description=(
"プロジェクトを tar over ssh でサーバへ送る．.venv / output / .git / キャッシュ / "
"configs/_tmp_* は送らない．上書き展開のため，手元で消したファイルはサーバに残る"
"（src/ tests/ scripts/ は --prune で掃除できる）．"
),
)
p_push.add_argument(
"--include",
action="append",
default=[],
metavar="PATH",
help="除外対象（configs/_tmp_* 等）を追加で送る．複数指定可．",
)
p_push.add_argument(
"--only",
action="append",
default=[],
metavar="PATH",
help="指定したパスだけを送る（コード本体は送らない）．複数指定可．",
)
p_push.add_argument(
"--prune",
action="store_true",
help="送信後，src/ tests/ scripts/ のうち手元に無いファイルをサーバから削除する．",
)
p_push.add_argument("--compress", "-z", action="store_true", help="gzip 圧縮して送る．")
p_push.add_argument(
"--dry-run",
action="store_true",
help="送信せず，対象ファイルの一覧と合計サイズだけ表示する．",
)
p_push.add_argument(
"--local-dir",
type=Path,
default=None,
metavar="DIR",
help="送信元のプロジェクトルート（既定: カレントから上方向に探索）．",
)
p_run = sub.add_parser(
"run",
parents=[common],
help="サーバの tmux セッションでジョブを起動する",
description=(
"サーバのプロジェクトルートで `uv run <COMMAND>` を tmux セッション（デタッチ）で"
f"起動する．出力は {REMOTE_LOG_DIR}/<NAME>.log に追記され，接続が切れても継続する．"
"コマンドは -- の後に書く．"
),
)
p_run.add_argument(
"--name",
"-n",
default=None,
metavar="NAME",
help="ジョブ名（tmux セッション名・ログ名）．英数字・_・-．既定: job-YYYYmmdd-HHMMSS．",
)
p_run.add_argument(
"--gpu",
default=None,
metavar="IDS",
help=(
"CUDA_VISIBLE_DEVICES に渡す GPU 番号（例: 0 / 0,2）．"
"status で空きを確認してから指定する．"
),
)
p_run.add_argument(
"--variant",
default=None,
metavar="VARIANT",
help=(
f"{VARIANT_ENV_VAR} としてジョブに渡す Mitsuba variant"
"（sm-render が YAML より優先して使う）．"
f"既定: 環境変数 {ENV_VARIANT} または {DEFAULT_VARIANT}．"
),
)
p_run.add_argument(
"--yaml-variant",
action="store_true",
help="variant を上書きせず YAML の render.variant を使う（pytest を回すとき等）．",
)
p_run.add_argument(
"--threads",
type=int,
default=None,
metavar="N",
help=(
f"{THREADS_ENV_VAR} としてジョブに渡す CPU スレッド数上限"
"（llvm / scalar で共用サーバの全コアを占有しないため）．"
),
)
p_run.add_argument(
"--env",
action="append",
default=[],
metavar="KEY=VALUE",
help="ジョブに追加で渡す環境変数．複数指定可．",
)
p_run.add_argument(
"--no-uv", action="store_true", help="uv run を付けずにコマンドをそのまま実行する．"
)
p_run.add_argument(
"--uv",
default=None,
metavar="PATH",
help=f"サーバ上の uv の場所．既定: 環境変数 {ENV_UV} または {DEFAULT_UV}．",
)
p_run.add_argument(
"--dry-run", action="store_true", help="実行せず，サーバへ送るシェルコマンドを表示する．"
)
p_run.add_argument(
"command",
nargs=argparse.REMAINDER,
metavar="-- COMMAND",
help="サーバで実行するコマンド（uv run の後ろに付く部分）．",
)
sub.add_parser(
"status",
parents=[common],
help="実行中ジョブ・最近のジョブログ・GPU・負荷を表示する",
)
p_logs = sub.add_parser("logs", parents=[common], help="ジョブログを表示する")
p_logs.add_argument("name", metavar="NAME", help="ジョブ名．")
p_logs.add_argument(
"--lines", "-n", type=int, default=40, metavar="N", help="表示する末尾の行数（既定 40）．"
)
p_logs.add_argument(
"--follow",
"-f",
action="store_true",
help="追記を追従表示する（Ctrl+C で終了．ジョブは継続）．",
)
p_attach = sub.add_parser(
"attach", parents=[common], help="ジョブの tmux セッションに接続する（PowerShell 推奨）"
)
p_attach.add_argument("name", metavar="NAME", help="ジョブ名．")
p_kill = sub.add_parser("kill", parents=[common], help="ジョブ（tmux セッション）を終了する")
p_kill.add_argument("name", metavar="NAME", help="ジョブ名．")
p_pull = sub.add_parser(
"pull",
parents=[common],
help="output/ のサブディレクトリを持ち帰る",
description=(
"サーバの output/<PATH> を tar over ssh で受け取り，手元の同じ場所に展開する"
"（同名ファイルは上書き）．"
),
)
p_pull.add_argument(
"paths",
nargs="+",
metavar="PATH",
help="output/ 配下のパス（例: offaxis_led_sweep / output/_remote_logs）．複数指定可．",
)
p_pull.add_argument(
"--dest",
type=Path,
default=None,
metavar="DIR",
help="展開先（既定: 手元のプロジェクトルート．output/<PATH> がその下にできる）．",
)
p_pull.add_argument("--compress", "-z", action="store_true", help="gzip 圧縮して受け取る．")
return parser.parse_args(argv)
_HANDLERS: dict[str, Callable[[argparse.Namespace, RemoteSettings], int]] = {
"push": _cmd_push,
"run": _cmd_run,
"status": _cmd_status,
"logs": _cmd_logs,
"attach": _cmd_attach,
"kill": _cmd_kill,
"pull": _cmd_pull,
}
def main(argv: list[str] | None = None) -> int:
"""sm-remote コマンドのメインエントリポイント．
Args:
argv: 引数リスト．None のとき sys.argv[1:] を使用．
Returns:
終了コード（成功時 0．ssh・リモートの失敗時はその終了コード，入力エラーは 1）．
"""
args = _parse_args(argv)
try:
settings = resolve_settings(
args.host, args.remote_dir, getattr(args, "uv", None), os.environ
)
on_server = _detect_on_this_machine(settings)
if on_server is not None:
print(
f"エラー: ここは接続先サーバ（{settings.host}）です（{on_server}）．\n"
" sm-remote は手元の PC の PowerShell 等から実行してください．\n"
" サーバ上で直接回すなら，tmux の中で uv run ... を実行してください"
f"（例: tmux new -s job → cd {settings.remote_dir} && uv run sm-render ...）．\n"
f" 判定が誤りなら環境変数 {ENV_ALLOW_SELF}=1 で無効にできます．",
file=sys.stderr,
)
return 1
return _HANDLERS[args.subcommand](args, settings)
except ValueError as exc:
print(f"エラー: {exc}", file=sys.stderr)
return 1
except KeyboardInterrupt:
print("\n中断しました．", file=sys.stderr)
return 130
if __name__ == "__main__":
# uv の生成するランチャ（sm-remote.exe）が Windows の Smart App Control に
# ブロックされた場合の代替として `uv run python -m sm_tias_lightsim.cli.remote` を
# 使えるようにする
sys.exit(main())
