Merge pull request #68 from alandtse/api_service

feat: add tesla_custom.api service

custom_components/tesla_custom/__init__.py

@@ -35,6 +35,7 @@
     MIN_SCAN_INTERVAL,
     PLATFORMS,
 )
+from .services import async_setup_services, async_unload_services
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -133,6 +134,8 @@ async def async_setup_entry(hass, config_entry):
     # Because users can have multiple accounts, we always create a new session so they have separate cookies
     async_client = httpx.AsyncClient(headers={USER_AGENT: SERVER_SOFTWARE}, timeout=60)
     email = config_entry.title
+    if not hass.data[DOMAIN]:
+        async_setup_services(hass)
     if email in hass.data[DOMAIN] and CONF_SCAN_INTERVAL in hass.data[DOMAIN][email]:
         scan_interval = hass.data[DOMAIN][email][CONF_SCAN_INTERVAL]
         hass.config_entries.async_update_entry(
@@ -233,6 +236,8 @@ async def async_unload_entry(hass, config_entry) -> bool:
     if unload_ok:
         hass.data[DOMAIN].pop(config_entry.entry_id)
         _LOGGER.debug("Unloaded entry for %s", username)
+        if not hass.data[DOMAIN]:
+            async_unload_services(hass)
         return True
     return False
 

custom_components/tesla_custom/const.py

@@ -37,3 +37,7 @@
 AUTH_CALLBACK_NAME = "auth:tesla:callback"
 AUTH_PROXY_PATH = "/auth/tesla/proxy"
 AUTH_PROXY_NAME = "auth:tesla:proxy"
+
+ATTR_PARAMETERS = "parameters"
+ATTR_PATH_VARS = "path_vars"
+SERVICE_API = "api"

custom_components/tesla_custom/services.py

@@ -0,0 +1,95 @@
+"""Support for Tesla services.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+import logging
+from homeassistant.const import CONF_EMAIL
+
+from teslajsonpy import Controller
+from homeassistant.core import callback
+from homeassistant.helpers import config_validation as cv
+import voluptuous as vol
+
+from homeassistant.const import ATTR_COMMAND
+from .const import (
+    ATTR_PARAMETERS,
+    ATTR_PATH_VARS,
+    DOMAIN,
+    SERVICE_API,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+API_SCHEMA = vol.Schema(
+    {
+        vol.Optional(CONF_EMAIL): vol.All(cv.string, vol.Length(min=1)),
+        vol.Required(ATTR_COMMAND, default=""): vol.All(cv.string, vol.Length(min=1)),
+        vol.Optional(ATTR_PARAMETERS, default={}): dict,
+    }
+)
+
+
+@callback
+def async_setup_services(hass) -> None:
+    """Set up services for Tesla integration."""
+
+    async def async_call_tesla_service(service_call) -> None:
+        """Call correct Tesla service."""
+        service = service_call.service
+
+        if service == SERVICE_API:
+            await api(service_call)
+
+    hass.services.async_register(
+        DOMAIN,
+        SERVICE_API,
+        async_call_tesla_service,
+        schema=API_SCHEMA,
+    )
+
+    async def api(call):
+        """Handle api service request.
+
+        Arguments
+            call.CONF_EMAIL {str: ""} -- email, optional
+            call.ATTR_COMMAND {str: ""} -- Command
+            call.ATTR_PARAMETERS {dict:} -- Parameters dictionary
+
+        Returns
+            bool -- True if api called successfully
+
+        """
+        _LOGGER.debug("call %s", call)
+        service_data = call.data
+        email = service_data.get(CONF_EMAIL, "")
+
+        if len(hass.config_entries.async_entries(DOMAIN)) > 1 and not email:
+            raise ValueError("Email address missing")
+        controller: Controller = None
+        for entry in hass.config_entries.async_entries(DOMAIN):
+            if (
+                len(hass.config_entries.async_entries(DOMAIN)) > 1
+                and entry.title != email
+            ):
+                continue
+            controller = hass.data[DOMAIN].get(entry.entry_id)["coordinator"].controller
+        if controller is None:
+            raise ValueError(f"No Tesla controllers found for email {email}")
+        command = call.data.get(ATTR_COMMAND)
+        parameters: dict = call.data.get(ATTR_PARAMETERS, {})
+        _LOGGER.debug(
+            "Service api called with email: %s command: %s parameters: %s",
+            email,
+            command,
+            parameters,
+        )
+        path_vars = parameters.pop(ATTR_PATH_VARS)
+        return await controller.api(name=command, path_vars=path_vars, **parameters)
+
+
+@callback
+def async_unload_services(hass) -> None:
+    """Unload Tesla services."""
+    hass.services.async_remove(DOMAIN, SERVICE_API)

custom_components/tesla_custom/services.yaml

@@ -0,0 +1,31 @@
+#  SPDX-License-Identifier: Apache-2.0
+api:
+  # Description of the service
+  description: Run an API command using controller.api. https://teslajsonpy.readthedocs.io/en/latest/teslajsonpy/teslajsonpy.html#teslajsonpy.Controller.api
+  # Different fields that your service accepts
+  fields:
+    # Key of the field
+    email:
+      # Description of the field
+      description: Email address (optional if only one account)
+      # Example value that can be passed for this field
+      example: "elon@tesla.com"
+      required: false
+      selector:
+        text:
+    command:
+      # Description of the field
+      description: Command to run. See https://github.com/zabuldon/teslajsonpy/blob/master/teslajsonpy/endpoints.json
+      # Example value that can be passed for this field
+      example: "WAKE_UP"
+      required: true
+      selector:
+        text:
+    parameters:
+      # Description of the field
+      description: Parameters in a dictionary. `path_vars` replace variables in endpoints.json path. All others are passed directly to controller.api. For command parameters see https://tesla-api.timdorr.com/vehicle/commands.
+      # Example value that can be passed for this field
+      example: '{"path_vars": {"vehicle_id":"1"}, "wake_if_asleep":True}'
+      required: true
+      selector:
+        object: