333 lines
11 KiB
Python
333 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
import shlex
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
|
|
|
|
DEFAULT_DEPLOY_ENV = Path("config/deploy.env")
|
|
DEFAULT_REMOTE_DIR = "~/ha-deconz-bridge"
|
|
APP_ARCHIVE_NAME = "ha-deconz-bridge.tar.gz"
|
|
RUNTIME_ENV_PREFIX = "HA_DECONZ_BRIDGE_"
|
|
EXCLUDED_DIR_NAMES = {
|
|
".git",
|
|
".venv",
|
|
".pytest_cache",
|
|
"__pycache__",
|
|
"config",
|
|
"old attempt",
|
|
}
|
|
EXCLUDED_SUFFIXES = {".pyc"}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Deploy ha-deconz-bridge to a Raspberry Pi or similar Linux host."
|
|
)
|
|
parser.add_argument(
|
|
"--deploy-env",
|
|
default=str(DEFAULT_DEPLOY_ENV),
|
|
help="Local deployment env file. Defaults to config/deploy.env.",
|
|
)
|
|
parser.add_argument("--host", help="SSH target, overriding DEPLOY_HOST.")
|
|
parser.add_argument("--remote-dir", help="Remote install directory, overriding DEPLOY_REMOTE_DIR.")
|
|
parser.add_argument(
|
|
"--config",
|
|
help=(
|
|
"Local runtime config entry point. All .py files next to it are copied "
|
|
"to the remote config directory; the entry point is also copied as "
|
|
"lights_config.py."
|
|
),
|
|
)
|
|
parser.add_argument("--service-env", help="Local service env copied to remote config/service.env.")
|
|
parser.add_argument(
|
|
"--config-only",
|
|
action="store_true",
|
|
help="Only copy config/service env and restart the service.",
|
|
)
|
|
parser.add_argument(
|
|
"--no-start",
|
|
action="store_true",
|
|
help="Deploy files but do not enable, start, or restart the service.",
|
|
)
|
|
parser.add_argument(
|
|
"--restart-debug",
|
|
action="store_true",
|
|
help="Restart the optional web debugger service after deployment.",
|
|
)
|
|
parser.add_argument(
|
|
"--with-speed",
|
|
action="store_true",
|
|
help="Install optional speed dependencies, overriding DEPLOY_INSTALL_SPEED.",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
repo_root = Path(__file__).resolve().parents[1]
|
|
deploy_env_path = _resolve_path(repo_root, args.deploy_env)
|
|
deploy_env = _read_env_file(deploy_env_path)
|
|
|
|
host = args.host or deploy_env.get("DEPLOY_HOST")
|
|
if not host:
|
|
raise SystemExit(
|
|
"Missing DEPLOY_HOST. Copy config.example/deploy.env.example to "
|
|
"config/deploy.env and edit it, or pass --host."
|
|
)
|
|
|
|
remote_dir = args.remote_dir or deploy_env.get("DEPLOY_REMOTE_DIR", DEFAULT_REMOTE_DIR)
|
|
config_path = _local_file(
|
|
repo_root,
|
|
args.config or deploy_env.get("DEPLOY_CONFIG", "config/lights_config.py"),
|
|
required=True,
|
|
)
|
|
service_env_path = _service_env_path(repo_root, args.service_env, deploy_env)
|
|
install_speed = args.with_speed or _env_bool(deploy_env.get("DEPLOY_INSTALL_SPEED", "false"))
|
|
restart_debug = args.restart_debug or _env_bool(deploy_env.get("DEPLOY_RESTART_DEBUG", "false"))
|
|
|
|
_deploy(
|
|
host=host,
|
|
remote_dir=remote_dir,
|
|
repo_root=repo_root,
|
|
deploy_env=deploy_env,
|
|
config_path=config_path,
|
|
service_env_path=service_env_path,
|
|
config_only=args.config_only,
|
|
no_start=args.no_start,
|
|
install_speed=install_speed,
|
|
restart_debug=restart_debug,
|
|
)
|
|
return 0
|
|
|
|
|
|
def _deploy(
|
|
*,
|
|
host: str,
|
|
remote_dir: str,
|
|
repo_root: Path,
|
|
deploy_env: dict[str, str],
|
|
config_path: Path,
|
|
service_env_path: Path | None,
|
|
config_only: bool,
|
|
no_start: bool,
|
|
install_speed: bool,
|
|
restart_debug: bool,
|
|
) -> None:
|
|
print(f"Deploying to {host}:{remote_dir}")
|
|
_ssh(host, f"mkdir -p {_q(remote_dir)}")
|
|
install_dir = _remote_realpath(host, remote_dir)
|
|
remote_app = f"{install_dir}/app"
|
|
remote_config = f"{install_dir}/config"
|
|
remote_venv = f"{install_dir}/.venv"
|
|
|
|
_ssh(host, f"mkdir -p {_q(remote_app)} {_q(remote_config)} {_q('~/.config/systemd/user')}")
|
|
|
|
if not config_only:
|
|
_ssh(host, f"rm -rf {_q(remote_app)} && mkdir -p {_q(remote_app)}")
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
archive = Path(temp_dir) / APP_ARCHIVE_NAME
|
|
_create_archive(repo_root, archive)
|
|
remote_archive = f"{install_dir}/{APP_ARCHIVE_NAME}"
|
|
_scp(host, archive, remote_archive)
|
|
_ssh(host, f"tar -xzf {_q(remote_archive)} -C {_q(remote_app)} && rm -f {_q(remote_archive)}")
|
|
|
|
_copy_runtime_config(host, config_path, remote_config)
|
|
_write_and_copy_service_env(host, deploy_env, service_env_path, f"{remote_config}/service.env")
|
|
_install_service_units(host, repo_root, install_dir)
|
|
|
|
if not config_only:
|
|
_ssh(host, f"python3 -m venv {_q(remote_venv)}")
|
|
_ssh(host, f"{_q(f'{remote_venv}/bin/pip')} install --upgrade pip")
|
|
package = f"{remote_app}[speed]" if install_speed else remote_app
|
|
_ssh(host, f"{_q(f'{remote_venv}/bin/pip')} install -e {_q(package)}")
|
|
|
|
_ssh(host, "systemctl --user daemon-reload")
|
|
if not no_start:
|
|
_ssh(host, "systemctl --user enable ha-deconz-bridge.service")
|
|
_ssh(host, "systemctl --user restart ha-deconz-bridge.service")
|
|
if restart_debug:
|
|
_ssh(host, "systemctl --user restart ha-deconz-bridge-debug.service")
|
|
|
|
print()
|
|
print("Deploy complete.")
|
|
print(f"Install dir: {install_dir}")
|
|
print(f"Remote config: {remote_config}/lights_config.py")
|
|
print(f"Remote env file: {remote_config}/service.env")
|
|
print("Service unit: ~/.config/systemd/user/ha-deconz-bridge.service")
|
|
print()
|
|
print("Useful commands:")
|
|
print(f" ssh {host} 'systemctl --user status ha-deconz-bridge.service --no-pager'")
|
|
print(f" ssh {host} 'journalctl --user -u ha-deconz-bridge.service -f'")
|
|
print(f" ssh {host} 'systemctl --user start ha-deconz-bridge-debug.service'")
|
|
|
|
|
|
def _read_env_file(path: Path) -> dict[str, str]:
|
|
if not path.exists():
|
|
return {}
|
|
values: dict[str, str] = {}
|
|
for line in path.read_text(encoding="utf-8").splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
if "=" not in stripped:
|
|
raise SystemExit(f"Invalid env line in {path}: {line!r}")
|
|
key, value = stripped.split("=", 1)
|
|
values[key.strip()] = _strip_env_value(value.strip())
|
|
return values
|
|
|
|
|
|
def _strip_env_value(value: str) -> str:
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
|
|
return value[1:-1]
|
|
return value
|
|
|
|
|
|
def _service_env_path(
|
|
repo_root: Path,
|
|
override: str | None,
|
|
deploy_env: dict[str, str],
|
|
) -> Path | None:
|
|
candidate = override or deploy_env.get("DEPLOY_SERVICE_ENV")
|
|
if candidate:
|
|
return _local_file(repo_root, candidate, required=True)
|
|
default = repo_root / "config" / "service.env"
|
|
return default if default.exists() else None
|
|
|
|
|
|
def _local_file(repo_root: Path, value: str, *, required: bool) -> Path:
|
|
path = _resolve_path(repo_root, value)
|
|
if required and not path.is_file():
|
|
raise SystemExit(f"Required file not found: {path}")
|
|
return path
|
|
|
|
|
|
def _resolve_path(repo_root: Path, value: str | Path) -> Path:
|
|
path = Path(value).expanduser()
|
|
if not path.is_absolute():
|
|
path = repo_root / path
|
|
return path
|
|
|
|
|
|
def _write_and_copy_service_env(
|
|
host: str,
|
|
deploy_env: dict[str, str],
|
|
service_env_path: Path | None,
|
|
remote_path: str,
|
|
) -> None:
|
|
runtime_values = {
|
|
key: value
|
|
for key, value in deploy_env.items()
|
|
if key.startswith(RUNTIME_ENV_PREFIX)
|
|
}
|
|
if service_env_path is not None:
|
|
runtime_values.update(_read_env_file(service_env_path))
|
|
if not runtime_values:
|
|
if _remote_file_exists(host, remote_path):
|
|
print(f"Keeping existing remote service env: {remote_path}")
|
|
return
|
|
raise SystemExit(
|
|
"No service environment values found and no remote service.env exists. "
|
|
"Add HA_DECONZ_BRIDGE_* values to config/deploy.env or provide "
|
|
"DEPLOY_SERVICE_ENV=config/service.env."
|
|
)
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
local_env = Path(temp_dir) / "service.env"
|
|
lines = [f"{key}={_quote_env_value(value)}" for key, value in sorted(runtime_values.items())]
|
|
local_env.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
_scp(host, local_env, remote_path)
|
|
|
|
|
|
def _copy_runtime_config(host: str, config_path: Path, remote_config: str) -> None:
|
|
config_dir = config_path.parent
|
|
config_files = sorted(path for path in config_dir.glob("*.py") if path.is_file())
|
|
if config_path not in config_files:
|
|
config_files.append(config_path)
|
|
|
|
_ssh(host, f"find {_q(remote_config)} -maxdepth 1 -type f -name '*.py' -delete")
|
|
for path in config_files:
|
|
_scp(host, path, f"{remote_config}/{path.name}")
|
|
|
|
if config_path.name != "lights_config.py":
|
|
_scp(host, config_path, f"{remote_config}/lights_config.py")
|
|
|
|
|
|
def _quote_env_value(value: str) -> str:
|
|
if not value or any(character.isspace() or character in {'"', "'", "\\"} for character in value):
|
|
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
|
|
return f'"{escaped}"'
|
|
return value
|
|
|
|
|
|
def _install_service_units(host: str, repo_root: Path, install_dir: str) -> None:
|
|
for unit_name in ("ha-deconz-bridge.service", "ha-deconz-bridge-debug.service"):
|
|
template = repo_root / "deploy" / unit_name
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
unit = Path(temp_dir) / unit_name
|
|
unit.write_text(
|
|
template.read_text(encoding="utf-8").replace("__INSTALL_DIR__", install_dir),
|
|
encoding="utf-8",
|
|
)
|
|
_scp(host, unit, f"~/.config/systemd/user/{unit_name}")
|
|
|
|
|
|
def _create_archive(repo_root: Path, archive: Path) -> None:
|
|
with tarfile.open(archive, "w:gz") as tar:
|
|
for path in repo_root.rglob("*"):
|
|
relative = path.relative_to(repo_root)
|
|
if _excluded(relative, path):
|
|
continue
|
|
tar.add(path, arcname=relative, recursive=False)
|
|
|
|
|
|
def _excluded(relative: Path, path: Path) -> bool:
|
|
if any(part in EXCLUDED_DIR_NAMES for part in relative.parts):
|
|
return True
|
|
if any(part.endswith(".egg-info") for part in relative.parts):
|
|
return True
|
|
if path.is_file() and path.suffix in EXCLUDED_SUFFIXES:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _env_bool(value: str) -> bool:
|
|
return value.strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def _remote_realpath(host: str, path: str) -> str:
|
|
result = subprocess.run(
|
|
["ssh", host, f"cd {_q(path)} && pwd -P"],
|
|
check=True,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
)
|
|
return result.stdout.strip()
|
|
|
|
|
|
def _remote_file_exists(host: str, path: str) -> bool:
|
|
return subprocess.run(
|
|
["ssh", host, f"test -f {_q(path)}"],
|
|
check=False,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
).returncode == 0
|
|
|
|
|
|
def _ssh(host: str, command: str) -> None:
|
|
subprocess.run(["ssh", host, command], check=True)
|
|
|
|
|
|
def _scp(host: str, source: Path, remote_path: str) -> None:
|
|
subprocess.run(["scp", str(source), f"{host}:{remote_path}"], check=True)
|
|
|
|
|
|
def _q(value: str) -> str:
|
|
if value.startswith("~/"):
|
|
return "~/" + shlex.quote(value[2:])
|
|
return shlex.quote(value)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|