Add notify function for BMW Connected Drive (#33484)

* Add notify function to BMW Connected Drive

* Move side effects out of init

homeassistant/components/bmw_connected_drive/__init__.py

@@ -32,7 +32,7 @@ CONFIG_SCHEMA = vol.Schema({DOMAIN: {cv.string: ACCOUNT_SCHEMA}}, extra=vol.ALLO
 SERVICE_SCHEMA = vol.Schema({vol.Required(ATTR_VIN): cv.string})
 
 
-BMW_COMPONENTS = ["binary_sensor", "device_tracker", "lock", "sensor"]
+BMW_COMPONENTS = ["binary_sensor", "device_tracker", "lock", "notify", "sensor"]
 UPDATE_INTERVAL = 5  # in minutes
 
 SERVICE_UPDATE_STATE = "update_state"

homeassistant/components/bmw_connected_drive/manifest.json

@@ -2,6 +2,7 @@
   "domain": "bmw_connected_drive",
   "name": "BMW Connected Drive",
   "documentation": "https://www.home-assistant.io/integrations/bmw_connected_drive",
-  "requirements": ["bimmer_connected==0.7.1"],
+  "requirements": ["bimmer_connected==0.7.5"],
+  "dependencies": [],
   "codeowners": ["@gerard33"]
 }

homeassistant/components/bmw_connected_drive/notify.py

@@ -0,0 +1,74 @@
+"""Support for BMW notifications."""
+import logging
+
+from homeassistant.components.notify import (
+    ATTR_DATA,
+    ATTR_TARGET,
+    ATTR_TITLE,
+    ATTR_TITLE_DEFAULT,
+    BaseNotificationService,
+)
+from homeassistant.const import ATTR_LATITUDE, ATTR_LOCATION, ATTR_LONGITUDE, ATTR_NAME
+
+from . import DOMAIN as BMW_DOMAIN
+
+ATTR_LAT = "lat"
+ATTR_LOCATION_ATTRIBUTES = ["street", "city", "postal_code", "country"]
+ATTR_LON = "lon"
+ATTR_SUBJECT = "subject"
+ATTR_TEXT = "text"
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def get_service(hass, config, discovery_info=None):
+    """Get the BMW notification service."""
+    accounts = hass.data[BMW_DOMAIN]
+    _LOGGER.debug("Found BMW accounts: %s", ", ".join([a.name for a in accounts]))
+    svc = BMWNotificationService()
+    svc.setup(accounts)
+    return svc
+
+
+class BMWNotificationService(BaseNotificationService):
+    """Send Notifications to BMW."""
+
+    def __init__(self):
+        """Set up the notification service."""
+        self.targets = {}
+
+    def setup(self, accounts):
+        """Get the BMW vehicle(s) for the account(s)."""
+        for account in accounts:
+            self.targets.update({v.name: v for v in account.account.vehicles})
+
+    def send_message(self, message="", **kwargs):
+        """Send a message or POI to the car."""
+        for _vehicle in kwargs[ATTR_TARGET]:
+            _LOGGER.debug("Sending message to %s", _vehicle.name)
+
+            # Extract params from data dict
+            title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
+            data = kwargs.get(ATTR_DATA)
+
+            # Check if message is a POI
+            if data is not None and ATTR_LOCATION in data:
+                location_dict = {
+                    ATTR_LAT: data[ATTR_LOCATION][ATTR_LATITUDE],
+                    ATTR_LON: data[ATTR_LOCATION][ATTR_LONGITUDE],
+                    ATTR_NAME: message,
+                }
+                # Update dictionary with additional attributes if available
+                location_dict.update(
+                    {
+                        k: v
+                        for k, v in data[ATTR_LOCATION].items()
+                        if k in ATTR_LOCATION_ATTRIBUTES
+                    }
+                )
+
+                _vehicle.remote_services.trigger_send_poi(location_dict)
+            else:
+                _vehicle.remote_services.trigger_send_message(
+                    {ATTR_TEXT: message, ATTR_SUBJECT: title}
+                )

requirements_all.txt

@@ -320,7 +320,7 @@ beewi_smartclim==0.0.7
 bellows-homeassistant==0.15.2
 
 # homeassistant.components.bmw_connected_drive
-bimmer_connected==0.7.1
+bimmer_connected==0.7.5
 
 # homeassistant.components.bizkaibus
 bizkaibus==0.1.1