a9c8c99442
Make the git src the primary one, since it can be replaced with a local checkout to run bisects. We now copy the translations from the sdist published on PyPi into the source tree, instead of previously the tests from the git source tree.
263 lines
8.4 KiB
Python
Executable file
263 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", "--ignore=E501", __file__])
|
|
run_sync(["isort", __file__])
|
|
asyncio.run(main())
|