Split scaffolding script (#26832)

* Add scaffolding split

* Add second config flow method

script/scaffold/__main__.py

@@ -1,9 +1,42 @@
 """Validate manifests."""
+import argparse
 from pathlib import Path
 import subprocess
 import sys
 
-from . import gather_info, generate, error, model
+from . import gather_info, generate, error
+from .const import COMPONENT_DIR
+
+
+TEMPLATES = [
+    p.name for p in (Path(__file__).parent / "templates").glob("*") if p.is_dir()
+]
+
+
+def valid_integration(integration):
+    """Test if it's a valid integration."""
+    if not (COMPONENT_DIR / integration).exists():
+        raise argparse.ArgumentTypeError(
+            f"The integration {integration} does not exist."
+        )
+
+    return integration
+
+
+def get_arguments() -> argparse.Namespace:
+    """Get parsed passed in arguments."""
+    parser = argparse.ArgumentParser(description="Home Assistant Scaffolder")
+    parser.add_argument("template", type=str, choices=TEMPLATES)
+    parser.add_argument(
+        "--develop", action="store_true", help="Automatically fill in info"
+    )
+    parser.add_argument(
+        "--integration", type=valid_integration, help="Integration to target."
+    )
+
+    arguments = parser.parse_args()
+
+    return arguments
 
 
 def main():
@@ -12,29 +45,22 @@ def main():
         print("Run from project root")
         return 1
 
-    print("Creating a new integration for Home Assistant.")
+    args = get_arguments()
 
-    if "--develop" in sys.argv:
+    info = gather_info.gather_info(args)
-        print("Running in developer mode. Automatically filling in info.")
-        print()
 
-        info = model.Info(
+    generate.generate(args.template, info)
-            domain="develop",
-            name="Develop Hub",
-            codeowner="@developer",
-            requirement="aiodevelop==1.2.3",
-        )
-    else:
-        try:
-            info = gather_info.gather_info()
-        except error.ExitApp as err:
-            print()
-            print(err.reason)
-            return err.exit_code
 
-    generate.generate(info)
+    # If creating new integration, create config flow too
+    if args.template == "integration":
+        if info.authentication or not info.discoverable:
+            template = "config_flow"
+        else:
+            template = "config_flow_discovery"
 
-    print("Running hassfest to pick up new codeowner and config flow.")
+        generate.generate(template, info)
+
+    print("Running hassfest to pick up new information.")
     subprocess.run("python -m script.hassfest", shell=True)
     print()
 
@@ -47,10 +73,15 @@ def main():
         return 1
     print()
 
-    print(f"Successfully created the {info.domain} integration!")
+    print(f"Done!")
 
     return 0
 
 
 if __name__ == "__main__":
-    sys.exit(main())
+    try:
+        sys.exit(main())
+    except error.ExitApp as err:
+        print()
+        print(f"Fatal Error: {err.reason}")
+        sys.exit(err.exit_code)

script/scaffold/docs.py

@@ -0,0 +1,22 @@
+"""Print links to relevant docs."""
+from .model import Info
+
+
+def print_relevant_docs(template: str, info: Info) -> None:
+    """Print relevant docs."""
+    if template == "integration":
+        print(
+            f"""
+Your integration has been created at {info.integration_dir} . Next step is to fill in the blanks for the code marked with TODO.
+
+For a breakdown of each file, check the developer documentation at:
+https://developers.home-assistant.io/docs/en/creating_integration_file_structure.html
+"""
+        )
+
+    elif template == "config_flow":
+        print(
+            f"""
+The config flow has been added to the {info.domain} integration. Next step is to fill in the blanks for the code marked with TODO.
+"""
+        )

script/scaffold/error.py

@@ -4,7 +4,7 @@
 class ExitApp(Exception):
     """Exception to indicate app should exit."""
 
-    def __init__(self, reason, exit_code):
+    def __init__(self, reason, exit_code=1):
         """Initialize the exit app exception."""
         self.reason = reason
         self.exit_code = exit_code

script/scaffold/gather_info.py

@@ -1,4 +1,6 @@
 """Gather info for scaffolding."""
+import json
+
 from homeassistant.util import slugify
 
 from .const import COMPONENT_DIR
@@ -9,49 +11,142 @@ from .error import ExitApp
 CHECK_EMPTY = ["Cannot be empty", lambda value: value]
 
 
-FIELDS = {
+def gather_info(arguments) -> Info:
-    "domain": {
+    """Gather info."""
-        "prompt": "What is the domain?",
+    existing = arguments.template != "integration"
-        "validators": [
+
-            CHECK_EMPTY,
+    if arguments.develop:
-            [
+        print("Running in developer mode. Automatically filling in info.")
-                "Domains cannot contain spaces or special characters.",
+        print()
-                lambda value: value == slugify(value),
+
-            ],
+    if existing:
-            [
+        if arguments.develop:
-                "There already is an integration with this domain.",
+            return _load_existing_integration("develop")
-                lambda value: not (COMPONENT_DIR / value).exists(),
+
-            ],
+        if arguments.integration:
-        ],
+            return _load_existing_integration(arguments.integration)
-    },
+
-    "name": {
+        return gather_existing_integration()
-        "prompt": "What is the name of your integration?",
+
-        "validators": [CHECK_EMPTY],
+    if arguments.develop:
-    },
+        return Info(
-    "codeowner": {
+            domain="develop",
-        "prompt": "What is your GitHub handle?",
+            name="Develop Hub",
-        "validators": [
+            codeowner="@developer",
-            CHECK_EMPTY,
+            requirement="aiodevelop==1.2.3",
-            [
+        )
-                'GitHub handles need to start with an "@"',
+
-                lambda value: value.startswith("@"),
+    return gather_new_integration()
-            ],
-        ],
-    },
-    "requirement": {
-        "prompt": "What PyPI package and version do you depend on? Leave blank for none.",
-        "validators": [
-            ["Versions should be pinned using '=='.", lambda value: "==" in value]
-        ],
-    },
-}
 
 
-def gather_info() -> Info:
+def gather_new_integration() -> Info:
+    """Gather info about new integration from user."""
+    return Info(
+        **_gather_info(
+            {
+                "domain": {
+                    "prompt": "What is the domain?",
+                    "validators": [
+                        CHECK_EMPTY,
+                        [
+                            "Domains cannot contain spaces or special characters.",
+                            lambda value: value == slugify(value),
+                        ],
+                        [
+                            "There already is an integration with this domain.",
+                            lambda value: not (COMPONENT_DIR / value).exists(),
+                        ],
+                    ],
+                },
+                "name": {
+                    "prompt": "What is the name of your integration?",
+                    "validators": [CHECK_EMPTY],
+                },
+                "codeowner": {
+                    "prompt": "What is your GitHub handle?",
+                    "validators": [
+                        CHECK_EMPTY,
+                        [
+                            'GitHub handles need to start with an "@"',
+                            lambda value: value.startswith("@"),
+                        ],
+                    ],
+                },
+                "requirement": {
+                    "prompt": "What PyPI package and version do you depend on? Leave blank for none.",
+                    "validators": [
+                        [
+                            "Versions should be pinned using '=='.",
+                            lambda value: not value or "==" in value,
+                        ]
+                    ],
+                },
+                "authentication": {
+                    "prompt": "Does Home Assistant need the user to authenticate to control the device/service? (yes/no)",
+                    "default": "yes",
+                    "validators": [
+                        [
+                            "Type either 'yes' or 'no'",
+                            lambda value: value in ("yes", "no"),
+                        ]
+                    ],
+                    "convertor": lambda value: value == "yes",
+                },
+                "discoverable": {
+                    "prompt": "Is the device/service discoverable on the local network? (yes/no)",
+                    "default": "no",
+                    "validators": [
+                        [
+                            "Type either 'yes' or 'no'",
+                            lambda value: value in ("yes", "no"),
+                        ]
+                    ],
+                    "convertor": lambda value: value == "yes",
+                },
+            }
+        )
+    )
+
+
+def gather_existing_integration() -> Info:
+    """Gather info about existing integration from user."""
+    answers = _gather_info(
+        {
+            "domain": {
+                "prompt": "What is the domain?",
+                "validators": [
+                    CHECK_EMPTY,
+                    [
+                        "Domains cannot contain spaces or special characters.",
+                        lambda value: value == slugify(value),
+                    ],
+                    [
+                        "This integration does not exist.",
+                        lambda value: (COMPONENT_DIR / value).exists(),
+                    ],
+                ],
+            }
+        }
+    )
+
+    return _load_existing_integration(answers["domain"])
+
+
+def _load_existing_integration(domain) -> Info:
+    """Load an existing integration."""
+    if not (COMPONENT_DIR / domain).exists():
+        raise ExitApp("Integration does not exist", 1)
+
+    manifest = json.loads((COMPONENT_DIR / domain / "manifest.json").read_text())
+
+    return Info(domain=domain, name=manifest["name"])
+
+
+def _gather_info(fields) -> dict:
     """Gather info from user."""
     answers = {}
 
-    for key, info in FIELDS.items():
+    for key, info in fields.items():
         hint = None
         while key not in answers:
             if hint is not None:
@@ -60,11 +155,18 @@ def gather_info() -> Info:
 
             try:
                 print()
-                value = input(info["prompt"] + "\n> ")
+                msg = info["prompt"]
+                if "default" in info:
+                    msg += f" [{info['default']}]"
+                value = input(f"{msg}\n> ")
             except (KeyboardInterrupt, EOFError):
                 raise ExitApp("Interrupted!", 1)
 
             value = value.strip()
+
+            if value == "" and "default" in info:
+                value = info["default"]
+
             hint = None
 
             for validator_hint, validator in info["validators"]:
@@ -73,7 +175,9 @@ def gather_info() -> Info:
                     break
 
             if hint is None:
+                if "convertor" in info:
+                    value = info["convertor"](value)
                 answers[key] = value
 
     print()
-    return Info(**answers)
+    return answers

script/scaffold/generate.py

@@ -1,8 +1,7 @@
 """Generate an integration."""
-import json
 from pathlib import Path
 
-from .const import COMPONENT_DIR, TESTS_DIR
+from .error import ExitApp
 from .model import Info
 
 TEMPLATE_DIR = Path(__file__).parent / "templates"
@@ -10,38 +9,113 @@ TEMPLATE_INTEGRATION = TEMPLATE_DIR / "integration"
 TEMPLATE_TESTS = TEMPLATE_DIR / "tests"
 
 
-def generate(info: Info) -> None:
+def generate(template: str, info: Info) -> None:
-    """Generate an integration."""
+    """Generate a template."""
-    print(f"Generating the {info.domain} integration...")
+    _validate(template, info)
-    integration_dir = COMPONENT_DIR / info.domain
-    test_dir = TESTS_DIR / info.domain
-
-    replaces = {
-        "NEW_DOMAIN": info.domain,
-        "NEW_NAME": info.name,
-        "NEW_CODEOWNER": info.codeowner,
-        # Special case because we need to keep the list empty if there is none.
-        '"MANIFEST_NEW_REQUIREMENT"': (
-            json.dumps(info.requirement) if info.requirement else ""
-        ),
-    }
-
-    for src_dir, target_dir in (
-        (TEMPLATE_INTEGRATION, integration_dir),
-        (TEMPLATE_TESTS, test_dir),
-    ):
-        # Guard making it for test purposes.
-        if not target_dir.exists():
-            target_dir.mkdir()
-
-        for source_file in src_dir.glob("**/*"):
-            content = source_file.read_text()
-
-            for to_search, to_replace in replaces.items():
-                content = content.replace(to_search, to_replace)
-
-            target_file = target_dir / source_file.relative_to(src_dir)
-            print(f"Writing {target_file}")
-            target_file.write_text(content)
 
+    print(f"Scaffolding {template} for the {info.domain} integration...")
+    _ensure_tests_dir_exists(info)
+    _generate(TEMPLATE_DIR / template / "integration", info.integration_dir, info)
+    _generate(TEMPLATE_DIR / template / "tests", info.tests_dir, info)
+    _custom_tasks(template, info)
     print()
+
+
+def _validate(template, info):
+    """Validate we can run this task."""
+    if template == "config_flow":
+        if (info.integration_dir / "config_flow.py").exists():
+            raise ExitApp(f"Integration {info.domain} already has a config flow.")
+
+
+def _generate(src_dir, target_dir, info: Info) -> None:
+    """Generate an integration."""
+    replaces = {"NEW_DOMAIN": info.domain, "NEW_NAME": info.name}
+
+    if not target_dir.exists():
+        target_dir.mkdir()
+
+    for source_file in src_dir.glob("**/*"):
+        content = source_file.read_text()
+
+        for to_search, to_replace in replaces.items():
+            content = content.replace(to_search, to_replace)
+
+        target_file = target_dir / source_file.relative_to(src_dir)
+        print(f"Writing {target_file}")
+        target_file.write_text(content)
+
+
+def _ensure_tests_dir_exists(info: Info) -> None:
+    """Ensure a test dir exists."""
+    if info.tests_dir.exists():
+        return
+
+    info.tests_dir.mkdir()
+    print(f"Writing {info.tests_dir / '__init__.py'}")
+    (info.tests_dir / "__init__.py").write_text(
+        f'"""Tests for the {info.name} integration."""\n'
+    )
+
+
+def _custom_tasks(template, info) -> None:
+    """Handle custom tasks for templates."""
+    if template == "integration":
+        changes = {"codeowners": [info.codeowner]}
+
+        if info.requirement:
+            changes["requirements"] = [info.requirement]
+
+        info.update_manifest(**changes)
+
+    if template == "config_flow":
+        info.update_manifest(config_flow=True)
+        info.update_strings(
+            config={
+                "title": info.name,
+                "step": {
+                    "user": {"title": "Connect to the device", "data": {"host": "Host"}}
+                },
+                "error": {
+                    "cannot_connect": "Failed to connect, please try again",
+                    "invalid_auth": "Invalid authentication",
+                    "unknown": "Unexpected error",
+                },
+                "abort": {"already_configured": "Device is already configured"},
+            }
+        )
+
+    if template == "config_flow_discovery":
+        info.update_manifest(config_flow=True)
+        info.update_strings(
+            config={
+                "title": info.name,
+                "step": {
+                    "confirm": {
+                        "title": info.name,
+                        "description": f"Do you want to set up {info.name}?",
+                    }
+                },
+                "abort": {
+                    "single_instance_allowed": f"Only a single configuration of {info.name} is possible.",
+                    "no_devices_found": f"No {info.name} devices found on the network.",
+                },
+            }
+        )
+
+    if template in ("config_flow", "config_flow_discovery"):
+        init_file = info.integration_dir / "__init__.py"
+        init_file.write_text(
+            init_file.read_text()
+            + """
+
+async def async_setup_entry(hass, entry):
+    \"\"\"Set up a config entry for NEW_NAME.\"\"\"
+    # TODO forward the entry for each platform that you want to set up.
+    # hass.async_create_task(
+    #     hass.config_entries.async_forward_entry_setup(entry, "media_player")
+    # )
+
+    return True
+"""
+        )

script/scaffold/model.py

@@ -1,6 +1,11 @@
 """Models for scaffolding."""
+import json
+from pathlib import Path
+
 import attr
 
+from .const import COMPONENT_DIR, TESTS_DIR
+
 
 @attr.s
 class Info:
@@ -8,5 +13,49 @@ class Info:
 
     domain: str = attr.ib()
     name: str = attr.ib()
-    codeowner: str = attr.ib()
+    codeowner: str = attr.ib(default=None)
-    requirement: str = attr.ib()
+    requirement: str = attr.ib(default=None)
+    authentication: str = attr.ib(default=None)
+    discoverable: str = attr.ib(default=None)
+
+    @property
+    def integration_dir(self) -> Path:
+        """Return directory if integration."""
+        return COMPONENT_DIR / self.domain
+
+    @property
+    def tests_dir(self) -> Path:
+        """Return test directory."""
+        return TESTS_DIR / self.domain
+
+    @property
+    def manifest_path(self) -> Path:
+        """Path to the manifest."""
+        return COMPONENT_DIR / self.domain / "manifest.json"
+
+    def manifest(self) -> dict:
+        """Return integration manifest."""
+        return json.loads(self.manifest_path.read_text())
+
+    def update_manifest(self, **kwargs) -> None:
+        """Update the integration manifest."""
+        print(f"Updating {self.domain} manifest: {kwargs}")
+        self.manifest_path.write_text(
+            json.dumps({**self.manifest(), **kwargs}, indent=2)
+        )
+
+    @property
+    def strings_path(self) -> Path:
+        """Path to the strings."""
+        return COMPONENT_DIR / self.domain / "strings.json"
+
+    def strings(self) -> dict:
+        """Return integration strings."""
+        if not self.strings_path.exists():
+            return {}
+        return json.loads(self.strings_path.read_text())
+
+    def update_strings(self, **kwargs) -> None:
+        """Update the integration strings."""
+        print(f"Updating {self.domain} strings: {list(kwargs)}")
+        self.strings_path.write_text(json.dumps({**self.strings(), **kwargs}, indent=2))

script/scaffold/templates/config_flow/integration/config_flow.py

@@ -3,10 +3,9 @@ import logging
 
 import voluptuous as vol
 
-from homeassistant import core, config_entries
+from homeassistant import core, config_entries, exceptions
 
 from .const import DOMAIN  # pylint:disable=unused-import
-from .error import CannotConnect, InvalidAuth
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -33,7 +32,7 @@ class DomainConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
     """Handle a config flow for NEW_NAME."""
 
     VERSION = 1
-    # TODO pick one of the available connection classes
+    # TODO pick one of the available connection classes in homeassistant/config_entries.py
     CONNECTION_CLASS = config_entries.CONN_CLASS_UNKNOWN
 
     async def async_step_user(self, user_input=None):
@@ -55,3 +54,11 @@ class DomainConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
         return self.async_show_form(
             step_id="user", data_schema=DATA_SCHEMA, errors=errors
         )
+
+
+class CannotConnect(exceptions.HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(exceptions.HomeAssistantError):
+    """Error to indicate there is invalid auth."""

script/scaffold/templates/config_flow/tests/test_config_flow.py

@@ -3,7 +3,7 @@ from unittest.mock import patch
 
 from homeassistant import config_entries, setup
 from homeassistant.components.NEW_DOMAIN.const import DOMAIN
-from homeassistant.components.NEW_DOMAIN.error import CannotConnect, InvalidAuth
+from homeassistant.components.NEW_DOMAIN.config_flow import CannotConnect, InvalidAuth
 
 from tests.common import mock_coro
 

script/scaffold/templates/config_flow_discovery/integration/config_flow.py

@@ -0,0 +1,18 @@
+"""Config flow for NEW_NAME."""
+import my_pypi_dependency
+
+from homeassistant.helpers import config_entry_flow
+from homeassistant import config_entries
+from .const import DOMAIN
+
+
+async def _async_has_devices(hass) -> bool:
+    """Return if there are devices that can be discovered."""
+    # TODO Check if there are any devices that can be discovered in the network.
+    devices = await hass.async_add_executor_job(my_pypi_dependency.discover)
+    return len(devices) > 0
+
+
+config_entry_flow.register_discovery_flow(
+    DOMAIN, "NEW_NAME", _async_has_devices, config_entries.CONN_CLASS_UNKNOWN
+)

script/scaffold/templates/integration/__init__.py

@@ -1,19 +0,0 @@
-"""The NEW_NAME integration."""
-
-from .const import DOMAIN
-
-
-async def async_setup(hass, config):
-    """Set up the NEW_NAME integration."""
-    hass.data[DOMAIN] = config.get(DOMAIN, {})
-    return True
-
-
-async def async_setup_entry(hass, entry):
-    """Set up a config entry for NEW_NAME."""
-    # TODO forward the entry for each platform that you want to set up.
-    # hass.async_create_task(
-    #     hass.config_entries.async_forward_entry_setup(entry, "media_player")
-    # )
-
-    return True

script/scaffold/templates/integration/error.py

@@ -1,10 +0,0 @@
-"""Errors for the NEW_NAME integration."""
-from homeassistant.exceptions import HomeAssistantError
-
-
-class CannotConnect(HomeAssistantError):
-    """Error to indicate we cannot connect."""
-
-
-class InvalidAuth(HomeAssistantError):
-    """Error to indicate there is invalid auth."""

script/scaffold/templates/integration/integration/__init__.py

@@ -0,0 +1,12 @@
+"""The NEW_NAME integration."""
+import voluptuous as vol
+
+from .const import DOMAIN
+
+
+CONFIG_SCHEMA = vol.Schema({vol.Optional(DOMAIN): {}})
+
+
+async def async_setup(hass, config):
+    """Set up the NEW_NAME integration."""
+    return True

script/scaffold/templates/integration/integration/manifest.json

@@ -1,11 +1,11 @@
 {
   "domain": "NEW_DOMAIN",
   "name": "NEW_NAME",
-  "config_flow": true,
+  "config_flow": false,
   "documentation": "https://www.home-assistant.io/components/NEW_DOMAIN",
-  "requirements": ["MANIFEST_NEW_REQUIREMENT"],
+  "requirements": [],
   "ssdp": {},
   "homekit": {},
   "dependencies": [],
-  "codeowners": ["NEW_CODEOWNER"]
+  "codeowners": []
 }

script/scaffold/templates/integration/strings.json

@@ -1,21 +0,0 @@
-{
-  "config": {
-    "title": "NEW_NAME",
-    "step": {
-      "user": {
-        "title": "Connect to the device",
-        "data": {
-          "host": "Host"
-        }
-      }
-    },
-    "error": {
-      "cannot_connect": "Failed to connect, please try again",
-      "invalid_auth": "Invalid authentication",
-      "unknown": "Unexpected error"
-    },
-    "abort": {
-      "already_configured": "Device is already configured"
-    }
-  }
-}

script/scaffold/templates/tests/__init__.py

@@ -1 +0,0 @@
-"""Tests for the NEW_NAME integration."""