tjh.dev
nixpkgs mirror (for testing)
github.com/NixOS/nixpkgs
nix
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 nodePackages.pyright ruff isort
3
4import asyncio
5import json
6import os
7import re
8import sys
9from subprocess import check_output, run
10from typing import Dict, Final, List, Optional, Union
11
12import aiohttp
13from aiohttp import ClientSession
14from packaging.version import Version
15
16ROOT: Final = check_output([
17 "git",
18 "rev-parse",
19 "--show-toplevel",
20]).decode().strip()
21
22
23def 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
31async 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
47async 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
64class 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
97class 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
110class 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
146class 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.src.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.gitSrc.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/parse-requirements.py"
226 ])
227
228
229async 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
259if __name__ == "__main__":
260 run_sync(["pyright", __file__])
261 run_sync(["ruff", "--ignore=E501", __file__])
262 run_sync(["isort", __file__])
263 asyncio.run(main())