Compare commits
1 Commits
Author | SHA1 | Date | |
---|---|---|---|
0fd0a99c52 |
3
.gitignore
vendored
3
.gitignore
vendored
@ -141,6 +141,3 @@ cython_debug/
|
|||||||
tags
|
tags
|
||||||
unhacs.txt
|
unhacs.txt
|
||||||
poetry.lock
|
poetry.lock
|
||||||
custom_components/
|
|
||||||
themes/
|
|
||||||
unhacs.yaml
|
|
||||||
|
23
Makefile
23
Makefile
@ -24,7 +24,7 @@ lint: devenv
|
|||||||
# Runs tests
|
# Runs tests
|
||||||
.PHONY: test
|
.PHONY: test
|
||||||
test: devenv
|
test: devenv
|
||||||
poetry run python -m unittest discover tests --pattern "*_test.py"
|
@echo TODO: poetry run pytest
|
||||||
|
|
||||||
# Builds wheel for package to upload
|
# Builds wheel for package to upload
|
||||||
.PHONY: build
|
.PHONY: build
|
||||||
@ -37,27 +37,6 @@ verify-tag-version:
|
|||||||
$(eval TAG_NAME = $(shell [ -n "$(DRONE_TAG)" ] && echo $(DRONE_TAG) || git describe --tags --exact-match))
|
$(eval TAG_NAME = $(shell [ -n "$(DRONE_TAG)" ] && echo $(DRONE_TAG) || git describe --tags --exact-match))
|
||||||
test "v$(shell poetry version | awk '{print $$2}')" = "$(TAG_NAME)"
|
test "v$(shell poetry version | awk '{print $$2}')" = "$(TAG_NAME)"
|
||||||
|
|
||||||
.PHONY: bump-patch
|
|
||||||
bump-patch:
|
|
||||||
$(eval NEW_VERSION = $(shell poetry version patch | awk '{print $$6}'))
|
|
||||||
git add pyproject.toml
|
|
||||||
git commit -m "Bump version to $(NEW_VERSION)"
|
|
||||||
git tag "v$(NEW_VERSION)"
|
|
||||||
|
|
||||||
.PHONY: bump-minor
|
|
||||||
bump-minor:
|
|
||||||
$(eval NEW_VERSION = $(shell poetry version minor | awk '{print $$6}'))
|
|
||||||
git add pyproject.toml
|
|
||||||
git commit -m "Bump version to $(NEW_VERSION)"
|
|
||||||
git tag "v$(NEW_VERSION)"
|
|
||||||
|
|
||||||
.PHONY: bump-major
|
|
||||||
bump-major:
|
|
||||||
$(eval NEW_VERSION = $(shell poetry version major | awk '{print $$6}'))
|
|
||||||
git add pyproject.toml
|
|
||||||
git commit -m "Bump version to $(NEW_VERSION)"
|
|
||||||
git tag "v$(NEW_VERSION)"
|
|
||||||
|
|
||||||
# Upload to pypi
|
# Upload to pypi
|
||||||
.PHONY: upload
|
.PHONY: upload
|
||||||
upload: verify-tag-version build
|
upload: verify-tag-version build
|
||||||
|
32
README.md
32
README.md
@ -17,33 +17,13 @@ Unhacs provides several commands to manage your Home Assistant packages:
|
|||||||
To add a package, use the `add` command followed by the URL of the package. Optionally, you can specify the package name and version:
|
To add a package, use the `add` command followed by the URL of the package. Optionally, you can specify the package name and version:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
unhacs add <package_url> --version <version>
|
unhacs add --url <package_url> --name <package_name> --version <version>
|
||||||
```
|
```
|
||||||
|
|
||||||
If the package already exists, you can update it by adding the `--update` flag:
|
If the package already exists, you can update it by adding the `--update` flag:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
unhacs add <package_url> --update
|
unhacs add --url <package_url> --update
|
||||||
```
|
|
||||||
|
|
||||||
If the package is a Lovelace plugin, you must specify it using the `--plugin` flag:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
unhacs add --plugin <package_url>
|
|
||||||
```
|
|
||||||
|
|
||||||
If you already have a list of packages in a file, you can add them all at once using the `--file` flag:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
unhacs add --file <file_path>
|
|
||||||
```
|
|
||||||
|
|
||||||
### Add a component from a forked Home Assistant Core repository
|
|
||||||
|
|
||||||
To add a component from a fork of home-assistant/core, use the `--forked-component` flag followed by the URL of the forked repository and then specify the branch with the `--branch` flag:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
unhacs add --forked-component <forked_repo_url> --branch <branch>
|
|
||||||
```
|
```
|
||||||
|
|
||||||
### List packages
|
### List packages
|
||||||
@ -96,14 +76,6 @@ To upgrade specific packages, add their names after the `upgrade` command:
|
|||||||
unhacs upgrade <package_name_1> <package_name_2> ...
|
unhacs upgrade <package_name_1> <package_name_2> ...
|
||||||
```
|
```
|
||||||
|
|
||||||
## Use git tags
|
|
||||||
|
|
||||||
By default, identification of releases uses the GitHub API. If you want to use git tags instead, you can add the `--git-tags` flag to the base command:
|
|
||||||
|
|
||||||
```bash
|
|
||||||
unhacs --git-tags add <package_url>
|
|
||||||
```
|
|
||||||
|
|
||||||
## License
|
## License
|
||||||
|
|
||||||
Unhacs is licensed under the MIT License. See the LICENSE file for more details.
|
Unhacs is licensed under the MIT License. See the LICENSE file for more details.
|
||||||
|
@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
|
|||||||
|
|
||||||
[tool.poetry]
|
[tool.poetry]
|
||||||
name = "unhacs"
|
name = "unhacs"
|
||||||
version = "0.7.1"
|
version = "0.4.0"
|
||||||
description = "Command line interface to install Home Assistant Community Store packages"
|
description = "Command line interface to install Home Assistant Community Store packages"
|
||||||
authors = ["Ian Fijolek <ian@iamthefij.com>"]
|
authors = ["Ian Fijolek <ian@iamthefij.com>"]
|
||||||
license = "MIT"
|
license = "MIT"
|
||||||
@ -12,16 +12,14 @@ readme = "README.md"
|
|||||||
|
|
||||||
[tool.poetry.dependencies]
|
[tool.poetry.dependencies]
|
||||||
python = "^3.11"
|
python = "^3.11"
|
||||||
requests = "^2.32.0"
|
requests = "^2.32.3"
|
||||||
pyyaml = "^6.0.0"
|
pyyaml = "^6.0.1"
|
||||||
|
|
||||||
[tool.poetry.group.dev.dependencies]
|
[tool.poetry.group.dev.dependencies]
|
||||||
black = "^24.4.2"
|
black = "^24.4.2"
|
||||||
isort = "^5.13.2"
|
|
||||||
mypy = "^1.10.0"
|
mypy = "^1.10.0"
|
||||||
pre-commit = "^3.7.1"
|
pre-commit = "^3.7.1"
|
||||||
types-requests = "^2.32.0"
|
types-requests = "^2.32.0.20240602"
|
||||||
types-pyyaml = "^6.0.0"
|
|
||||||
|
|
||||||
[tool.poetry.scripts]
|
[tool.poetry.scripts]
|
||||||
unhacs = 'unhacs.main:main'
|
unhacs = 'unhacs.main:main'
|
||||||
|
@ -1,260 +0,0 @@
|
|||||||
import os
|
|
||||||
import shutil
|
|
||||||
import tempfile
|
|
||||||
import unittest
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from unhacs.main import main
|
|
||||||
from unhacs.packages import get_installed_packages
|
|
||||||
from unhacs.packages import read_lock_packages
|
|
||||||
|
|
||||||
INTEGRATION_URL = "https://github.com/simbaja/ha_gehome"
|
|
||||||
INTEGRATION_VERSION = "v0.6.9"
|
|
||||||
|
|
||||||
PLUGIN_URL = "https://github.com/kalkih/mini-media-player"
|
|
||||||
PLUGIN_VERSION = "v1.16.8"
|
|
||||||
|
|
||||||
THEME_URL = "https://github.com/basnijholt/lovelace-ios-themes"
|
|
||||||
THEME_VERSION = "v3.0.1"
|
|
||||||
|
|
||||||
FORK_URL = "https://github.com/ViViDboarder/home-assistant"
|
|
||||||
FORK_BRANCH = "dev"
|
|
||||||
FORK_COMPONENT = "nextbus"
|
|
||||||
FORK_VERSION = "3b2893f2f4e16f9a05d9cc4a7ba9f31984c841be"
|
|
||||||
|
|
||||||
|
|
||||||
class TestMainIntegrarion(unittest.TestCase):
|
|
||||||
test_dir: str
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
self.test_dir = tempfile.mkdtemp()
|
|
||||||
os.chdir(self.test_dir)
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
shutil.rmtree(self.test_dir)
|
|
||||||
pass
|
|
||||||
|
|
||||||
def run_itest(
|
|
||||||
self,
|
|
||||||
test_name: str,
|
|
||||||
command: str,
|
|
||||||
expected_files: list[str] | None = None,
|
|
||||||
expect_missing_files: list[str] | None = None,
|
|
||||||
expected_code: int = 0,
|
|
||||||
):
|
|
||||||
with self.subTest(test_name, command=command):
|
|
||||||
self.assertEqual(main(command.split()), expected_code)
|
|
||||||
|
|
||||||
# Verify that the package was installed by checking the filesystem
|
|
||||||
if expected_files:
|
|
||||||
expected_files = [
|
|
||||||
os.path.join(self.test_dir, file) for file in expected_files
|
|
||||||
]
|
|
||||||
missing_files = [
|
|
||||||
file for file in expected_files if not os.path.exists(file)
|
|
||||||
]
|
|
||||||
if missing_files:
|
|
||||||
self.fail(f"Missing files: {missing_files}")
|
|
||||||
|
|
||||||
if expect_missing_files:
|
|
||||||
expect_missing_files = [
|
|
||||||
os.path.join(self.test_dir, file) for file in expect_missing_files
|
|
||||||
]
|
|
||||||
existing_files = [
|
|
||||||
file for file in expect_missing_files if os.path.exists(file)
|
|
||||||
]
|
|
||||||
if existing_files:
|
|
||||||
self.fail(f"Files should not exist: {existing_files}")
|
|
||||||
|
|
||||||
def test_integration(self):
|
|
||||||
self.run_itest(
|
|
||||||
"Add integration",
|
|
||||||
f"add {INTEGRATION_URL} --version {INTEGRATION_VERSION}",
|
|
||||||
expected_files=[
|
|
||||||
"custom_components/ge_home/__init__.py",
|
|
||||||
"custom_components/ge_home/manifest.json",
|
|
||||||
"custom_components/ge_home/switch.py",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"List installed packages",
|
|
||||||
"list",
|
|
||||||
)
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 1)
|
|
||||||
self.assertEqual(installed[0].url, INTEGRATION_URL)
|
|
||||||
self.assertEqual(installed[0].version, INTEGRATION_VERSION)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"Double add",
|
|
||||||
f"add {INTEGRATION_URL}",
|
|
||||||
expected_code=1,
|
|
||||||
)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"Upgrade to latest version",
|
|
||||||
"upgrade ha_gehome --yes",
|
|
||||||
expected_files=[
|
|
||||||
"custom_components/ge_home/__init__.py",
|
|
||||||
"custom_components/ge_home/manifest.json",
|
|
||||||
"custom_components/ge_home/switch.py",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 1)
|
|
||||||
self.assertEqual(installed[0].url, INTEGRATION_URL)
|
|
||||||
self.assertNotEqual(installed[0].version, INTEGRATION_VERSION)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"Downgrade integration",
|
|
||||||
f"add {INTEGRATION_URL} --version {INTEGRATION_VERSION} --update",
|
|
||||||
expected_files=[
|
|
||||||
"custom_components/ge_home/__init__.py",
|
|
||||||
"custom_components/ge_home/manifest.json",
|
|
||||||
"custom_components/ge_home/switch.py",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"List installed packages",
|
|
||||||
"list",
|
|
||||||
)
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 1)
|
|
||||||
self.assertEqual(installed[0].url, INTEGRATION_URL)
|
|
||||||
self.assertEqual(installed[0].version, INTEGRATION_VERSION)
|
|
||||||
|
|
||||||
# Delete the custom_components folder and re-install the integration using the lock file
|
|
||||||
shutil.rmtree(os.path.join(self.test_dir, "custom_components"))
|
|
||||||
self.run_itest(
|
|
||||||
"Re-install integration using lock file",
|
|
||||||
"add --file unhacs.yaml",
|
|
||||||
expected_files=[
|
|
||||||
"custom_components/ge_home/__init__.py",
|
|
||||||
"custom_components/ge_home/manifest.json",
|
|
||||||
"custom_components/ge_home/switch.py",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
# Delete the lock file and then regenerate it
|
|
||||||
os.remove(os.path.join(self.test_dir, "unhacs.yaml"))
|
|
||||||
self.run_itest(
|
|
||||||
"Regenerate lock file",
|
|
||||||
"list --freeze",
|
|
||||||
expected_files=[
|
|
||||||
"unhacs.yaml",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
self.assertGreater(len(read_lock_packages()), 0)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"Remove integration",
|
|
||||||
"remove ha_gehome --yes",
|
|
||||||
expect_missing_files=[
|
|
||||||
"custom_components/ge_home/__init__.py",
|
|
||||||
"custom_components/ge_home/manifest.json",
|
|
||||||
"custom_components/ge_home/switch.py",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 0)
|
|
||||||
|
|
||||||
def test_plugin(self):
|
|
||||||
self.run_itest(
|
|
||||||
"Add plugin",
|
|
||||||
f"add --plugin {PLUGIN_URL} --version {PLUGIN_VERSION}",
|
|
||||||
expected_files=[
|
|
||||||
"www/js/mini-media-player-bundle.js",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"List installed packages",
|
|
||||||
"list",
|
|
||||||
)
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 1)
|
|
||||||
self.assertEqual(installed[0].url, PLUGIN_URL)
|
|
||||||
self.assertEqual(installed[0].version, PLUGIN_VERSION)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"Remove plugin",
|
|
||||||
"remove mini-media-player --yes",
|
|
||||||
expect_missing_files=[
|
|
||||||
"www/js/mini-media-player-bundle.js",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 0)
|
|
||||||
|
|
||||||
def test_theme(self):
|
|
||||||
self.run_itest(
|
|
||||||
"Add theme",
|
|
||||||
f"add --theme {THEME_URL} --version {THEME_VERSION}",
|
|
||||||
expected_files=[
|
|
||||||
"themes/ios-themes.yaml",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"List installed packages",
|
|
||||||
"list",
|
|
||||||
)
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 1)
|
|
||||||
self.assertEqual(installed[0].url, THEME_URL)
|
|
||||||
self.assertEqual(installed[0].version, THEME_VERSION)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"Remove theme",
|
|
||||||
"remove lovelace-ios-themes --yes",
|
|
||||||
expect_missing_files=[
|
|
||||||
"themes/ios-themes.yaml",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 0)
|
|
||||||
|
|
||||||
def test_fork(self):
|
|
||||||
self.run_itest(
|
|
||||||
"Add fork",
|
|
||||||
f"add {FORK_URL} --fork-component {FORK_COMPONENT} --fork-branch {FORK_BRANCH} --version {FORK_VERSION}",
|
|
||||||
expected_files=[
|
|
||||||
"custom_components/nextbus/__init__.py",
|
|
||||||
"custom_components/nextbus/manifest.json",
|
|
||||||
"custom_components/nextbus/sensor.py",
|
|
||||||
"custom_components/nextbus/unhacs.yaml",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"List installed packages",
|
|
||||||
"list",
|
|
||||||
)
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 1)
|
|
||||||
self.assertEqual(installed[0].url, FORK_URL)
|
|
||||||
self.assertEqual(installed[0].version, FORK_VERSION)
|
|
||||||
|
|
||||||
self.run_itest(
|
|
||||||
"Remove fork",
|
|
||||||
f"remove {FORK_URL} --yes",
|
|
||||||
expect_missing_files=[
|
|
||||||
"custom_components/nextbus/__init__.py",
|
|
||||||
"custom_components/nextbus/manifest.json",
|
|
||||||
"custom_components/nextbus/sensor.py",
|
|
||||||
"custom_components/nextbus/unhacs.yaml",
|
|
||||||
],
|
|
||||||
)
|
|
||||||
|
|
||||||
installed = get_installed_packages()
|
|
||||||
self.assertEqual(len(installed), 0)
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main()
|
|
@ -1,4 +1,4 @@
|
|||||||
from unhacs.main import main
|
from unhacs.main import main
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
exit(main())
|
main()
|
||||||
|
@ -67,30 +67,5 @@ def get_repo_tags(repository_url: str) -> list[str]:
|
|||||||
return [tag.name for tag in tags]
|
return [tag.name for tag in tags]
|
||||||
|
|
||||||
|
|
||||||
def get_latest_sha(repository_url: str, branch_name: str) -> str:
|
def get_ref_zip(repository_url: str, tag_name: str) -> str:
|
||||||
command = f"git ls-remote {repository_url} {branch_name}"
|
|
||||||
result = subprocess.run(
|
|
||||||
command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
|
|
||||||
)
|
|
||||||
|
|
||||||
# Check for errors
|
|
||||||
if result.returncode != 0:
|
|
||||||
raise Exception(f"Error running command: {command}\n{result.stderr.decode()}")
|
|
||||||
|
|
||||||
for line in result.stdout.decode().split("\n"):
|
|
||||||
if line:
|
|
||||||
return line.partition("\t")[0]
|
|
||||||
|
|
||||||
raise ValueError(f"branch name '{branch_name}' not found for {repository_url}")
|
|
||||||
|
|
||||||
|
|
||||||
def get_tag_zip(repository_url: str, tag_name: str) -> str:
|
|
||||||
return f"{repository_url}/archive/refs/tags/{tag_name}.zip"
|
return f"{repository_url}/archive/refs/tags/{tag_name}.zip"
|
||||||
|
|
||||||
|
|
||||||
def get_branch_zip(repository_url: str, branch_name: str) -> str:
|
|
||||||
return f"{repository_url}/archive/{branch_name}.zip"
|
|
||||||
|
|
||||||
|
|
||||||
def get_sha_zip(repository_url: str, sha: str) -> str:
|
|
||||||
return f"{repository_url}/archive/{sha}.zip"
|
|
||||||
|
238
unhacs/main.py
238
unhacs/main.py
@ -1,30 +1,18 @@
|
|||||||
import sys
|
|
||||||
from argparse import ArgumentParser
|
from argparse import ArgumentParser
|
||||||
from collections.abc import Iterable
|
from collections.abc import Iterable
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
from unhacs.git import get_repo_tags
|
from unhacs.git import get_repo_tags
|
||||||
|
from unhacs.packages import DEFAULT_HASS_CONFIG_PATH
|
||||||
|
from unhacs.packages import DEFAULT_PACKAGE_FILE
|
||||||
from unhacs.packages import Package
|
from unhacs.packages import Package
|
||||||
|
from unhacs.packages import PackageType
|
||||||
from unhacs.packages import get_installed_packages
|
from unhacs.packages import get_installed_packages
|
||||||
from unhacs.packages import read_lock_packages
|
from unhacs.packages import read_lock_packages
|
||||||
from unhacs.packages import write_lock_packages
|
from unhacs.packages import write_lock_packages
|
||||||
from unhacs.packages.fork import Fork
|
|
||||||
from unhacs.packages.integration import Integration
|
|
||||||
from unhacs.packages.plugin import Plugin
|
|
||||||
from unhacs.packages.theme import Theme
|
|
||||||
from unhacs.utils import DEFAULT_HASS_CONFIG_PATH
|
|
||||||
from unhacs.utils import DEFAULT_PACKAGE_FILE
|
|
||||||
|
|
||||||
|
|
||||||
class InvalidArgumentsError(ValueError):
|
def create_parser():
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class DuplicatePackageError(ValueError):
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
def parse_args(argv: list[str]):
|
|
||||||
parser = ArgumentParser(
|
parser = ArgumentParser(
|
||||||
description="Unhacs - Command line interface for the Home Assistant Community Store"
|
description="Unhacs - Command line interface for the Home Assistant Community Store"
|
||||||
)
|
)
|
||||||
@ -43,78 +31,34 @@ def parse_args(argv: list[str]):
|
|||||||
help="The path to the package file.",
|
help="The path to the package file.",
|
||||||
)
|
)
|
||||||
parser.add_argument(
|
parser.add_argument(
|
||||||
"--git-tags",
|
"--use-git",
|
||||||
"-g",
|
"-g",
|
||||||
action="store_true",
|
action="store_true",
|
||||||
help="Use git to search for version tags. This will avoid GitHub API limits.",
|
help="Use git to install packages. This will avoid GitHub API limits.",
|
||||||
)
|
)
|
||||||
|
|
||||||
subparsers = parser.add_subparsers(dest="subcommand", required=True)
|
subparsers = parser.add_subparsers(dest="subcommand", required=True)
|
||||||
|
|
||||||
# List installed packages
|
|
||||||
list_parser = subparsers.add_parser("list", description="List installed packages.")
|
list_parser = subparsers.add_parser("list", description="List installed packages.")
|
||||||
list_parser.add_argument("--verbose", "-v", action="store_true")
|
list_parser.add_argument("--verbose", "-v", action="store_true")
|
||||||
list_parser.add_argument(
|
|
||||||
"--freeze",
|
|
||||||
"-f",
|
|
||||||
action="store_true",
|
|
||||||
help="Regenerate unhacs.yaml with installed packages.",
|
|
||||||
)
|
|
||||||
|
|
||||||
# List git tags for a given package
|
|
||||||
list_tags_parser = subparsers.add_parser("tags", help="List tags for a package.")
|
list_tags_parser = subparsers.add_parser("tags", help="List tags for a package.")
|
||||||
list_tags_parser.add_argument("url", type=str, help="The URL of the package.")
|
list_tags_parser.add_argument("url", type=str, help="The URL of the package.")
|
||||||
list_tags_parser.add_argument(
|
list_tags_parser.add_argument(
|
||||||
"--limit", type=int, default=10, help="The number of tags to display."
|
"--limit", type=int, default=10, help="The number of tags to display."
|
||||||
)
|
)
|
||||||
|
|
||||||
# Add packages
|
|
||||||
add_parser = subparsers.add_parser("add", description="Add or install packages.")
|
add_parser = subparsers.add_parser("add", description="Add or install packages.")
|
||||||
|
add_parser.add_argument(
|
||||||
package_group = add_parser.add_mutually_exclusive_group(required=True)
|
|
||||||
package_group.add_argument(
|
|
||||||
"--file", "-f", type=Path, help="The path to a package file."
|
"--file", "-f", type=Path, help="The path to a package file."
|
||||||
)
|
)
|
||||||
package_group.add_argument(
|
|
||||||
"url", nargs="?", type=str, help="The URL of the package."
|
|
||||||
)
|
|
||||||
|
|
||||||
package_type_group = add_parser.add_mutually_exclusive_group()
|
|
||||||
package_type_group.add_argument(
|
|
||||||
"--integration",
|
|
||||||
action="store_const",
|
|
||||||
dest="type",
|
|
||||||
const=Integration,
|
|
||||||
default=Integration,
|
|
||||||
help="The package is an integration.",
|
|
||||||
)
|
|
||||||
package_type_group.add_argument(
|
|
||||||
"--plugin",
|
|
||||||
action="store_const",
|
|
||||||
dest="type",
|
|
||||||
const=Plugin,
|
|
||||||
help="The package is a JavaScript plugin.",
|
|
||||||
)
|
|
||||||
package_type_group.add_argument(
|
|
||||||
"--theme",
|
|
||||||
action="store_const",
|
|
||||||
dest="type",
|
|
||||||
const=Theme,
|
|
||||||
help="The package is a theme.",
|
|
||||||
)
|
|
||||||
package_type_group.add_argument(
|
|
||||||
"--fork-component",
|
|
||||||
type=str,
|
|
||||||
help="Name of component from forked core repo.",
|
|
||||||
)
|
|
||||||
# Additional arguments for forked packages
|
|
||||||
add_parser.add_argument(
|
add_parser.add_argument(
|
||||||
"--fork-branch",
|
"--type",
|
||||||
"-b",
|
"-t",
|
||||||
type=str,
|
type=PackageType,
|
||||||
help="Name of branch of forked core repo. (Only for forked components.)",
|
help="The type of the package. Defaults to 'integration'.",
|
||||||
)
|
)
|
||||||
|
add_parser.add_argument("url", nargs="?", type=str, help="The URL of the package.")
|
||||||
add_parser.add_argument(
|
add_parser.add_argument(
|
||||||
"--version", "-v", type=str, help="The version of the package."
|
"--version", "-v", type=str, help="The version of the package."
|
||||||
)
|
)
|
||||||
@ -131,38 +75,17 @@ def parse_args(argv: list[str]):
|
|||||||
help="The version of the package to ignore. Multiple can be split by a comma.",
|
help="The version of the package to ignore. Multiple can be split by a comma.",
|
||||||
)
|
)
|
||||||
|
|
||||||
# Remove packages
|
|
||||||
remove_parser = subparsers.add_parser(
|
remove_parser = subparsers.add_parser(
|
||||||
"remove", description="Remove installed packages."
|
"remove", description="Remove installed packages."
|
||||||
)
|
)
|
||||||
remove_parser.add_argument(
|
|
||||||
"--yes", "-y", action="store_true", help="Do not prompt for confirmation."
|
|
||||||
)
|
|
||||||
remove_parser.add_argument("packages", nargs="+")
|
remove_parser.add_argument("packages", nargs="+")
|
||||||
|
|
||||||
# Upgrade packages
|
|
||||||
update_parser = subparsers.add_parser(
|
update_parser = subparsers.add_parser(
|
||||||
"upgrade", description="Upgrade installed packages."
|
"upgrade", description="Upgrade installed packages."
|
||||||
)
|
)
|
||||||
update_parser.add_argument(
|
|
||||||
"--yes", "-y", action="store_true", help="Do not prompt for confirmation."
|
|
||||||
)
|
|
||||||
update_parser.add_argument("packages", nargs="*")
|
update_parser.add_argument("packages", nargs="*")
|
||||||
|
|
||||||
args = parser.parse_args(argv)
|
return parser
|
||||||
|
|
||||||
if args.subcommand == "add":
|
|
||||||
# Component implies forked package
|
|
||||||
if args.fork_component and args.type != Fork:
|
|
||||||
args.type = Fork
|
|
||||||
|
|
||||||
# Branch is only valid for forked packages
|
|
||||||
if args.type != Fork and args.fork_branch:
|
|
||||||
raise InvalidArgumentsError(
|
|
||||||
"Branch and component can only be used with forked packages"
|
|
||||||
)
|
|
||||||
|
|
||||||
return args
|
|
||||||
|
|
||||||
|
|
||||||
class Unhacs:
|
class Unhacs:
|
||||||
@ -182,29 +105,37 @@ class Unhacs:
|
|||||||
|
|
||||||
def add_package(
|
def add_package(
|
||||||
self,
|
self,
|
||||||
package: Package,
|
package_url: str,
|
||||||
|
version: str | None = None,
|
||||||
update: bool = False,
|
update: bool = False,
|
||||||
|
package_type: PackageType = PackageType.INTEGRATION,
|
||||||
|
ignore_versions: set[str] | None = None,
|
||||||
):
|
):
|
||||||
"""Install and add a package to the lock or install a specific version."""
|
"""Install and add a package to the lock or install a specific version."""
|
||||||
|
package = Package(
|
||||||
|
package_url,
|
||||||
|
version=version,
|
||||||
|
package_type=package_type,
|
||||||
|
ignored_versions=ignore_versions,
|
||||||
|
)
|
||||||
packages = self.read_lock_packages()
|
packages = self.read_lock_packages()
|
||||||
|
|
||||||
# Raise an error if the package is already in the list
|
# Raise an error if the package is already in the list
|
||||||
if existing_package := next((p for p in packages if p.same(package)), None):
|
existing_package = next((p for p in packages if p.url == package.url), None)
|
||||||
|
if existing_package:
|
||||||
if update:
|
if update:
|
||||||
# Remove old version of the package
|
# Remove old version of the package
|
||||||
packages = [p for p in packages if p == existing_package]
|
packages = [p for p in packages if p.url != package.url]
|
||||||
else:
|
else:
|
||||||
raise DuplicatePackageError("Package already exists in the list")
|
raise ValueError("Package already exists in the list")
|
||||||
|
|
||||||
package.install(self.hass_config)
|
package.install(self.hass_config)
|
||||||
|
|
||||||
packages.append(package)
|
packages.append(package)
|
||||||
self.write_lock_packages(packages)
|
self.write_lock_packages(packages)
|
||||||
|
|
||||||
def upgrade_packages(self, package_names: list[str], yes: bool = False):
|
def upgrade_packages(self, package_names: list[str]):
|
||||||
"""Uograde to latest version of packages and update lock."""
|
"""Uograde to latest version of packages and update lock."""
|
||||||
installed_packages: Iterable[Package]
|
|
||||||
|
|
||||||
if not package_names:
|
if not package_names:
|
||||||
installed_packages = get_installed_packages(self.hass_config)
|
installed_packages = get_installed_packages(self.hass_config)
|
||||||
else:
|
else:
|
||||||
@ -225,60 +156,35 @@ class Unhacs:
|
|||||||
)
|
)
|
||||||
outdated_packages.append(latest_package)
|
outdated_packages.append(latest_package)
|
||||||
|
|
||||||
confirmed = yes or input("Upgrade all packages? (y/N) ").lower() == "y"
|
if outdated_packages and input("Upgrade all packages? (y/N) ").lower() != "y":
|
||||||
if outdated_packages and not confirmed:
|
|
||||||
return
|
return
|
||||||
|
|
||||||
for installed_package in outdated_packages:
|
for installed_package in outdated_packages:
|
||||||
installed_package.install(self.hass_config)
|
installed_package.install(self.hass_config)
|
||||||
|
|
||||||
# Update lock file to latest now that we know they are uograded
|
# Update lock file to latest now that we know they are uograded
|
||||||
latest_lookup = {p: p for p in latest_packages}
|
latest_lookup = {p.url: p for p in latest_packages}
|
||||||
packages = [latest_lookup.get(p, p) for p in self.read_lock_packages()]
|
packages = [latest_lookup.get(p.url, p) for p in self.read_lock_packages()]
|
||||||
|
|
||||||
self.write_lock_packages(packages)
|
self.write_lock_packages(packages)
|
||||||
|
|
||||||
def list_packages(self, verbose: bool = False, freeze: bool = False):
|
def list_packages(self, verbose: bool = False):
|
||||||
"""List installed packages and their versions."""
|
"""List installed packages and their versions."""
|
||||||
installed_packages = get_installed_packages()
|
for package in get_installed_packages():
|
||||||
for package in installed_packages:
|
|
||||||
print(package.verbose_str() if verbose else str(package))
|
print(package.verbose_str() if verbose else str(package))
|
||||||
|
|
||||||
if freeze:
|
|
||||||
self.write_lock_packages(installed_packages)
|
|
||||||
|
|
||||||
def list_tags(self, url: str, limit: int = 10):
|
def list_tags(self, url: str, limit: int = 10):
|
||||||
print(f"Tags for {url}:")
|
print(f"Tags for {url}:")
|
||||||
for tag in get_repo_tags(url)[-1 * limit :]:
|
for tag in get_repo_tags(url)[-1 * limit :]:
|
||||||
print(tag)
|
print(tag)
|
||||||
|
|
||||||
def remove_packages(self, package_names: list[str], yes: bool = False):
|
def remove_packages(self, package_names: list[str]):
|
||||||
"""Remove installed packages and uodate lock."""
|
"""Remove installed packages and uodate lock."""
|
||||||
packages_to_remove = [
|
packages_to_remove = [
|
||||||
package
|
package
|
||||||
for package in get_installed_packages()
|
for package in get_installed_packages()
|
||||||
if (
|
if package.name in package_names
|
||||||
package.name in package_names
|
|
||||||
or package.url in package_names
|
|
||||||
or (
|
|
||||||
hasattr(package, "fork_component")
|
|
||||||
and getattr(package, "fork_component") in package_names
|
|
||||||
)
|
|
||||||
)
|
|
||||||
]
|
]
|
||||||
|
|
||||||
if package_names and not packages_to_remove:
|
|
||||||
print("No packages found to remove")
|
|
||||||
return
|
|
||||||
|
|
||||||
print("Packages to remove:")
|
|
||||||
for package in packages_to_remove:
|
|
||||||
print(package)
|
|
||||||
|
|
||||||
confirmed = yes or input("Remove listed packages? (y/N) ").lower() == "y"
|
|
||||||
if packages_to_remove and not confirmed:
|
|
||||||
return
|
|
||||||
|
|
||||||
remaining_packages = [
|
remaining_packages = [
|
||||||
package
|
package
|
||||||
for package in self.read_lock_packages()
|
for package in self.read_lock_packages()
|
||||||
@ -291,39 +197,13 @@ class Unhacs:
|
|||||||
self.write_lock_packages(remaining_packages)
|
self.write_lock_packages(remaining_packages)
|
||||||
|
|
||||||
|
|
||||||
def args_to_package(args) -> Package:
|
def main():
|
||||||
ignore_versions = (
|
# If the sub command is add package, it should pass the parsed arguments to the add_package function and return
|
||||||
{version for version in args.ignore_versions.split(",")}
|
parser = create_parser()
|
||||||
if args.ignore_versions
|
args = parser.parse_args()
|
||||||
else None
|
|
||||||
)
|
|
||||||
|
|
||||||
if args.type == Fork:
|
|
||||||
if not args.fork_branch:
|
|
||||||
raise InvalidArgumentsError(
|
|
||||||
"A branch must be provided for forked components"
|
|
||||||
)
|
|
||||||
if not args.fork_component:
|
|
||||||
raise InvalidArgumentsError(
|
|
||||||
"A component must be provided for forked components"
|
|
||||||
)
|
|
||||||
|
|
||||||
return Fork(
|
|
||||||
args.url,
|
|
||||||
branch_name=args.fork_branch,
|
|
||||||
fork_component=args.fork_component,
|
|
||||||
version=args.version,
|
|
||||||
ignored_versions=ignore_versions,
|
|
||||||
)
|
|
||||||
|
|
||||||
return args.type(args.url, version=args.version, ignored_versions=ignore_versions)
|
|
||||||
|
|
||||||
|
|
||||||
def main(argv: list[str] | None = None) -> int:
|
|
||||||
args = parse_args(argv or sys.argv[1:])
|
|
||||||
|
|
||||||
unhacs = Unhacs(args.config, args.package_file)
|
unhacs = Unhacs(args.config, args.package_file)
|
||||||
Package.git_tags = args.git_tags
|
Package.use_git = args.use_git
|
||||||
|
|
||||||
if args.subcommand == "add":
|
if args.subcommand == "add":
|
||||||
# If a file was provided, update all packages based on the lock file
|
# If a file was provided, update all packages based on the lock file
|
||||||
@ -331,40 +211,38 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
packages = read_lock_packages(args.file)
|
packages = read_lock_packages(args.file)
|
||||||
for package in packages:
|
for package in packages:
|
||||||
unhacs.add_package(
|
unhacs.add_package(
|
||||||
package,
|
package.url,
|
||||||
|
package.version,
|
||||||
update=True,
|
update=True,
|
||||||
|
package_type=package.package_type,
|
||||||
|
ignore_versions=package.ignored_versions,
|
||||||
)
|
)
|
||||||
elif args.url:
|
elif args.url:
|
||||||
try:
|
|
||||||
new_package = args_to_package(args)
|
|
||||||
except InvalidArgumentsError as e:
|
|
||||||
print(e)
|
|
||||||
return 1
|
|
||||||
try:
|
|
||||||
unhacs.add_package(
|
unhacs.add_package(
|
||||||
new_package,
|
args.url,
|
||||||
|
version=args.version,
|
||||||
update=args.update,
|
update=args.update,
|
||||||
|
package_type=args.type,
|
||||||
|
ignore_versions=(
|
||||||
|
{version for version in args.ignore_versions.split(",")}
|
||||||
|
if args.ignore_versions
|
||||||
|
else None
|
||||||
|
),
|
||||||
)
|
)
|
||||||
except DuplicatePackageError as e:
|
|
||||||
print(e)
|
|
||||||
return 1
|
|
||||||
else:
|
else:
|
||||||
print("Either a file or a URL must be provided")
|
raise ValueError("Either a file or a URL must be provided")
|
||||||
return 1
|
|
||||||
elif args.subcommand == "list":
|
elif args.subcommand == "list":
|
||||||
unhacs.list_packages(args.verbose, args.freeze)
|
unhacs.list_packages(args.verbose)
|
||||||
elif args.subcommand == "tags":
|
elif args.subcommand == "tags":
|
||||||
unhacs.list_tags(args.url, limit=args.limit)
|
unhacs.list_tags(args.url, limit=args.limit)
|
||||||
elif args.subcommand == "remove":
|
elif args.subcommand == "remove":
|
||||||
unhacs.remove_packages(args.packages, yes=args.yes)
|
unhacs.remove_packages(args.packages)
|
||||||
elif args.subcommand == "upgrade":
|
elif args.subcommand == "upgrade":
|
||||||
unhacs.upgrade_packages(args.packages, yes=args.yes)
|
unhacs.upgrade_packages(args.packages)
|
||||||
else:
|
else:
|
||||||
print(f"Command {args.subcommand} is not implemented")
|
print(f"Command {args.subcommand} is not implemented")
|
||||||
return 1
|
exit(1)
|
||||||
|
|
||||||
return 0
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
exit(main())
|
main()
|
||||||
|
353
unhacs/packages.py
Normal file
353
unhacs/packages.py
Normal file
@ -0,0 +1,353 @@
|
|||||||
|
import json
|
||||||
|
import shutil
|
||||||
|
import tempfile
|
||||||
|
from collections.abc import Iterable
|
||||||
|
from enum import StrEnum
|
||||||
|
from enum import auto
|
||||||
|
from io import BytesIO
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import cast
|
||||||
|
from zipfile import ZipFile
|
||||||
|
|
||||||
|
import requests
|
||||||
|
import yaml
|
||||||
|
|
||||||
|
from unhacs.git import get_ref_zip
|
||||||
|
from unhacs.git import get_repo_tags
|
||||||
|
|
||||||
|
DEFAULT_HASS_CONFIG_PATH: Path = Path(".")
|
||||||
|
DEFAULT_PACKAGE_FILE = Path("unhacs.yaml")
|
||||||
|
|
||||||
|
|
||||||
|
def extract_zip(zip_file: ZipFile, dest_dir: Path):
|
||||||
|
for info in zip_file.infolist():
|
||||||
|
if info.is_dir():
|
||||||
|
continue
|
||||||
|
file = Path(info.filename)
|
||||||
|
# Strip top directory from path
|
||||||
|
file = Path(*file.parts[1:])
|
||||||
|
path = dest_dir / file
|
||||||
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
with zip_file.open(info) as source, open(path, "wb") as dest:
|
||||||
|
dest.write(source.read())
|
||||||
|
|
||||||
|
|
||||||
|
class PackageType(StrEnum):
|
||||||
|
INTEGRATION = auto()
|
||||||
|
PLUGIN = auto()
|
||||||
|
|
||||||
|
|
||||||
|
class Package:
|
||||||
|
use_git = False
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
url: str,
|
||||||
|
version: str | None = None,
|
||||||
|
package_type: PackageType = PackageType.INTEGRATION,
|
||||||
|
ignored_versions: set[str] | None = None,
|
||||||
|
):
|
||||||
|
self.url = url
|
||||||
|
self.package_type = package_type
|
||||||
|
self.ignored_versions = ignored_versions or set()
|
||||||
|
|
||||||
|
parts = self.url.split("/")
|
||||||
|
self.owner = parts[-2]
|
||||||
|
self.name = parts[-1]
|
||||||
|
|
||||||
|
self.download_url: str | None = None
|
||||||
|
self.path: Path | None = None
|
||||||
|
|
||||||
|
if not version:
|
||||||
|
self.version, self.download_url = self.fetch_version_release()
|
||||||
|
else:
|
||||||
|
self.version = version
|
||||||
|
|
||||||
|
def __str__(self):
|
||||||
|
return f"{self.name} {self.version}"
|
||||||
|
|
||||||
|
def __eq__(self, other):
|
||||||
|
return self.url == other.url and self.version == other.version
|
||||||
|
|
||||||
|
def verbose_str(self):
|
||||||
|
return f"{self.name} {self.version} ({self.url})"
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def from_yaml(yaml: dict) -> "Package":
|
||||||
|
# Convert package_type to enum
|
||||||
|
package_type = yaml.pop("package_type")
|
||||||
|
if package_type and isinstance(package_type, str):
|
||||||
|
package_type = PackageType(package_type)
|
||||||
|
yaml["package_type"] = package_type
|
||||||
|
|
||||||
|
return Package(**yaml)
|
||||||
|
|
||||||
|
def to_yaml(self: "Package") -> dict:
|
||||||
|
return {
|
||||||
|
"url": self.url,
|
||||||
|
"version": self.version,
|
||||||
|
"package_type": str(self.package_type),
|
||||||
|
}
|
||||||
|
|
||||||
|
def add_ignored_version(self, version: str):
|
||||||
|
self.ignored_versions.add(version)
|
||||||
|
|
||||||
|
def _fetch_version_release_releases(
|
||||||
|
self, version: str | None = None
|
||||||
|
) -> tuple[str, str]:
|
||||||
|
# Fetch the releases from the GitHub API
|
||||||
|
response = requests.get(
|
||||||
|
f"https://api.github.com/repos/{self.owner}/{self.name}/releases"
|
||||||
|
)
|
||||||
|
response.raise_for_status()
|
||||||
|
releases = response.json()
|
||||||
|
|
||||||
|
if not releases:
|
||||||
|
raise ValueError(f"No releases found for package {self.name}")
|
||||||
|
|
||||||
|
# Default to latest
|
||||||
|
desired_release = releases[0]
|
||||||
|
|
||||||
|
# If a version is provided, check if it exists in the releases
|
||||||
|
if version:
|
||||||
|
for release in releases:
|
||||||
|
if release["tag_name"] == version:
|
||||||
|
desired_release = release
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
raise ValueError(f"Version {version} does not exist for this package")
|
||||||
|
|
||||||
|
version = cast(str, desired_release["tag_name"])
|
||||||
|
hacs_json = self.get_hacs_json(version)
|
||||||
|
|
||||||
|
# Based on type, if we have no hacs json, we can provide some possible paths for the download but won't know
|
||||||
|
# If a plugin:
|
||||||
|
# First, check in root/dist/ for a js file named the same as the repo or with "lovelace-" prefix removed
|
||||||
|
# Second will be looking for a realeases for a js file named the same name as the repo or with loveace- prefix removed
|
||||||
|
# Third will be looking in the root dir for a js file named the same as the repo or with loveace- prefix removed
|
||||||
|
# If an integration:
|
||||||
|
# We always use the zipball_url
|
||||||
|
|
||||||
|
download_url = None
|
||||||
|
if filename := hacs_json.get("filename"):
|
||||||
|
for asset in desired_release["assets"]:
|
||||||
|
if asset["name"] == filename:
|
||||||
|
download_url = cast(str, asset["browser_download_url"])
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
download_url = cast(str, desired_release["zipball_url"])
|
||||||
|
|
||||||
|
if not download_url:
|
||||||
|
raise ValueError("No filename found in hacs.json")
|
||||||
|
|
||||||
|
return version, download_url
|
||||||
|
|
||||||
|
def _fetch_version_release_git(self, version: str | None = None) -> tuple[str, str]:
|
||||||
|
tags = get_repo_tags(self.url)
|
||||||
|
if not tags:
|
||||||
|
raise ValueError(f"No tags found for package {self.name}")
|
||||||
|
if version and version not in tags:
|
||||||
|
raise ValueError(f"Version {version} does not exist for this package")
|
||||||
|
|
||||||
|
tags = [tag for tag in tags if tag not in self.ignored_versions]
|
||||||
|
if not version:
|
||||||
|
version = tags[-1]
|
||||||
|
|
||||||
|
return version, get_ref_zip(self.url, version)
|
||||||
|
|
||||||
|
def fetch_version_release(self, version: str | None = None) -> tuple[str, str]:
|
||||||
|
if self.use_git:
|
||||||
|
return self._fetch_version_release_git(version)
|
||||||
|
else:
|
||||||
|
return self._fetch_version_release_releases(version)
|
||||||
|
|
||||||
|
def fetch_versions(self) -> list[str]:
|
||||||
|
return get_repo_tags(self.url)
|
||||||
|
|
||||||
|
def get_hacs_json(self, version: str | None = None) -> dict:
|
||||||
|
version = version or self.version
|
||||||
|
response = requests.get(
|
||||||
|
f"https://raw.githubusercontent.com/{self.owner}/{self.name}/{version}/hacs.json"
|
||||||
|
)
|
||||||
|
|
||||||
|
if response.status_code == 404:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
response.raise_for_status()
|
||||||
|
return response.json()
|
||||||
|
|
||||||
|
def install_plugin(self, hass_config_path: Path):
|
||||||
|
# First, check in root/dist/ for a js file named the same as the repo or with "lovelace-" prefix removed
|
||||||
|
# Second will be looking for a realeases for a js file named the same name as the repo or with loveace- prefix removed
|
||||||
|
# Third will be looking in the root dir for a js file named the same as the repo or with loveace- prefix removed
|
||||||
|
# If none of these are found, raise an error
|
||||||
|
# If a file is found, write it to www/js/<filename>.js and write a file www/js/<filename>-unhacs.txt with the
|
||||||
|
# serialized package
|
||||||
|
|
||||||
|
valid_filenames: Iterable[str]
|
||||||
|
if filename := self.get_hacs_json().get("filename"):
|
||||||
|
valid_filenames = (cast(str, filename),)
|
||||||
|
else:
|
||||||
|
valid_filenames = (
|
||||||
|
f"{self.name.removeprefix('lovelace-')}.js",
|
||||||
|
f"{self.name}.js",
|
||||||
|
f"{self.name}-umd.js",
|
||||||
|
f"{self.name}-bundle.js",
|
||||||
|
)
|
||||||
|
|
||||||
|
def real_get(filename) -> requests.Response:
|
||||||
|
plugin = requests.get(
|
||||||
|
f"https://raw.githubusercontent.com/{self.owner}/{self.version}/dist/{filename}"
|
||||||
|
)
|
||||||
|
if plugin.status_code == 404:
|
||||||
|
plugin = requests.get(
|
||||||
|
f"https://github.com/{self.owner}/{self.name}/releases/download/{self.version}/{filename}"
|
||||||
|
)
|
||||||
|
if plugin.status_code == 404:
|
||||||
|
plugin = requests.get(
|
||||||
|
f"https://raw.githubusercontent.com/{self.owner}/{self.version}/{filename}"
|
||||||
|
)
|
||||||
|
|
||||||
|
plugin.raise_for_status()
|
||||||
|
return plugin
|
||||||
|
|
||||||
|
for filename in valid_filenames:
|
||||||
|
try:
|
||||||
|
plugin = real_get(filename)
|
||||||
|
break
|
||||||
|
except requests.HTTPError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise ValueError(f"No valid filename found for package {self.name}")
|
||||||
|
|
||||||
|
js_path = hass_config_path / "www" / "js"
|
||||||
|
js_path.mkdir(parents=True, exist_ok=True)
|
||||||
|
js_path.joinpath(filename).write_text(plugin.text)
|
||||||
|
|
||||||
|
yaml.dump(self.to_yaml(), js_path.joinpath(f"{filename}-unhacs.yaml").open("w"))
|
||||||
|
|
||||||
|
def install_integration(self, hass_config_path: Path):
|
||||||
|
zipball_url = f"https://codeload.github.com/{self.owner}/{self.name}/zip/refs/tags/{self.version}"
|
||||||
|
response = requests.get(zipball_url)
|
||||||
|
response.raise_for_status()
|
||||||
|
|
||||||
|
with tempfile.TemporaryDirectory(prefix="unhacs-") as tempdir:
|
||||||
|
tmpdir = Path(tempdir)
|
||||||
|
extract_zip(ZipFile(BytesIO(response.content)), tmpdir)
|
||||||
|
|
||||||
|
source, dest = None, None
|
||||||
|
for custom_component in tmpdir.glob("custom_components/*"):
|
||||||
|
source = custom_component
|
||||||
|
dest = hass_config_path / "custom_components" / custom_component.name
|
||||||
|
break
|
||||||
|
else:
|
||||||
|
hacs_json = json.loads((tmpdir / "hacs.json").read_text())
|
||||||
|
if hacs_json.get("content_in_root"):
|
||||||
|
source = tmpdir
|
||||||
|
dest = hass_config_path / "custom_components" / self.name
|
||||||
|
|
||||||
|
if not source or not dest:
|
||||||
|
raise ValueError("No custom_components directory found")
|
||||||
|
|
||||||
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
shutil.rmtree(dest, ignore_errors=True)
|
||||||
|
shutil.move(source, dest)
|
||||||
|
|
||||||
|
yaml.dump(self.to_yaml(), dest.joinpath("unhacs.yaml").open("w"))
|
||||||
|
|
||||||
|
def install(self, hass_config_path: Path):
|
||||||
|
if self.package_type == PackageType.PLUGIN:
|
||||||
|
self.install_plugin(hass_config_path)
|
||||||
|
elif self.package_type == PackageType.INTEGRATION:
|
||||||
|
self.install_integration(hass_config_path)
|
||||||
|
else:
|
||||||
|
raise NotImplementedError(f"Unknown package type {self.package_type}")
|
||||||
|
|
||||||
|
def uninstall(self, hass_config_path: Path) -> bool:
|
||||||
|
if self.path:
|
||||||
|
if self.path.is_dir():
|
||||||
|
shutil.rmtree(self.path)
|
||||||
|
else:
|
||||||
|
self.path.unlink()
|
||||||
|
self.path.with_name(f"{self.path.name}-unhacs.yaml").unlink()
|
||||||
|
return True
|
||||||
|
|
||||||
|
installed_package = self.installed_package(hass_config_path)
|
||||||
|
if installed_package:
|
||||||
|
installed_package.uninstall(hass_config_path)
|
||||||
|
return True
|
||||||
|
|
||||||
|
return False
|
||||||
|
|
||||||
|
def installed_package(self, hass_config_path: Path) -> "Package|None":
|
||||||
|
for custom_component in (hass_config_path / "custom_components").glob("*"):
|
||||||
|
unhacs = custom_component / "unhacs.yaml"
|
||||||
|
if unhacs.exists():
|
||||||
|
installed_package = Package.from_yaml(yaml.safe_load(unhacs.open()))
|
||||||
|
installed_package.path = custom_component
|
||||||
|
if (
|
||||||
|
installed_package.name == self.name
|
||||||
|
and installed_package.url == self.url
|
||||||
|
):
|
||||||
|
return installed_package
|
||||||
|
|
||||||
|
for js_unhacs in (hass_config_path / "www" / "js").glob("*-unhacs.yaml"):
|
||||||
|
installed_package = Package.from_yaml(yaml.safe_load(js_unhacs.open()))
|
||||||
|
installed_package.path = js_unhacs.with_name(
|
||||||
|
js_unhacs.name.removesuffix("-unhacs.yaml")
|
||||||
|
)
|
||||||
|
if (
|
||||||
|
installed_package.name == self.name
|
||||||
|
and installed_package.url == self.url
|
||||||
|
):
|
||||||
|
return installed_package
|
||||||
|
|
||||||
|
return None
|
||||||
|
|
||||||
|
def is_update(self, hass_config_path: Path) -> bool:
|
||||||
|
installed_package = self.installed_package(hass_config_path)
|
||||||
|
return installed_package is None or installed_package.version != self.version
|
||||||
|
|
||||||
|
def get_latest(self) -> "Package":
|
||||||
|
package = self.to_yaml()
|
||||||
|
package.pop("version")
|
||||||
|
return Package(**package)
|
||||||
|
|
||||||
|
|
||||||
|
def get_installed_packages(
|
||||||
|
hass_config_path: Path = DEFAULT_HASS_CONFIG_PATH,
|
||||||
|
) -> list[Package]:
|
||||||
|
packages = []
|
||||||
|
|
||||||
|
# Integration packages
|
||||||
|
for custom_component in (hass_config_path / "custom_components").glob("*"):
|
||||||
|
unhacs = custom_component / "unhacs.yaml"
|
||||||
|
if unhacs.exists():
|
||||||
|
package = Package.from_yaml(yaml.safe_load(unhacs.open()))
|
||||||
|
package.path = custom_component
|
||||||
|
packages.append(package)
|
||||||
|
|
||||||
|
# Plugin packages
|
||||||
|
for js_unhacs in (hass_config_path / "www" / "js").glob("*-unhacs.yaml"):
|
||||||
|
package = Package.from_yaml(yaml.safe_load(js_unhacs.open()))
|
||||||
|
package.path = js_unhacs.with_name(js_unhacs.name.removesuffix("-unhacs.yaml"))
|
||||||
|
packages.append(package)
|
||||||
|
|
||||||
|
return packages
|
||||||
|
|
||||||
|
|
||||||
|
# Read a list of Packages from a text file in the plain text format "URL version name"
|
||||||
|
def read_lock_packages(package_file: Path = DEFAULT_PACKAGE_FILE) -> list[Package]:
|
||||||
|
if package_file.exists():
|
||||||
|
return [
|
||||||
|
Package.from_yaml(p)
|
||||||
|
for p in yaml.safe_load(package_file.open())["packages"]
|
||||||
|
]
|
||||||
|
return []
|
||||||
|
|
||||||
|
|
||||||
|
# Write a list of Packages to a text file in the format URL version name
|
||||||
|
def write_lock_packages(
|
||||||
|
packages: Iterable[Package], package_file: Path = DEFAULT_PACKAGE_FILE
|
||||||
|
):
|
||||||
|
yaml.dump({"packages": [p.to_yaml() for p in packages]}, package_file.open("w"))
|
@ -1,82 +0,0 @@
|
|||||||
from collections.abc import Iterable
|
|
||||||
from pathlib import Path
|
|
||||||
from typing import cast
|
|
||||||
|
|
||||||
import yaml
|
|
||||||
|
|
||||||
from unhacs.packages.common import Package
|
|
||||||
from unhacs.packages.common import PackageType
|
|
||||||
from unhacs.packages.fork import Fork
|
|
||||||
from unhacs.packages.integration import Integration
|
|
||||||
from unhacs.packages.plugin import Plugin
|
|
||||||
from unhacs.packages.theme import Theme
|
|
||||||
from unhacs.utils import DEFAULT_HASS_CONFIG_PATH
|
|
||||||
from unhacs.utils import DEFAULT_PACKAGE_FILE
|
|
||||||
|
|
||||||
|
|
||||||
def from_yaml(data: dict | Path | str) -> Package:
|
|
||||||
if isinstance(data, Path):
|
|
||||||
data = yaml.safe_load(data.open())
|
|
||||||
elif isinstance(data, str):
|
|
||||||
data = yaml.safe_load(data)
|
|
||||||
|
|
||||||
data = cast(dict, data)
|
|
||||||
|
|
||||||
# Convert package_type to enum
|
|
||||||
package_type = data.pop("package_type", None)
|
|
||||||
if package_type and isinstance(package_type, str):
|
|
||||||
package_type = PackageType(package_type)
|
|
||||||
|
|
||||||
url = data.pop("url")
|
|
||||||
|
|
||||||
return {
|
|
||||||
PackageType.INTEGRATION: Integration,
|
|
||||||
PackageType.PLUGIN: Plugin,
|
|
||||||
PackageType.THEME: Theme,
|
|
||||||
PackageType.FORK: Fork,
|
|
||||||
}[package_type](url, **data)
|
|
||||||
|
|
||||||
|
|
||||||
def get_installed_packages(
|
|
||||||
hass_config_path: Path = DEFAULT_HASS_CONFIG_PATH,
|
|
||||||
package_types: Iterable[PackageType] = (
|
|
||||||
PackageType.INTEGRATION,
|
|
||||||
PackageType.FORK,
|
|
||||||
PackageType.PLUGIN,
|
|
||||||
PackageType.THEME,
|
|
||||||
),
|
|
||||||
) -> list[Package]:
|
|
||||||
# Integration packages
|
|
||||||
packages: list[Package] = []
|
|
||||||
|
|
||||||
if PackageType.INTEGRATION in package_types:
|
|
||||||
packages.extend(Integration.find_installed(hass_config_path))
|
|
||||||
|
|
||||||
if PackageType.FORK in package_types:
|
|
||||||
packages.extend(Fork.find_installed(hass_config_path))
|
|
||||||
|
|
||||||
# Plugin packages
|
|
||||||
if PackageType.PLUGIN in package_types:
|
|
||||||
packages.extend(Plugin.find_installed(hass_config_path))
|
|
||||||
|
|
||||||
# Theme packages
|
|
||||||
if PackageType.THEME in package_types:
|
|
||||||
packages.extend(Theme.find_installed(hass_config_path))
|
|
||||||
|
|
||||||
return packages
|
|
||||||
|
|
||||||
|
|
||||||
# Read a list of Packages from a text file in the plain text format "URL version name"
|
|
||||||
def read_lock_packages(package_file: Path = DEFAULT_PACKAGE_FILE) -> list[Package]:
|
|
||||||
if package_file.exists():
|
|
||||||
with package_file.open() as f:
|
|
||||||
return [from_yaml(p) for p in yaml.safe_load(f)["packages"]]
|
|
||||||
return []
|
|
||||||
|
|
||||||
|
|
||||||
# Write a list of Packages to a text file in the format URL version name
|
|
||||||
def write_lock_packages(
|
|
||||||
packages: Iterable[Package], package_file: Path = DEFAULT_PACKAGE_FILE
|
|
||||||
):
|
|
||||||
with open(package_file, "w") as f:
|
|
||||||
yaml.dump({"packages": [p.to_yaml() for p in packages]}, f)
|
|
@ -1,225 +0,0 @@
|
|||||||
import shutil
|
|
||||||
from enum import StrEnum
|
|
||||||
from enum import auto
|
|
||||||
from pathlib import Path
|
|
||||||
from typing import Any
|
|
||||||
from typing import cast
|
|
||||||
|
|
||||||
import requests
|
|
||||||
import yaml
|
|
||||||
|
|
||||||
from unhacs.git import get_repo_tags
|
|
||||||
|
|
||||||
|
|
||||||
class PackageType(StrEnum):
|
|
||||||
INTEGRATION = auto()
|
|
||||||
PLUGIN = auto()
|
|
||||||
FORK = auto()
|
|
||||||
THEME = auto()
|
|
||||||
|
|
||||||
|
|
||||||
class Package:
|
|
||||||
git_tags = False
|
|
||||||
package_type: PackageType
|
|
||||||
|
|
||||||
other_fields: list[str] = []
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
url: str,
|
|
||||||
version: str | None = None,
|
|
||||||
ignored_versions: set[str] | None = None,
|
|
||||||
):
|
|
||||||
self.url = url
|
|
||||||
self.ignored_versions = ignored_versions or set()
|
|
||||||
|
|
||||||
parts = self.url.split("/")
|
|
||||||
self.owner = parts[-2]
|
|
||||||
self.name = parts[-1]
|
|
||||||
|
|
||||||
self.path: Path | None = None
|
|
||||||
|
|
||||||
if not version:
|
|
||||||
self.version = self.fetch_version_release()
|
|
||||||
else:
|
|
||||||
self.version = version
|
|
||||||
|
|
||||||
def __str__(self):
|
|
||||||
return f"{self.package_type}: {self.name} {self.version}"
|
|
||||||
|
|
||||||
def __eq__(self, other):
|
|
||||||
return all(
|
|
||||||
(
|
|
||||||
self.same(other),
|
|
||||||
self.version == other.version,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
|
|
||||||
def same(self, other):
|
|
||||||
fields = list(["url"] + self.other_fields)
|
|
||||||
|
|
||||||
return all((getattr(self, field) == getattr(other, field) for field in fields))
|
|
||||||
|
|
||||||
def __hash__(self):
|
|
||||||
fields = list(["url"] + self.other_fields)
|
|
||||||
|
|
||||||
return hash(tuple(getattr(self, field) for field in fields))
|
|
||||||
|
|
||||||
def verbose_str(self):
|
|
||||||
return f"{str(self)} ({self.url})"
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def from_yaml(cls, data: dict | Path | str) -> "Package":
|
|
||||||
if isinstance(data, Path):
|
|
||||||
with data.open() as f:
|
|
||||||
data = yaml.safe_load(f)
|
|
||||||
elif isinstance(data, str):
|
|
||||||
data = yaml.safe_load(data)
|
|
||||||
|
|
||||||
data = cast(dict, data)
|
|
||||||
|
|
||||||
if (package_type := data.pop("package_type")) != cls.package_type:
|
|
||||||
raise ValueError(
|
|
||||||
f"Invalid package_type ({package_type}) for this class {cls.package_type}"
|
|
||||||
)
|
|
||||||
|
|
||||||
return cls(data.pop("url"), **data)
|
|
||||||
|
|
||||||
def to_yaml(self, dest: Path | None = None) -> dict:
|
|
||||||
data: dict[str, Any] = {
|
|
||||||
"url": self.url,
|
|
||||||
"version": self.version,
|
|
||||||
"package_type": str(self.package_type),
|
|
||||||
}
|
|
||||||
|
|
||||||
if self.ignored_versions:
|
|
||||||
data["ignored_versions"] = self.ignored_versions
|
|
||||||
|
|
||||||
for field in self.other_fields:
|
|
||||||
if hasattr(self, field):
|
|
||||||
data[field] = getattr(self, field)
|
|
||||||
|
|
||||||
if dest:
|
|
||||||
with dest.open("w") as f:
|
|
||||||
yaml.dump(self.to_yaml(), f)
|
|
||||||
|
|
||||||
return data
|
|
||||||
|
|
||||||
def add_ignored_version(self, version: str):
|
|
||||||
self.ignored_versions.add(version)
|
|
||||||
|
|
||||||
def _fetch_version_release_releases(self, version: str | None = None) -> str:
|
|
||||||
# Fetch the releases from the GitHub API
|
|
||||||
response = requests.get(
|
|
||||||
f"https://api.github.com/repos/{self.owner}/{self.name}/releases"
|
|
||||||
)
|
|
||||||
response.raise_for_status()
|
|
||||||
releases = response.json()
|
|
||||||
|
|
||||||
if not releases:
|
|
||||||
raise ValueError(f"No releases found for package {self.name}")
|
|
||||||
|
|
||||||
# Default to latest
|
|
||||||
desired_release = releases[0]
|
|
||||||
|
|
||||||
# If a version is provided, check if it exists in the releases
|
|
||||||
if version:
|
|
||||||
for release in releases:
|
|
||||||
if release["tag_name"] == version:
|
|
||||||
desired_release = release
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
raise ValueError(f"Version {version} does not exist for this package")
|
|
||||||
|
|
||||||
return cast(str, desired_release["tag_name"])
|
|
||||||
|
|
||||||
def _fetch_version_release_git(self, version: str | None = None) -> str:
|
|
||||||
tags = get_repo_tags(self.url)
|
|
||||||
if not tags:
|
|
||||||
raise ValueError(f"No tags found for package {self.name}")
|
|
||||||
if version and version not in tags:
|
|
||||||
raise ValueError(f"Version {version} does not exist for this package")
|
|
||||||
|
|
||||||
tags = [tag for tag in tags if tag not in self.ignored_versions]
|
|
||||||
if not version:
|
|
||||||
version = tags[-1]
|
|
||||||
|
|
||||||
return version
|
|
||||||
|
|
||||||
def fetch_version_release(self, version: str | None = None) -> str:
|
|
||||||
if self.git_tags:
|
|
||||||
return self._fetch_version_release_git(version)
|
|
||||||
else:
|
|
||||||
return self._fetch_version_release_releases(version)
|
|
||||||
|
|
||||||
def _fetch_versions(self) -> list[str]:
|
|
||||||
return get_repo_tags(self.url)
|
|
||||||
|
|
||||||
def get_hacs_json(self, version: str | None = None) -> dict:
|
|
||||||
"""Fetches the hacs.json file for the package."""
|
|
||||||
version = version or self.version
|
|
||||||
response = requests.get(
|
|
||||||
f"https://raw.githubusercontent.com/{self.owner}/{self.name}/{version}/hacs.json"
|
|
||||||
)
|
|
||||||
|
|
||||||
if response.status_code == 404:
|
|
||||||
return {}
|
|
||||||
|
|
||||||
response.raise_for_status()
|
|
||||||
return response.json()
|
|
||||||
|
|
||||||
def install(self, hass_config_path: Path):
|
|
||||||
raise NotImplementedError()
|
|
||||||
|
|
||||||
@property
|
|
||||||
def unhacs_path(self) -> Path | None:
|
|
||||||
if self.path is None:
|
|
||||||
return None
|
|
||||||
|
|
||||||
return self.path / "unhacs.yaml"
|
|
||||||
|
|
||||||
def uninstall(self, hass_config_path: Path) -> bool:
|
|
||||||
"""Uninstalls the package if it is installed, returning True if it was uninstalled."""
|
|
||||||
if not self.path:
|
|
||||||
if installed_package := self.installed_package(hass_config_path):
|
|
||||||
installed_package.uninstall(hass_config_path)
|
|
||||||
return True
|
|
||||||
|
|
||||||
return False
|
|
||||||
|
|
||||||
if self.path.is_dir():
|
|
||||||
shutil.rmtree(self.path)
|
|
||||||
else:
|
|
||||||
self.path.unlink()
|
|
||||||
if self.unhacs_path and self.unhacs_path.exists():
|
|
||||||
self.unhacs_path.unlink()
|
|
||||||
|
|
||||||
return True
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def get_install_dir(cls, hass_config_path: Path) -> Path:
|
|
||||||
raise NotImplementedError()
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def find_installed(cls, hass_config_path: Path) -> list["Package"]:
|
|
||||||
raise NotImplementedError()
|
|
||||||
|
|
||||||
def installed_package(self, hass_config_path: Path) -> "Package|None":
|
|
||||||
"""Returns the installed package if it exists, otherwise None."""
|
|
||||||
for package in self.find_installed(hass_config_path):
|
|
||||||
if self.same(package):
|
|
||||||
return package
|
|
||||||
|
|
||||||
return None
|
|
||||||
|
|
||||||
def is_update(self, hass_config_path: Path) -> bool:
|
|
||||||
"""Returns True if the package is not installed or the installed version is different from the latest."""
|
|
||||||
installed_package = self.installed_package(hass_config_path)
|
|
||||||
return installed_package is None or installed_package.version != self.version
|
|
||||||
|
|
||||||
def get_latest(self) -> "Package":
|
|
||||||
"""Returns a new Package representing the latest version of this package."""
|
|
||||||
package = self.to_yaml()
|
|
||||||
package.pop("version")
|
|
||||||
package.pop("package_type")
|
|
||||||
return self.__class__(package.pop("url"), **package)
|
|
@ -1,106 +0,0 @@
|
|||||||
import json
|
|
||||||
import shutil
|
|
||||||
import tempfile
|
|
||||||
from io import BytesIO
|
|
||||||
from pathlib import Path
|
|
||||||
from zipfile import ZipFile
|
|
||||||
|
|
||||||
import requests
|
|
||||||
import yaml
|
|
||||||
|
|
||||||
from unhacs.git import get_branch_zip
|
|
||||||
from unhacs.git import get_latest_sha
|
|
||||||
from unhacs.git import get_sha_zip
|
|
||||||
from unhacs.packages import PackageType
|
|
||||||
from unhacs.packages.common import Package
|
|
||||||
from unhacs.packages.integration import Integration
|
|
||||||
from unhacs.utils import extract_zip
|
|
||||||
|
|
||||||
|
|
||||||
class Fork(Integration):
|
|
||||||
other_fields = ["fork_component", "branch_name"]
|
|
||||||
package_type = PackageType.FORK
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
url: str,
|
|
||||||
fork_component: str,
|
|
||||||
branch_name: str,
|
|
||||||
version: str | None = None,
|
|
||||||
ignored_versions: set[str] | None = None,
|
|
||||||
):
|
|
||||||
self.fork_component = fork_component
|
|
||||||
self.branch_name = branch_name
|
|
||||||
|
|
||||||
super().__init__(
|
|
||||||
url,
|
|
||||||
version=version,
|
|
||||||
ignored_versions=ignored_versions,
|
|
||||||
)
|
|
||||||
|
|
||||||
def __str__(self):
|
|
||||||
return f"{self.package_type}: {self.fork_component} ({self.owner}/{self.name}@{self.branch_name}) {self.version}"
|
|
||||||
|
|
||||||
def fetch_version_release(self, version: str | None = None) -> str:
|
|
||||||
if version:
|
|
||||||
return version
|
|
||||||
|
|
||||||
return get_latest_sha(self.url, self.branch_name)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def find_installed(cls, hass_config_path: Path) -> list[Package]:
|
|
||||||
packages: list[Package] = []
|
|
||||||
|
|
||||||
for custom_component in cls.get_install_dir(hass_config_path).glob("*"):
|
|
||||||
unhacs = custom_component / "unhacs.yaml"
|
|
||||||
if unhacs.exists():
|
|
||||||
data = yaml.safe_load(unhacs.read_text())
|
|
||||||
if data["package_type"] != "fork":
|
|
||||||
continue
|
|
||||||
package = cls.from_yaml(data)
|
|
||||||
package.path = custom_component
|
|
||||||
packages.append(package)
|
|
||||||
|
|
||||||
return packages
|
|
||||||
|
|
||||||
def install(self, hass_config_path: Path) -> None:
|
|
||||||
"""Installs the integration from hass fork."""
|
|
||||||
if self.version:
|
|
||||||
zipball_url = get_sha_zip(self.url, self.version)
|
|
||||||
else:
|
|
||||||
zipball_url = get_branch_zip(self.url, self.branch_name)
|
|
||||||
|
|
||||||
response = requests.get(zipball_url)
|
|
||||||
response.raise_for_status()
|
|
||||||
|
|
||||||
with tempfile.TemporaryDirectory(prefix="unhacs-") as tempdir:
|
|
||||||
tmpdir = Path(tempdir)
|
|
||||||
extract_zip(ZipFile(BytesIO(response.content)), tmpdir)
|
|
||||||
|
|
||||||
source, dest = None, None
|
|
||||||
source = tmpdir / "homeassistant" / "components" / self.fork_component
|
|
||||||
if not source.exists() or not source.is_dir():
|
|
||||||
raise ValueError(
|
|
||||||
f"Could not find {self.fork_component} in {self.url}@{self.version}"
|
|
||||||
)
|
|
||||||
|
|
||||||
# Add version to manifest
|
|
||||||
manifest_file = source / "manifest.json"
|
|
||||||
manifest: dict[str, str]
|
|
||||||
with manifest_file.open("r") as f:
|
|
||||||
manifest = json.load(f)
|
|
||||||
manifest["version"] = "0.0.0"
|
|
||||||
with manifest_file.open("w") as f:
|
|
||||||
json.dump(manifest, f)
|
|
||||||
|
|
||||||
dest = self.get_install_dir(hass_config_path) / source.name
|
|
||||||
|
|
||||||
if not source or not dest:
|
|
||||||
raise ValueError("No custom_components directory found")
|
|
||||||
|
|
||||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
||||||
shutil.rmtree(dest, ignore_errors=True)
|
|
||||||
shutil.move(source, dest)
|
|
||||||
self.path = dest
|
|
||||||
|
|
||||||
self.to_yaml(self.unhacs_path)
|
|
@ -1,81 +0,0 @@
|
|||||||
import json
|
|
||||||
import shutil
|
|
||||||
import tempfile
|
|
||||||
from io import BytesIO
|
|
||||||
from pathlib import Path
|
|
||||||
from zipfile import ZipFile
|
|
||||||
|
|
||||||
import requests
|
|
||||||
import yaml
|
|
||||||
|
|
||||||
from unhacs.git import get_tag_zip
|
|
||||||
from unhacs.packages import Package
|
|
||||||
from unhacs.packages import PackageType
|
|
||||||
from unhacs.utils import extract_zip
|
|
||||||
|
|
||||||
|
|
||||||
class Integration(Package):
|
|
||||||
package_type = PackageType.INTEGRATION
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
url: str,
|
|
||||||
version: str | None = None,
|
|
||||||
ignored_versions: set[str] | None = None,
|
|
||||||
):
|
|
||||||
super().__init__(
|
|
||||||
url,
|
|
||||||
version=version,
|
|
||||||
ignored_versions=ignored_versions,
|
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def get_install_dir(cls, hass_config_path: Path) -> Path:
|
|
||||||
return hass_config_path / "custom_components"
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def find_installed(cls, hass_config_path: Path) -> list[Package]:
|
|
||||||
packages: list[Package] = []
|
|
||||||
|
|
||||||
for custom_component in cls.get_install_dir(hass_config_path).glob("*"):
|
|
||||||
unhacs = custom_component / "unhacs.yaml"
|
|
||||||
if unhacs.exists():
|
|
||||||
data = yaml.safe_load(unhacs.read_text())
|
|
||||||
if data["package_type"] == "fork":
|
|
||||||
continue
|
|
||||||
package = cls.from_yaml(data)
|
|
||||||
package.path = custom_component
|
|
||||||
packages.append(package)
|
|
||||||
|
|
||||||
return packages
|
|
||||||
|
|
||||||
def install(self, hass_config_path: Path) -> None:
|
|
||||||
"""Installs the integration package."""
|
|
||||||
zipball_url = get_tag_zip(self.url, self.version)
|
|
||||||
response = requests.get(zipball_url)
|
|
||||||
response.raise_for_status()
|
|
||||||
|
|
||||||
with tempfile.TemporaryDirectory(prefix="unhacs-") as tempdir:
|
|
||||||
tmpdir = Path(tempdir)
|
|
||||||
extract_zip(ZipFile(BytesIO(response.content)), tmpdir)
|
|
||||||
|
|
||||||
source, dest = None, None
|
|
||||||
for custom_component in tmpdir.glob("custom_components/*"):
|
|
||||||
source = custom_component
|
|
||||||
dest = self.get_install_dir(hass_config_path) / custom_component.name
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
hacs_json = json.loads((tmpdir / "hacs.json").read_text())
|
|
||||||
if hacs_json.get("content_in_root"):
|
|
||||||
source = tmpdir
|
|
||||||
dest = self.get_install_dir(hass_config_path) / self.name
|
|
||||||
|
|
||||||
if not source or not dest:
|
|
||||||
raise ValueError("No custom_components directory found")
|
|
||||||
|
|
||||||
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
||||||
shutil.rmtree(dest, ignore_errors=True)
|
|
||||||
shutil.move(source, dest)
|
|
||||||
self.path = dest
|
|
||||||
|
|
||||||
self.to_yaml(self.unhacs_path)
|
|
@ -1,95 +0,0 @@
|
|||||||
from pathlib import Path
|
|
||||||
from typing import cast
|
|
||||||
|
|
||||||
import requests
|
|
||||||
|
|
||||||
from unhacs.packages import Package
|
|
||||||
from unhacs.packages import PackageType
|
|
||||||
|
|
||||||
|
|
||||||
class Plugin(Package):
|
|
||||||
package_type = PackageType.PLUGIN
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
url: str,
|
|
||||||
version: str | None = None,
|
|
||||||
ignored_versions: set[str] | None = None,
|
|
||||||
):
|
|
||||||
super().__init__(
|
|
||||||
url,
|
|
||||||
version=version,
|
|
||||||
ignored_versions=ignored_versions,
|
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def get_install_dir(cls, hass_config_path: Path) -> Path:
|
|
||||||
return hass_config_path / "www" / "js"
|
|
||||||
|
|
||||||
@property
|
|
||||||
def unhacs_path(self) -> Path | None:
|
|
||||||
if self.path is None:
|
|
||||||
return None
|
|
||||||
|
|
||||||
return self.path.with_name(f"{self.path.name}-unhacs.yaml")
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def find_installed(cls, hass_config_path: Path) -> list["Package"]:
|
|
||||||
packages: list[Package] = []
|
|
||||||
|
|
||||||
for js_unhacs in cls.get_install_dir(hass_config_path).glob("*-unhacs.yaml"):
|
|
||||||
package = cls.from_yaml(js_unhacs)
|
|
||||||
package.path = js_unhacs.with_name(
|
|
||||||
js_unhacs.name.removesuffix("-unhacs.yaml")
|
|
||||||
)
|
|
||||||
packages.append(package)
|
|
||||||
|
|
||||||
return packages
|
|
||||||
|
|
||||||
def install(self, hass_config_path: Path) -> None:
|
|
||||||
"""Installs the plugin package."""
|
|
||||||
|
|
||||||
valid_filenames: list[str]
|
|
||||||
if filename := self.get_hacs_json().get("filename"):
|
|
||||||
valid_filenames = [cast(str, filename)]
|
|
||||||
else:
|
|
||||||
valid_filenames = [
|
|
||||||
f"{self.name.removeprefix('lovelace-')}.js",
|
|
||||||
f"{self.name}.js",
|
|
||||||
f"{self.name}-umd.js",
|
|
||||||
f"{self.name}-bundle.js",
|
|
||||||
]
|
|
||||||
|
|
||||||
def real_get(filename) -> requests.Response | None:
|
|
||||||
urls = [
|
|
||||||
f"https://raw.githubusercontent.com/{self.owner}/{self.name}/{self.version}/dist/{filename}",
|
|
||||||
f"https://github.com/{self.owner}/{self.name}/releases/download/{self.version}/{filename}",
|
|
||||||
f"https://raw.githubusercontent.com/{self.owner}/{self.name}/{self.version}/{filename}",
|
|
||||||
]
|
|
||||||
|
|
||||||
for url in urls:
|
|
||||||
plugin = requests.get(url)
|
|
||||||
|
|
||||||
if int(plugin.status_code / 100) == 4:
|
|
||||||
continue
|
|
||||||
|
|
||||||
plugin.raise_for_status()
|
|
||||||
|
|
||||||
return plugin
|
|
||||||
|
|
||||||
return None
|
|
||||||
|
|
||||||
for filename in valid_filenames:
|
|
||||||
plugin = real_get(filename)
|
|
||||||
if plugin:
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
raise ValueError(f"No valid filename found for package {self.name}")
|
|
||||||
|
|
||||||
js_path = self.get_install_dir(hass_config_path)
|
|
||||||
js_path.mkdir(parents=True, exist_ok=True)
|
|
||||||
|
|
||||||
self.path = js_path.joinpath(filename)
|
|
||||||
self.path.write_text(plugin.text)
|
|
||||||
|
|
||||||
self.to_yaml(self.unhacs_path)
|
|
@ -1,63 +0,0 @@
|
|||||||
from pathlib import Path
|
|
||||||
from typing import cast
|
|
||||||
|
|
||||||
import requests
|
|
||||||
|
|
||||||
from unhacs.packages import Package
|
|
||||||
from unhacs.packages import PackageType
|
|
||||||
|
|
||||||
|
|
||||||
class Theme(Package):
|
|
||||||
package_type = PackageType.THEME
|
|
||||||
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
url: str,
|
|
||||||
version: str | None = None,
|
|
||||||
ignored_versions: set[str] | None = None,
|
|
||||||
):
|
|
||||||
super().__init__(
|
|
||||||
url,
|
|
||||||
version=version,
|
|
||||||
ignored_versions=ignored_versions,
|
|
||||||
)
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def get_install_dir(cls, hass_config_path: Path) -> Path:
|
|
||||||
return hass_config_path / "themes"
|
|
||||||
|
|
||||||
@property
|
|
||||||
def unhacs_path(self) -> Path | None:
|
|
||||||
if self.path is None:
|
|
||||||
return None
|
|
||||||
|
|
||||||
return self.path.with_name(f"{self.path.name}.unhacs")
|
|
||||||
|
|
||||||
@classmethod
|
|
||||||
def find_installed(cls, hass_config_path: Path) -> list["Package"]:
|
|
||||||
packages: list[Package] = []
|
|
||||||
|
|
||||||
for js_unhacs in cls.get_install_dir(hass_config_path).glob("*.unhacs"):
|
|
||||||
package = cls.from_yaml(js_unhacs)
|
|
||||||
package.path = js_unhacs.with_name(js_unhacs.name.removesuffix(".unhacs"))
|
|
||||||
packages.append(package)
|
|
||||||
|
|
||||||
return packages
|
|
||||||
|
|
||||||
def install(self, hass_config_path: Path) -> None:
|
|
||||||
"""Install theme yaml."""
|
|
||||||
filename = self.get_hacs_json().get("filename")
|
|
||||||
if not filename:
|
|
||||||
raise ValueError(f"No filename found for theme {self.name}")
|
|
||||||
|
|
||||||
filename = cast(str, filename)
|
|
||||||
url = f"https://raw.githubusercontent.com/{self.owner}/{self.name}/{self.version}/themes/{filename}"
|
|
||||||
theme = requests.get(url)
|
|
||||||
theme.raise_for_status()
|
|
||||||
|
|
||||||
themes_path = self.get_install_dir(hass_config_path)
|
|
||||||
themes_path.mkdir(parents=True, exist_ok=True)
|
|
||||||
self.path = themes_path.joinpath(filename)
|
|
||||||
self.path.write_text(theme.text)
|
|
||||||
|
|
||||||
self.to_yaml(self.unhacs_path)
|
|
@ -1,21 +0,0 @@
|
|||||||
from pathlib import Path
|
|
||||||
from zipfile import ZipFile
|
|
||||||
|
|
||||||
DEFAULT_HASS_CONFIG_PATH: Path = Path(".")
|
|
||||||
DEFAULT_PACKAGE_FILE = Path("unhacs.yaml")
|
|
||||||
|
|
||||||
|
|
||||||
def extract_zip(zip_file: ZipFile, dest_dir: Path) -> Path:
|
|
||||||
"""Extract a zip file to a directory."""
|
|
||||||
for info in zip_file.infolist():
|
|
||||||
if info.is_dir():
|
|
||||||
continue
|
|
||||||
file = Path(info.filename)
|
|
||||||
# Strip top directory from path
|
|
||||||
file = Path(*file.parts[1:])
|
|
||||||
path = dest_dir / file
|
|
||||||
path.parent.mkdir(parents=True, exist_ok=True)
|
|
||||||
with zip_file.open(info) as source, open(path, "wb") as dest:
|
|
||||||
dest.write(source.read())
|
|
||||||
|
|
||||||
return dest_dir
|
|
Loading…
Reference in New Issue
Block a user