innocuous-symmetry/core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Integrations v2.1: Virtual integrations (#80613)

Browse Source
This commit is contained in:
Franck Nijhof
2022-10-21 05:09:06 +02:00
committed by GitHub
parent 6c23de94e1
commit bb287dd0ed
66 changed files with 2968 additions and 2720 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
script/hassfest/__main__.py
View File

@@ -20,7 +20,6 @@ from . import (
requirements,
services,
ssdp,
supported_brands,
translations,
usb,
zeroconf,
@@ -39,7 +38,6 @@ INTEGRATION_PLUGINS = [
requirements,
services,
ssdp,
supported_brands,
translations,
usb,
zeroconf,

5
script/hassfest/codeowners.py
View File

@@ -47,7 +47,10 @@ def generate_and_validate(integrations: dict[str, Integration], config: Config):
for domain in sorted(integrations):
integration = integrations[domain]
if not integration.manifest:
if (
not integration.manifest
or integration.manifest.get("integration_type") == "virtual"
):
continue
codeowners = integration.manifest["codeowners"]

16
script/hassfest/config_flow.py
View File

@@ -171,14 +171,24 @@ def _generate_integrations(
integration = integrations[domain]
if integration.integration_type in ("entity", "system"):
continue
metadata["config_flow"] = integration.config_flow
metadata["iot_class"] = integration.iot_class
metadata["integration_type"] = integration.integration_type
if integration.translated_name:
result["translated_name"].add(domain)
else:
metadata["name"] = integration.name
metadata["integration_type"] = integration.integration_type
if integration.integration_type == "virtual":
if integration.supported_by:
metadata["supported_by"] = integration.supported_by
if integration.iot_standard:
metadata["iot_standard"] = integration.iot_standard
else:
metadata["config_flow"] = integration.config_flow
if integration.iot_class:
metadata["iot_class"] = integration.iot_class
if integration.integration_type == "helper":
result["helper"][domain] = metadata
else:

45
script/hassfest/manifest.py
View File

@@ -2,6 +2,7 @@
from __future__ import annotations
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
from awesomeversion import (
@@ -158,7 +159,7 @@ def verify_wildcard(value: str):
return value
MANIFEST_SCHEMA = vol.Schema(
INTEGRATION_MANIFEST_SCHEMA = vol.Schema(
{
vol.Required("domain"): str,
vol.Required("name"): str,
@@ -254,14 +255,32 @@ MANIFEST_SCHEMA = vol.Schema(
vol.Optional("loggers"): [str],
vol.Optional("disabled"): str,
vol.Optional("iot_class"): vol.In(SUPPORTED_IOT_CLASSES),
vol.Optional("supported_brands"): vol.Schema({str: str}),
}
)
CUSTOM_INTEGRATION_MANIFEST_SCHEMA = MANIFEST_SCHEMA.extend(
VIRTUAL_INTEGRATION_MANIFEST_SCHEMA = vol.Schema(
{
vol.Required("domain"): str,
vol.Required("name"): str,
vol.Required("integration_type"): "virtual",
vol.Exclusive("iot_standard", "virtual_integration"): vol.Any(
"homekit", "zigbee", "zwave"
),
vol.Exclusive("supported_by", "virtual_integration"): str,
}
)
def manifest_schema(value: dict[str, Any]) -> vol.Schema:
"""Validate integration manifest."""
if value.get("integration_type") == "virtual":
return VIRTUAL_INTEGRATION_MANIFEST_SCHEMA(value)
return INTEGRATION_MANIFEST_SCHEMA(value)
CUSTOM_INTEGRATION_MANIFEST_SCHEMA = INTEGRATION_MANIFEST_SCHEMA.extend(
{
vol.Optional("version"): vol.All(str, verify_version),
vol.Remove("supported_brands"): dict,
}
)
@@ -284,7 +303,7 @@ def validate_manifest(integration: Integration, core_components_dir: Path) -> No
try:
if integration.core:
MANIFEST_SCHEMA(integration.manifest)
manifest_schema(integration.manifest)
else:
CUSTOM_INTEGRATION_MANIFEST_SCHEMA(integration.manifest)
except vol.Invalid as err:
@@ -312,15 +331,19 @@ def validate_manifest(integration: Integration, core_components_dir: Path) -> No
if (
integration.manifest["domain"] not in NO_IOT_CLASS
and "iot_class" not in integration.manifest
and integration.manifest.get("integration_type") != "virtual"
):
integration.add_error("manifest", "Domain is missing an IoT Class")
for domain, _name in integration.manifest.get("supported_brands", {}).items():
if (core_components_dir / domain).exists():
integration.add_warning(
"manifest",
f"Supported brand domain {domain} collides with built-in core integration",
)
if (
integration.manifest.get("integration_type") == "virtual"
and (supported_by := integration.manifest.get("supported_by"))
and not (core_components_dir / supported_by).exists()
):
integration.add_error(
"manifest",
"Virtual integration points to non-existing supported_by integration",
)
if not integration.core:
validate_version(integration)

20
script/hassfest/model.py
View File

@@ -109,11 +109,12 @@ class Integration:
continue
init = fil / "__init__.py"
if not init.exists():
manifest = fil / "manifest.json"
if not init.exists() and not manifest.exists():
print(
f"Warning: {init} missing, skipping directory. "
"If this is your development environment, "
"you can safely delete this folder."
f"Warning: {init} and manifest.json missing, "
"skipping directory. If this is your development "
"environment, you can safely delete this folder."
)
continue
@@ -170,9 +171,9 @@ class Integration:
return self.manifest.get("dependencies", [])
@property
def supported_brands(self) -> dict[str]:
"""Return dict of supported brands."""
return self.manifest.get("supported_brands", {})
def supported_by(self) -> str:
"""Return the integration supported by this virtual integration."""
return self.manifest.get("supported_by", {})
@property
def integration_type(self) -> str:
@@ -184,6 +185,11 @@ class Integration:
"""Return the integration IoT Class."""
return self.manifest.get("iot_class")
@property
def iot_standard(self) -> str:
"""Return the IoT standard supported by this virtual integration."""
return self.manifest.get("iot_standard", {})
def add_error(self, *args: Any, **kwargs: Any) -> None:
"""Add an error."""
self.errors.append(Error(*args, **kwargs))

54
script/hassfest/supported_brands.py
View File

@@ -1,54 +0,0 @@
"""Generate supported_brands data."""
from __future__ import annotations
import black
from .model import Config, Integration
from .serializer import to_string
BASE = """
\"\"\"Automatically generated by hassfest.
To update, run python3 -m script.hassfest
\"\"\"
HAS_SUPPORTED_BRANDS = {}
""".strip()
def generate_and_validate(integrations: dict[str, Integration], config: Config) -> str:
"""Validate and generate supported_brands data."""
brands = [
domain
for domain, integration in sorted(integrations.items())
if integration.supported_brands
]
return black.format_str(BASE.format(to_string(brands)), mode=black.Mode())
def validate(integrations: dict[str, Integration], config: Config) -> None:
"""Validate supported_brands data."""
supported_brands_path = config.root / "homeassistant/generated/supported_brands.py"
config.cache["supported_brands"] = content = generate_and_validate(
integrations, config
)
if config.specific_integrations:
return
if supported_brands_path.read_text(encoding="utf-8") != content:
config.add_error(
"supported_brands",
"File supported_brands.py is not up to date. Run python3 -m script.hassfest",
fixable=True,
)
def generate(integrations: dict[str, Integration], config: Config):
"""Generate supported_brands data."""
supported_brands_path = config.root / "homeassistant/generated/supported_brands.py"
supported_brands_path.write_text(
f"{config.cache['supported_brands']}", encoding="utf-8"
)