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