from __future__ import annotations
import hashlib
import json
import os
import shutil
import socket
import subprocess
import time
import uuid
from pathlib import Path
from typing import Any
class LabError(RuntimeError):
def __init__(self, code: str, message: str, details: dict[str, Any] | None = None):
super().__init__(message)
self.code = code
self.message = message
self.details = details or {}
def to_dict(self) -> dict[str, Any]:
return {"code": self.code, "message": self.message, "details": self.details}
def require(condition: bool, code: str, message: str, **details: Any) -> None:
if not condition:
raise LabError(code=code, message=message, details=details)
def find_executable(candidates: list[str]) -> str | None:
for name in candidates:
found = shutil.which(name)
if found:
return found
return None
def run_command(
args: list[str],
*,
timeout: float = 60.0,
check: bool = False,
cwd: Path | None = None,
) -> subprocess.CompletedProcess[str]:
result = subprocess.run(
args,
capture_output=True,
text=True,
timeout=timeout,
cwd=str(cwd) if cwd else None,
)
if check and result.returncode != 0:
raise LabError(
code="command_failed",
message=f"Command failed: {' '.join(args)}",
details={
"return_code": result.returncode,
"stdout": result.stdout,
"stderr": result.stderr,
},
)
return result
def run_command_no_fail(args: list[str], timeout: float = 30.0) -> subprocess.CompletedProcess[str]:
try:
return run_command(args, timeout=timeout, check=False)
except Exception as exc:
return subprocess.CompletedProcess(args=args, returncode=1, stdout="", stderr=str(exc))
def compute_sha256(path: Path, block_size: int = 1024 * 1024) -> str:
h = hashlib.sha256()
with path.open("rb") as fp:
while True:
chunk = fp.read(block_size)
if not chunk:
break
h.update(chunk)
return h.hexdigest()
def ensure_dir(path: Path) -> Path:
path.mkdir(parents=True, exist_ok=True)
return path
def load_json(path: Path, default: Any) -> Any:
if not path.exists():
return default
with path.open("r", encoding="utf-8") as fp:
return json.load(fp)
def save_json(path: Path, value: Any) -> None:
ensure_dir(path.parent)
tmp_name = f".{path.name}.{uuid.uuid4().hex}.tmp"
tmp_path = path.parent / tmp_name
with tmp_path.open("w", encoding="utf-8") as fp:
json.dump(value, fp, indent=2, sort_keys=True)
os.replace(tmp_path, path)
def utc_timestamp() -> float:
return time.time()
def acquire_free_tcp_port(host: str = "127.0.0.1") -> int:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.bind((host, 0))
return int(sock.getsockname()[1])
def windows_local_appdata() -> Path:
local_appdata = os.environ.get("LOCALAPPDATA")
if not local_appdata:
raise LabError(
code="env_missing",
message="LOCALAPPDATA is not set; cannot resolve workspace path on Windows.",
)
return Path(local_appdata)
