pkgs/servers/home-assistant/update.py

   1 #!/usr/bin/env nix-shell
   2 #!nix-shell -I nixpkgs=channel:nixpkgs-unstable -i python3 -p "python3.withPackages (ps: with ps; [ aiohttp packaging ])" -p git nurl pyright ruff isort
   3
   4 import argparse
   5 import asyncio
   6 import json
   7 import os
   8 import re
   9 import sys
  10 from subprocess import check_output, run
  11 from typing import Dict, Final, List, Optional, Union
  12
  13 import aiohttp
  14 from aiohttp import ClientSession
  15 from packaging.version import Version
  16
  17 ROOT: Final = check_output([
  18     "git",
  19     "rev-parse",
  20     "--show-toplevel",
  21 ]).decode().strip()
  22
  23
  24 def run_sync(cmd: List[str]) -> None:
  25     print(f"$ {' '.join(cmd)}")
  26     process = run(cmd)
  27
  28     if process.returncode != 0:
  29         sys.exit(1)
  30
  31
  32 async def check_async(cmd: List[str]) -> str:
  33     print(f"$ {' '.join(cmd)}")
  34     process = await asyncio.create_subprocess_exec(
  35         *cmd,
  36         stdout=asyncio.subprocess.PIPE,
  37         stderr=asyncio.subprocess.PIPE
  38     )
  39     stdout, stderr = await process.communicate()
  40
  41     if process.returncode != 0:
  42         error = stderr.decode()
  43         raise RuntimeError(f"{cmd[0]} failed: {error}")
  44
  45     return stdout.decode().strip()
  46
  47
  48 async def run_async(cmd: List[str]):
  49     print(f"$ {' '.join(cmd)}")
  50
  51     process = await asyncio.create_subprocess_exec(
  52         *cmd,
  53         stdout=asyncio.subprocess.PIPE,
  54         stderr=asyncio.subprocess.PIPE,
  55     )
  56     stdout, stderr = await process.communicate()
  57
  58     print(stdout.decode())
  59
  60     if process.returncode != 0:
  61         error = stderr.decode()
  62         raise RuntimeError(f"{cmd[0]} failed: {error}")
  63
  64
  65 class File:
  66     def __init__(self, path: str):
  67         self.path = os.path.join(ROOT, path)
  68
  69     def __enter__(self):
  70         with open(self.path, "r") as handle:
  71             self.text = handle.read()
  72         return self
  73
  74     def get_exact_match(self, attr: str, value: str):
  75         matches = re.findall(
  76             rf'{re.escape(attr)}\s+=\s+\"?{re.escape(value)}\"?',
  77             self.text
  78         )
  79
  80         n = len(matches)
  81         if n > 1:
  82             raise ValueError(f"multiple occurrences found for {attr}={value}")
  83         elif n == 1:
  84             return matches.pop()
  85         else:
  86             raise ValueError(f"no occurrence found for {attr}={value}")
  87
  88     def substitute(self, attr: str, old_value: str, new_value: str) -> None:
  89         old_line = self.get_exact_match(attr, old_value)
  90         new_line = old_line.replace(old_value, new_value)
  91         self.text = self.text.replace(old_line, new_line)
  92         print(f"Substitute `{attr}` value `{old_value}` with `{new_value}`")
  93
  94     def __exit__(self, exc_type, exc_val, exc_tb):
  95         with open(self.path, "w") as handle:
  96             handle.write(self.text)
  97
  98 class Nurl:
  99     @classmethod
 100     async def prefetch(cls, url: str, version: str, *extra_args: str) -> str:
 101         cmd = [
 102             "nurl",
 103             "--hash",
 104             url,
 105             version,
 106         ]
 107         cmd.extend(extra_args)
 108         return await check_async(cmd)
 109
 110
 111 class Nix:
 112     base_cmd: Final = [
 113         "nix",
 114         "--show-trace",
 115         "--extra-experimental-features", "nix-command"
 116     ]
 117
 118     @classmethod
 119     async def _run(cls, args: List[str]) -> Optional[str]:
 120         return await check_async(cls.base_cmd + args)
 121
 122     @classmethod
 123     async def eval(cls, expr: str) -> Union[List, Dict, int, float, str, bool]:
 124         response = await cls._run([
 125             "eval",
 126             "-f", f"{ROOT}/default.nix",
 127             "--json",
 128             expr
 129         ])
 130         if response is None:
 131             raise RuntimeError("Nix eval expression returned no response")
 132         try:
 133             return json.loads(response)
 134         except (TypeError, ValueError):
 135             raise RuntimeError("Nix eval response could not be parsed from JSON")
 136
 137     @classmethod
 138     async def hash_to_sri(cls, algorithm: str, value: str) -> Optional[str]:
 139         return await cls._run([
 140             "hash",
 141             "to-sri",
 142             "--type", algorithm,
 143             value
 144         ])
 145
 146
 147 class HomeAssistant:
 148     def __init__(self, session: ClientSession):
 149         self._session = session
 150
 151     async def get_latest_core_version(
 152         self,
 153         owner: str = "home-assistant",
 154         repo: str = "core"
 155     ) -> str:
 156         async with self._session.get(
 157             f"https://api.github.com/repos/{owner}/{repo}/releases/latest"
 158         ) as response:
 159             document = await response.json()
 160         try:
 161             return str(document.get("name"))
 162         except KeyError:
 163             raise RuntimeError("No tag name in response document")
 164
 165
 166     async def get_latest_frontend_version(
 167         self,
 168         core_version: str
 169     ) -> str:
 170         async with self._session.get(
 171             f"https://raw.githubusercontent.com/home-assistant/core/{core_version}/homeassistant/components/frontend/manifest.json"
 172         ) as response:
 173             document = await response.json(content_type="text/plain")
 174
 175         requirements = [
 176             requirement
 177             for requirement in document.get("requirements", [])
 178             if requirement.startswith("home-assistant-frontend==")
 179         ]
 180
 181         if len(requirements) > 1:
 182             raise RuntimeError(
 183                 "Found more than one version specifier for the frontend package"
 184             )
 185         elif len(requirements) == 1:
 186             requirement = requirements.pop()
 187             _, version = requirement.split("==", maxsplit=1)
 188             return str(version)
 189         else:
 190             raise RuntimeError(
 191                 "Found no version specifier for frontend package"
 192             )
 193
 194
 195     async def update_core(self, old_version: str, new_version: str) -> None:
 196         old_sdist_hash = str(await Nix.eval("home-assistant.sdist.outputHash"))
 197         new_sdist_hash = await Nurl.prefetch("https://pypi.org/project/homeassistant/", new_version)
 198         print(f"sdist: {old_sdist_hash} -> {new_sdist_hash}")
 199
 200         old_git_hash = str(await Nix.eval("home-assistant.src.outputHash"))
 201         new_git_hash = await Nurl.prefetch("https://github.com/home-assistant/core/", new_version)
 202         print(f"git: {old_git_hash} -> {new_git_hash}")
 203
 204         with File("pkgs/servers/home-assistant/default.nix") as file:
 205             file.substitute("hassVersion", old_version, new_version)
 206             file.substitute("hash", old_sdist_hash, new_sdist_hash)
 207             file.substitute("hash", old_git_hash, new_git_hash)
 208
 209     async def update_frontend(self, old_version: str, new_version: str) -> None:
 210         old_hash = str(await Nix.eval("home-assistant.frontend.src.outputHash"))
 211         new_hash = await Nurl.prefetch(
 212             "https://pypi.org/project/home_assistant_frontend/",
 213             new_version,
 214             "-A", "format", "wheel",
 215             "-A", "dist", "py3",
 216             "-A", "python", "py3"
 217         )
 218         print(f"frontend: {old_hash} -> {new_hash}")
 219
 220         with File("pkgs/servers/home-assistant/frontend.nix") as file:
 221             file.substitute("version", old_version, new_version)
 222             file.substitute("hash", old_hash, new_hash)
 223
 224     async def update_components(self):
 225         await run_async([
 226             f"{ROOT}/pkgs/servers/home-assistant/update-component-packages.py"
 227         ])
 228
 229
 230 async def main(target_version: Optional[str] = None):
 231     headers = {}
 232     if token := os.environ.get("GITHUB_TOKEN", None):
 233         headers.update({"GITHUB_TOKEN": token})
 234
 235     async with aiohttp.ClientSession(headers=headers) as client:
 236         hass = HomeAssistant(client)
 237
 238         core_current = str(await Nix.eval("home-assistant.version"))
 239         core_latest = target_version or await hass.get_latest_core_version()
 240
 241         if Version(core_latest) > Version(core_current):
 242             print(f"New Home Assistant version {core_latest} is available")
 243             await hass.update_core(str(core_current), str(core_latest))
 244
 245             frontend_current = str(await Nix.eval("home-assistant.frontend.version"))
 246             frontend_latest = await hass.get_latest_frontend_version(str(core_latest))
 247
 248             if Version(frontend_latest) > Version(frontend_current):
 249                 await hass.update_frontend(str(frontend_current), str(frontend_latest))
 250
 251             await hass.update_components()
 252
 253         else:
 254             print(f"Home Assistant {core_current} is still the latest version.")
 255
 256         # wait for async client sessions to close
 257         # https://docs.aiohttp.org/en/stable/client_advanced.html#graceful-shutdown
 258         await asyncio.sleep(0)
 259
 260 if __name__ == "__main__":
 261     parser = argparse.ArgumentParser()
 262     parser.add_argument("version", nargs="?")
 263     args = parser.parse_args()
 264
 265     run_sync(["pyright", __file__])
 266     run_sync(["ruff", "check", "--ignore=E501", __file__])
 267     run_sync(["isort", __file__])
 268
 269     asyncio.run(main(args.version))