nixpkgs/pkgs/servers/home-assistant/update.py

264 lines
8.4 KiB
Python
Executable File

#!/usr/bin/env nix-shell
#!nix-shell -I nixpkgs=channel:nixpkgs-unstable -i python3 -p "python3.withPackages (ps: with ps; [ aiohttp packaging ])" -p git nurl nodePackages.pyright ruff isort
import asyncio
import json
import os
import re
import sys
from subprocess import check_output, run
from typing import Dict, Final, List, Optional, Union
import aiohttp
from aiohttp import ClientSession
from packaging.version import Version
ROOT: Final = check_output([
"git",
"rev-parse",
"--show-toplevel",
]).decode().strip()
def run_sync(cmd: List[str]) -> None:
print(f"$ {' '.join(cmd)}")
process = run(cmd)
if process.returncode != 0:
sys.exit(1)
async def check_async(cmd: List[str]) -> str:
print(f"$ {' '.join(cmd)}")
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
error = stderr.decode()
raise RuntimeError(f"{cmd[0]} failed: {error}")
return stdout.decode().strip()
async def run_async(cmd: List[str]):
print(f"$ {' '.join(cmd)}")
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
print(stdout.decode())
if process.returncode != 0:
error = stderr.decode()
raise RuntimeError(f"{cmd[0]} failed: {error}")
class File:
def __init__(self, path: str):
self.path = os.path.join(ROOT, path)
def __enter__(self):
with open(self.path, "r") as handle:
self.text = handle.read()
return self
def get_exact_match(self, attr: str, value: str):
matches = re.findall(
rf'{re.escape(attr)}\s+=\s+\"?{re.escape(value)}\"?',
self.text
)
n = len(matches)
if n > 1:
raise ValueError(f"multiple occurrences found for {attr}={value}")
elif n == 1:
return matches.pop()
else:
raise ValueError(f"no occurrence found for {attr}={value}")
def substitute(self, attr: str, old_value: str, new_value: str) -> None:
old_line = self.get_exact_match(attr, old_value)
new_line = old_line.replace(old_value, new_value)
self.text = self.text.replace(old_line, new_line)
print(f"Substitute `{attr}` value `{old_value}` with `{new_value}`")
def __exit__(self, exc_type, exc_val, exc_tb):
with open(self.path, "w") as handle:
handle.write(self.text)
class Nurl:
@classmethod
async def prefetch(cls, url: str, version: str, *extra_args: str) -> str:
cmd = [
"nurl",
"--hash",
url,
version,
]
cmd.extend(extra_args)
return await check_async(cmd)
class Nix:
base_cmd: Final = [
"nix",
"--show-trace",
"--extra-experimental-features", "nix-command"
]
@classmethod
async def _run(cls, args: List[str]) -> Optional[str]:
return await check_async(cls.base_cmd + args)
@classmethod
async def eval(cls, expr: str) -> Union[List, Dict, int, float, str, bool]:
response = await cls._run([
"eval",
"-f", f"{ROOT}/default.nix",
"--json",
expr
])
if response is None:
raise RuntimeError("Nix eval expression returned no response")
try:
return json.loads(response)
except (TypeError, ValueError):
raise RuntimeError("Nix eval response could not be parsed from JSON")
@classmethod
async def hash_to_sri(cls, algorithm: str, value: str) -> Optional[str]:
return await cls._run([
"hash",
"to-sri",
"--type", algorithm,
value
])
class HomeAssistant:
def __init__(self, session: ClientSession):
self._session = session
async def get_latest_core_version(
self,
owner: str = "home-assistant",
repo: str = "core"
) -> str:
async with self._session.get(
f"https://api.github.com/repos/{owner}/{repo}/releases/latest"
) as response:
document = await response.json()
try:
return str(document.get("name"))
except KeyError:
raise RuntimeError("No tag name in response document")
async def get_latest_frontend_version(
self,
core_version: str
) -> str:
async with self._session.get(
f"https://raw.githubusercontent.com/home-assistant/core/{core_version}/homeassistant/components/frontend/manifest.json"
) as response:
document = await response.json(content_type="text/plain")
requirements = [
requirement
for requirement in document.get("requirements", [])
if requirement.startswith("home-assistant-frontend==")
]
if len(requirements) > 1:
raise RuntimeError(
"Found more than one version specifier for the frontend package"
)
elif len(requirements) == 1:
requirement = requirements.pop()
_, version = requirement.split("==", maxsplit=1)
return str(version)
else:
raise RuntimeError(
"Found no version specifier for frontend package"
)
async def update_core(self, old_version: str, new_version: str) -> None:
old_sdist_hash = str(await Nix.eval("home-assistant.sdist.outputHash"))
new_sdist_hash = await Nurl.prefetch("https://pypi.org/project/homeassistant/", new_version)
print(f"sdist: {old_sdist_hash} -> {new_sdist_hash}")
old_git_hash = str(await Nix.eval("home-assistant.src.outputHash"))
new_git_hash = await Nurl.prefetch("https://github.com/home-assistant/core/", new_version)
print(f"git: {old_git_hash} -> {new_git_hash}")
with File("pkgs/servers/home-assistant/default.nix") as file:
file.substitute("hassVersion", old_version, new_version)
file.substitute("hash", old_sdist_hash, new_sdist_hash)
file.substitute("hash", old_git_hash, new_git_hash)
async def update_frontend(self, old_version: str, new_version: str) -> None:
old_hash = str(await Nix.eval("home-assistant.frontend.src.outputHash"))
new_hash = await Nurl.prefetch(
"https://pypi.org/project/home_assistant_frontend/",
new_version,
"-A", "format", "wheel",
"-A", "dist", "py3",
"-A", "python", "py3"
)
print(f"frontend: {old_hash} -> {new_hash}")
with File("pkgs/servers/home-assistant/frontend.nix") as file:
file.substitute("version", old_version, new_version)
file.substitute("hash", old_hash, new_hash)
async def update_components(self):
await run_async([
f"{ROOT}/pkgs/servers/home-assistant/parse-requirements.py"
])
async def main():
headers = {}
if token := os.environ.get("GITHUB_TOKEN", None):
headers.update({"GITHUB_TOKEN": token})
async with aiohttp.ClientSession(headers=headers) as client:
hass = HomeAssistant(client)
core_current = str(await Nix.eval("home-assistant.version"))
core_latest = await hass.get_latest_core_version()
if Version(core_latest) > Version(core_current):
print(f"New Home Assistant version {core_latest} is available")
await hass.update_core(str(core_current), str(core_latest))
frontend_current = str(await Nix.eval("home-assistant.frontend.version"))
frontend_latest = await hass.get_latest_frontend_version(str(core_latest))
if Version(frontend_latest) > Version(frontend_current):
await hass.update_frontend(str(frontend_current), str(frontend_latest))
await hass.update_components()
else:
print(f"Home Assistant {core_current} is still the latest version.")
# wait for async client sessions to close
# https://docs.aiohttp.org/en/stable/client_advanced.html#graceful-shutdown
await asyncio.sleep(0)
if __name__ == "__main__":
run_sync(["pyright", __file__])
run_sync(["ruff", "check", "--ignore=E501", __file__])
run_sync(["isort", __file__])
asyncio.run(main())