path: root/pkgs/servers/home-assistant/update.py

#!/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())