Formatr files

Author: ArnoutD

custom_components/plugwise_usb/__init__.py

@@ -1,4 +1,5 @@
 """Support for Plugwise devices connected to a Plugwise USB-stick."""
+
 import logging
 from typing import Any, TypedDict
 
@@ -29,11 +30,15 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry):
     """Establish connection with plugwise USB-stick."""
 
     @callback
-    def _async_migrate_entity_entry(entity_entry: er.RegistryEntry) -> dict[str, Any] | None:
+    def _async_migrate_entity_entry(
+        entity_entry: er.RegistryEntry,
+    ) -> dict[str, Any] | None:
         """Migrate Plugwise entity entry."""
         return async_migrate_entity_entry(config_entry, entity_entry)
 
-    await er.async_migrate_entries(hass, config_entry.entry_id, _async_migrate_entity_entry)
+    await er.async_migrate_entries(
+        hass, config_entry.entry_id, _async_migrate_entity_entry
+    )
 
     api_stick = Stick(config_entry.data[CONF_USB_PATH])
     api_stick.cache_folder = hass.config.path(
@@ -75,19 +80,15 @@ async def async_node_discovered(node_event: NodeEvent, mac: str) -> None:
         await api_stick.discover_coordinator(load=False)
     except StickError:
         await api_stick.disconnect()
-        raise ConfigEntryNotReady(
-            "Failed to connect to Circle+"
-        ) from StickError
+        raise ConfigEntryNotReady("Failed to connect to Circle+") from StickError
 
     # Load platforms to allow them to register for node events
     await hass.config_entries.async_forward_entry_setups(
         config_entry, PLUGWISE_USB_PLATFORMS
     )
 
     # Initiate background discovery task
-    hass.async_create_task(
-        api_stick.discover_nodes(load=True)
-    )
+    hass.async_create_task(api_stick.discover_nodes(load=True))
     return True
 
 

custom_components/plugwise_usb/binary_sensor.py

@@ -1,4 +1,5 @@
 """Plugwise USB Binary Sensor component for Home Assistant."""
+
 from __future__ import annotations
 
 from dataclasses import dataclass
@@ -25,7 +26,9 @@
 
 
 @dataclass(kw_only=True)
-class PlugwiseBinarySensorEntityDescription(PlugwiseUSBEntityDescription, BinarySensorEntityDescription):
+class PlugwiseBinarySensorEntityDescription(
+    PlugwiseUSBEntityDescription, BinarySensorEntityDescription
+):
     """Describes Plugwise binary sensor entity."""
 
 
@@ -52,7 +55,9 @@ async def async_add_binary_sensor(node_event: NodeEvent, mac: str) -> None:
             return
         entities: list[PlugwiseUSBEntity] = []
         if (node_duc := config_entry.runtime_data[NODES].get(mac)) is not None:
-            _LOGGER.debug("Add binary_sensor entities for %s | duc=%s", mac, node_duc.name)
+            _LOGGER.debug(
+                "Add binary_sensor entities for %s | duc=%s", mac, node_duc.name
+            )
             entities.extend(
                 [
                     PlugwiseUSBBinarySensor(node_duc, entity_description)
@@ -71,7 +76,7 @@ async def async_add_binary_sensor(node_event: NodeEvent, mac: str) -> None:
         api_stick.subscribe_to_node_events(
             async_add_binary_sensor,
             (NodeEvent.LOADED,),
-            )
+        )
     )
 
     # load current nodes
@@ -97,11 +102,11 @@ def _handle_coordinator_update(self) -> None:
         if self.coordinator.data[self.entity_description.node_feature] is None:
             _LOGGER.info(
                 "No binary sensor data for %s",
-                str(self.entity_description.node_feature)
+                str(self.entity_description.node_feature),
             )
             return
         self._attr_is_on = getattr(
             self.coordinator.data[self.entity_description.node_feature],
-            self.entity_description.key
+            self.entity_description.key,
         )
         self.async_write_ha_state()

custom_components/plugwise_usb/config_flow.py

@@ -1,4 +1,5 @@
 """Config flow for Plugwise USB integration."""
+
 from __future__ import annotations
 
 from typing import Any
@@ -21,12 +22,13 @@
 def plugwise_stick_entries(hass):
     """Return existing connections for Plugwise USB-stick domain."""
 
-    return [entry.data.get(CONF_USB_PATH) for entry in hass.config_entries.async_entries(DOMAIN)]
+    return [
+        entry.data.get(CONF_USB_PATH)
+        for entry in hass.config_entries.async_entries(DOMAIN)
+    ]
 
 
-async def validate_usb_connection(
-    self, device_path=None
-) -> tuple[dict[str, str], str]:
+async def validate_usb_connection(self, device_path=None) -> tuple[dict[str, str], str]:
     """Test if device_path is a real Plugwise USB-Stick."""
     errors = {}
 
@@ -111,9 +113,7 @@ async def async_step_manual_path(
         return self.async_show_form(
             step_id="manual_path",
             data_schema=vol.Schema(
-                {
-                    vol.Required(CONF_USB_PATH, default="/dev/ttyUSB0"): str
-                }
+                {vol.Required(CONF_USB_PATH, default="/dev/ttyUSB0"): str}
             ),
             errors=errors,
         )

custom_components/plugwise_usb/const.py

@@ -1,4 +1,5 @@
 """Constants for Plugwise USB component."""
+
 import logging
 from typing import Final
 

custom_components/plugwise_usb/coordinator.py

@@ -1,4 +1,5 @@
 """DataUpdateCoordinator for Plugwise USB-Stick."""
+
 from collections import Counter
 from datetime import timedelta
 import logging
@@ -38,7 +39,9 @@ def __init__(
                 always_update=True,
             )
         else:
-            _LOGGER.debug("Create DUC for %s with update interval %s", node.mac, update_interval)
+            _LOGGER.debug(
+                "Create DUC for %s with update interval %s", node.mac, update_interval
+            )
             super().__init__(
                 hass,
                 _LOGGER,

custom_components/plugwise_usb/entity.py

@@ -1,4 +1,5 @@
 """Plugwise USB stick base entity."""
+
 from __future__ import annotations
 
 from collections.abc import Callable
@@ -54,7 +55,8 @@ async def async_added_to_hass(self):
         """Subscribe for push updates."""
         await super().async_added_to_hass()
         push_features = tuple(
-            push_feature for push_feature in PUSHING_FEATURES
+            push_feature
+            for push_feature in PUSHING_FEATURES
             if push_feature in self._node_info.features
         )
         # Subscribe to events
@@ -65,9 +67,14 @@ async def async_added_to_hass(self):
             )
 
     async def async_push_event(self, feature: NodeFeature, state: Any) -> None:
-        """"Update data on pushed event."""
+        """Update data on pushed event."""
         if self.node_duc is None:
-            _LOGGER.warning("Unable to push event=%s, state=%s, mac=%s", feature, state, self._node_info.mac)
+            _LOGGER.warning(
+                "Unable to push event=%s, state=%s, mac=%s",
+                feature,
+                state,
+                self._node_info.mac,
+            )
         self.node_duc.async_set_updated_data(
             {
                 feature: state,

custom_components/plugwise_usb/sensor.py

@@ -1,4 +1,5 @@
 """Plugwise USN Sensor component for Home Assistant."""
+
 from __future__ import annotations
 
 from dataclasses import dataclass
@@ -35,7 +36,9 @@
 
 
 @dataclass(kw_only=True)
-class PlugwiseSensorEntityDescription(PlugwiseUSBEntityDescription, SensorEntityDescription):
+class PlugwiseSensorEntityDescription(
+    PlugwiseUSBEntityDescription, SensorEntityDescription
+):
     """Describes Plugwise sensor entity."""
 
 
@@ -160,7 +163,7 @@ async def async_add_sensor(node_event: NodeEvent, mac: str) -> None:
         api_stick.subscribe_to_node_events(
             async_add_sensor,
             (NodeEvent.LOADED,),
-            )
+        )
     )
 
     # load current nodes
@@ -192,16 +195,12 @@ def _handle_coordinator_update(self) -> None:
             )
             return
         self._attr_native_value = getattr(
-            self.coordinator.data[
-                self.entity_description.node_feature
-            ],
-            self.entity_description.key
+            self.coordinator.data[self.entity_description.node_feature],
+            self.entity_description.key,
         )
         if self.entity_description.node_feature == NodeFeature.ENERGY:
             self._attr_last_reset = getattr(
-                self.coordinator.data[
-                    self.entity_description.node_feature
-                ],
-                f"{self.entity_description.key}_reset"
+                self.coordinator.data[self.entity_description.node_feature],
+                f"{self.entity_description.key}_reset",
             )
         self.async_write_ha_state()

custom_components/plugwise_usb/switch.py

@@ -1,4 +1,5 @@
 """Plugwise USB Switch component for HomeAssistant."""
+
 from __future__ import annotations
 
 from dataclasses import dataclass
@@ -26,7 +27,9 @@
 
 
 @dataclass(kw_only=True)
-class PlugwiseSwitchEntityDescription(PlugwiseUSBEntityDescription, SwitchEntityDescription):
+class PlugwiseSwitchEntityDescription(
+    PlugwiseUSBEntityDescription, SwitchEntityDescription
+):
     """Describes Plugwise switch entity."""
 
     async_switch_fn: str = ""
@@ -83,7 +86,7 @@ async def async_add_switch(node_event: NodeEvent, mac: str) -> None:
         api_stick.subscribe_to_node_events(
             async_add_switch,
             (NodeEvent.LOADED,),
-            )
+        )
     )
 
     # load any current nodes
@@ -119,13 +122,12 @@ def _handle_coordinator_update(self) -> None:
         """Handle updated data from the coordinator."""
         if self.coordinator.data[self.entity_description.node_feature] is None:
             _LOGGER.info(
-                "No switch data for %s",
-                str(self.entity_description.node_feature)
+                "No switch data for %s", str(self.entity_description.node_feature)
             )
             return
         self._attr_is_on = getattr(
             self.coordinator.data[self.entity_description.node_feature],
-            self.entity_description.key
+            self.entity_description.key,
         )
         self.async_write_ha_state()