diff --git a/.github/workflows/ruff.yml b/.github/workflows/ruff.yml new file mode 100644 index 0000000..b268138 --- /dev/null +++ b/.github/workflows/ruff.yml @@ -0,0 +1,8 @@ +name: Ruff +on: [push, pull_request] +jobs: + ruff: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: chartboost/ruff-action@v1 diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..5541a0e --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,15 @@ +--- +default_stages: [commit, push] +default_language_version: + python: python3 +minimum_pre_commit_version: '3.2.0' +repos: + - repo: https://github.com/astral-sh/ruff-pre-commit + # Ruff version. + rev: v0.3.5 + hooks: + # Run the linter. + - id: ruff + args: [ --fix ] + # Run the formatter. + - id: ruff-format \ No newline at end of file diff --git a/custom_components/button_plus/__init__.py b/custom_components/button_plus/__init__.py index 83dc6d5..3f32854 100644 --- a/custom_components/button_plus/__init__.py +++ b/custom_components/button_plus/__init__.py @@ -1,11 +1,11 @@ """The Button+ integration.""" + from __future__ import annotations import logging from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant -from homeassistant.helpers.device_registry import DeviceEntry from custom_components.button_plus.button_plus_api.model import DeviceConfiguration from custom_components.button_plus.buttonplushub import ButtonPlusHub @@ -23,7 +23,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up Button+ from a config entry.""" _LOGGER.debug(f"Button+ init got new device entry! {entry.entry_id.title}") - device_configuration: DeviceConfiguration = DeviceConfiguration.from_json(entry.data.get("config")) + device_configuration: DeviceConfiguration = DeviceConfiguration.from_json( + entry.data.get("config") + ) hub = ButtonPlusHub(hass, device_configuration, entry) hass.data.setdefault(DOMAIN, {})[entry.entry_id] = hub @@ -35,16 +37,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # This creates each HA object for each platform your device requires. # It's done by calling the `async_setup_entry` function in each platform module. - await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS) return True + async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Unload a config entry.""" # This is called when an entry/configured device is to be removed. The class # needs to unload itself, and remove callbacks. See the classes for further # details - _LOGGER.debug(f"Removing async_unload_entry") + _LOGGER.debug("Removing async_unload_entry") unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS) if unload_ok: hass.data[DOMAIN].pop(entry.entry_id) diff --git a/custom_components/button_plus/button.py b/custom_components/button_plus/button.py index 9dccd07..c4b1fd3 100644 --- a/custom_components/button_plus/button.py +++ b/custom_components/button_plus/button.py @@ -1,4 +1,5 @@ -""" Platform for button integration. """ +"""Platform for button integration.""" + from __future__ import annotations import logging @@ -16,15 +17,14 @@ _LOGGER = logging.getLogger(__name__) - async def async_setup_entry( - hass: HomeAssistant, - config_entry: ConfigEntry, - async_add_entities: AddEntitiesCallback, + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, ) -> None: """Add button_entities for passed config_entry in HA.""" - button_entities :list[ButtonPlusButton] = [] + button_entities: list[ButtonPlusButton] = [] hub: ButtonPlusHub = hass.data[DOMAIN][config_entry.entry_id] active_connectors = active_connectors = [ @@ -33,10 +33,14 @@ async def async_setup_entry( if connector.connector_type_enum() in [ConnectorEnum.DISPLAY, ConnectorEnum.BAR] ] - buttons = filter(lambda b: b.button_id // 2 in active_connectors, hub.config.mqtt_buttons) + buttons = filter( + lambda b: b.button_id // 2 in active_connectors, hub.config.mqtt_buttons + ) for button in buttons: - _LOGGER.debug(f"Creating button with parameters: {button.button_id} {button.label} {hub.hub_id}") + _LOGGER.debug( + f"Creating button with parameters: {button.button_id} {button.label} {hub.hub_id}" + ) entity = ButtonPlusButton(button.button_id, hub) button_entities.append(entity) hub.add_button(button.button_id, entity) @@ -51,8 +55,8 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub): self._hub = hub self._btn_id = btn_id self.entity_id = f"button.{self._hub_id}_{btn_id}" - self._attr_name = f'button-{btn_id}' - self._name = f'Button {btn_id}' + self._attr_name = f"button-{btn_id}" + self._name = f"Button {btn_id}" self._device_class = ButtonDeviceClass.IDENTIFY self._connector: Connector = hub.config.info.connectors[btn_id // 2] self.unique_id = self.unique_id_gen() @@ -65,10 +69,10 @@ def unique_id_gen(self): return self.unique_id_gen_display() def unique_id_gen_bar(self): - return f'button_{self._hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}' + return f"button_{self._hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}" def unique_id_gen_display(self): - return f'button_{self._hub_id}_{self._btn_id}_display_module' + return f"button_{self._hub_id}_{self._btn_id}_display_module" @property def name(self) -> str: @@ -87,7 +91,12 @@ def device_info(self) -> DeviceInfo: match self._connector.connector_type_enum(): case ConnectorEnum.BAR: - identifiers = {(DOMAIN, f"{self._hub.hub_id} BAR Module {self._connector.connector_id}")} + identifiers = { + ( + DOMAIN, + f"{self._hub.hub_id} BAR Module {self._connector.connector_id}", + ) + } case ConnectorEnum.DISPLAY: identifiers = {(DOMAIN, f"{self._hub.hub_id} Display Module")} diff --git a/custom_components/button_plus/button_plus_api/__init__.py b/custom_components/button_plus/button_plus_api/__init__.py index d4523cc..facf099 100644 --- a/custom_components/button_plus/button_plus_api/__init__.py +++ b/custom_components/button_plus/button_plus_api/__init__.py @@ -1 +1 @@ -""" Initialize Button+ Api module """ +"""Initialize Button+ Api module""" diff --git a/custom_components/button_plus/button_plus_api/api_client.py b/custom_components/button_plus/button_plus_api/api_client.py index fc1bb70..a3116c1 100644 --- a/custom_components/button_plus/button_plus_api/api_client.py +++ b/custom_components/button_plus/button_plus_api/api_client.py @@ -8,7 +8,7 @@ class ApiClient: - """ Client to talk to Button+ website """ + """Client to talk to Button+ website""" def __init__(self, session, cookie=None) -> None: _LOGGER.debug(f"DEBUG CONFIG {cookie}") @@ -20,14 +20,14 @@ def __init__(self, session, cookie=None) -> None: self._cookie = cookie self._headers = { - 'authority': 'api.button.plus', - 'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', - 'accept-language': 'en-NL,en-US;q=0.9,en;q=0.8,nl-NL;q=0.7,nl;q=0.6,en-GB;q=0.5', - 'cache-control': 'no-cache', - 'cookie': self._cookie, + "authority": "api.button.plus", + "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7", + "accept-language": "en-NL,en-US;q=0.9,en;q=0.8,nl-NL;q=0.7,nl;q=0.6,en-GB;q=0.5", + "cache-control": "no-cache", + "cookie": self._cookie, } - _LOGGER.debug(f"Initialize Button+ API client") + _LOGGER.debug("Initialize Button+ API client") async def test_connection(self): url = f"{self._base}/button/buttons" @@ -54,22 +54,24 @@ async def get_cookie_from_login(self, email=str, password=str): json_data = json.dumps(data) _LOGGER.debug(f"json dump: {json_data}") headers = { - 'accept': '*/*', - 'accept-language': 'en-NL,en;q=0.9', - 'content-type': 'application/json', - 'origin': 'https://button.plus', - 'referer': 'https://button.plus/', - 'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' + "accept": "*/*", + "accept-language": "en-NL,en;q=0.9", + "content-type": "application/json", + "origin": "https://button.plus", + "referer": "https://button.plus/", + "user-agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", } async with self._session.post(url, data=json_data, headers=headers) as response: response_body = await response.text() if not response.cookies: - raise Exception(f"Login error with username and password, response: {response_body}") + raise Exception( + f"Login error with username and password, response: {response_body}" + ) cookie_string = str(response.cookies) - match = re.search(r'auth_cookie=[^;]+', cookie_string) + match = re.search(r"auth_cookie=[^;]+", cookie_string) auth_cookie = match.group() diff --git a/custom_components/button_plus/button_plus_api/connector_type.py b/custom_components/button_plus/button_plus_api/connector_type.py index 54a61a5..a6a26f7 100644 --- a/custom_components/button_plus/button_plus_api/connector_type.py +++ b/custom_components/button_plus/button_plus_api/connector_type.py @@ -1,6 +1,7 @@ from enum import Enum + class ConnectorEnum(Enum): NOT_CONNECTED = 0 BAR = 1 - DISPLAY = 2 \ No newline at end of file + DISPLAY = 2 diff --git a/custom_components/button_plus/button_plus_api/local_api_client.py b/custom_components/button_plus/button_plus_api/local_api_client.py index b8f2501..dcd6adf 100644 --- a/custom_components/button_plus/button_plus_api/local_api_client.py +++ b/custom_components/button_plus/button_plus_api/local_api_client.py @@ -6,7 +6,7 @@ class LocalApiClient: - """ Client to talk to Button+ local devices """ + """Client to talk to Button+ local devices""" def __init__(self, ip_address, session) -> None: self._base = f"http://{ip_address}" @@ -25,4 +25,3 @@ async def push_config(self, config): _LOGGER.debug(f"push_config {url}") async with self._session.post(url, data=config.to_json()) as response: return await response.text() - diff --git a/custom_components/button_plus/button_plus_api/model.py b/custom_components/button_plus/button_plus_api/model.py index a99116e..ebaa496 100644 --- a/custom_components/button_plus/button_plus_api/model.py +++ b/custom_components/button_plus/button_plus_api/model.py @@ -4,6 +4,7 @@ from .connector_type import ConnectorEnum from .event_type import EventType + class Connector: def __init__(self, connector_id: int, connector_type: int): self.connector_id = connector_id @@ -13,11 +14,8 @@ def connector_type_enum(self) -> ConnectorEnum: return ConnectorEnum(self.connector_type) @staticmethod - def from_dict(data: Dict[str, Any]) -> 'Connector': - return Connector( - connector_id=data['id'], - connector_type=data['type'] - ) + def from_dict(data: Dict[str, Any]) -> "Connector": + return Connector(connector_id=data["id"], connector_type=data["type"]) class Sensor: @@ -26,36 +24,44 @@ def __init__(self, sensor_id: int, description: str): self.description = description @staticmethod - def from_dict(data: Dict[str, Any]) -> 'Sensor': - return Sensor( - sensor_id=data['sensorid'], - description=data['description'] - ) + def from_dict(data: Dict[str, Any]) -> "Sensor": + return Sensor(sensor_id=data["sensorid"], description=data["description"]) class Info: - def __init__(self, device_id: str, mac: str, ip_address: str, firmware: str, large_display: int, - connectors: List[Connector], sensors: List[Sensor]): + def __init__( + self, + device_id: str, + mac: str, + ip_address: str, + firmware: str, + large_display: int, + connectors: List[Connector], + sensors: List[Sensor], + ): self.device_id = device_id self.mac = mac self.ip_address = ip_address self.firmware = firmware self.large_display = large_display - self.connectors : List[Connector] = connectors + self.connectors: List[Connector] = connectors self.sensors = sensors @staticmethod - def from_dict(data: Dict[str, Any]) -> 'Info': + def from_dict(data: Dict[str, Any]) -> "Info": return Info( - device_id=data['id'], - mac=data['mac'], - ip_address=data['ipaddress'], - firmware=data['firmware'], - large_display=data['largedisplay'], - connectors=[Connector.from_dict(connector) for connector in data['connectors']], - sensors=[Sensor.from_dict(sensor) for sensor in data['sensors']] + device_id=data["id"], + mac=data["mac"], + ip_address=data["ipaddress"], + firmware=data["firmware"], + large_display=data["largedisplay"], + connectors=[ + Connector.from_dict(connector) for connector in data["connectors"] + ], + sensors=[Sensor.from_dict(sensor) for sensor in data["sensors"]], ) + class Topic: def __init__(self, broker_id: str, topic: str, payload: str, event_type: int): self.broker_id = broker_id @@ -67,18 +73,28 @@ def connector_type_enum(self) -> EventType: return EventType(self.event_type) @staticmethod - def from_dict(data: Dict[str, Any]) -> 'Topic': + def from_dict(data: Dict[str, Any]) -> "Topic": return Topic( - broker_id=data['brokerid'], - topic=data['topic'], - payload=data['payload'], - event_type=data['eventtype'] + broker_id=data["brokerid"], + topic=data["topic"], + payload=data["payload"], + event_type=data["eventtype"], ) + class Core: - def __init__(self, name: str, location: str, auto_backup: bool, brightness_large_display: int, - brightness_mini_display: int, led_color_front: int, led_color_wall: int, color: int, - topics: List[Topic]): + def __init__( + self, + name: str, + location: str, + auto_backup: bool, + brightness_large_display: int, + brightness_mini_display: int, + led_color_front: int, + led_color_wall: int, + color: int, + topics: List[Topic], + ): self.name = name self.location = location self.auto_backup = auto_backup @@ -90,23 +106,32 @@ def __init__(self, name: str, location: str, auto_backup: bool, brightness_large self.topics = topics @staticmethod - def from_dict(data: Dict[str, Any]) -> 'Core': + def from_dict(data: Dict[str, Any]) -> "Core": return Core( - name=data['name'], - location=data['location'], - auto_backup=data['autobackup'], - brightness_large_display=data['brightnesslargedisplay'], - brightness_mini_display=data['brightnessminidisplay'], - led_color_front=data['ledcolorfront'], - led_color_wall=data['ledcolorwall'], - color=data['color'], - topics=[Topic.from_dict(topic) for topic in data.get('topics', [])] + name=data["name"], + location=data["location"], + auto_backup=data["autobackup"], + brightness_large_display=data["brightnesslargedisplay"], + brightness_mini_display=data["brightnessminidisplay"], + led_color_front=data["ledcolorfront"], + led_color_wall=data["ledcolorwall"], + color=data["color"], + topics=[Topic.from_dict(topic) for topic in data.get("topics", [])], ) class MqttButton: - def __init__(self, button_id: int, label: str, top_label: str, led_color_front: int, led_color_wall: int, - long_delay: int, long_repeat: int, topics: List[Topic]): + def __init__( + self, + button_id: int, + label: str, + top_label: str, + led_color_front: int, + led_color_wall: int, + long_delay: int, + long_repeat: int, + topics: List[Topic], + ): self.button_id = button_id self.label = label self.top_label = top_label @@ -117,22 +142,32 @@ def __init__(self, button_id: int, label: str, top_label: str, led_color_front: self.topics = topics @staticmethod - def from_dict(data: Dict[str, Any]) -> 'MqttButton': + def from_dict(data: Dict[str, Any]) -> "MqttButton": return MqttButton( - button_id=data['id'], - label=data['label'], - top_label=data['toplabel'], - led_color_front=data['ledcolorfront'], - led_color_wall=data['ledcolorwall'], - long_delay=data['longdelay'], - long_repeat=data['longrepeat'], - topics=[Topic.from_dict(topic) for topic in data.get('topics', [])] + button_id=data["id"], + label=data["label"], + top_label=data["toplabel"], + led_color_front=data["ledcolorfront"], + led_color_wall=data["ledcolorwall"], + long_delay=data["longdelay"], + long_repeat=data["longrepeat"], + topics=[Topic.from_dict(topic) for topic in data.get("topics", [])], ) class MqttDisplay: - def __init__(self, x: int, y: int, font_size: int, align: int, width: int, label: str, unit: str, round: int, - topics: List[Topic]): + def __init__( + self, + x: int, + y: int, + font_size: int, + align: int, + width: int, + label: str, + unit: str, + round: int, + topics: List[Topic], + ): self.x = x self.y = y self.font_size = font_size @@ -144,22 +179,30 @@ def __init__(self, x: int, y: int, font_size: int, align: int, width: int, label self.topics = topics @staticmethod - def from_dict(data: Dict[str, Any]) -> 'MqttDisplay': + def from_dict(data: Dict[str, Any]) -> "MqttDisplay": return MqttDisplay( - x=data['x'], - y=data['y'], - font_size=data['fontsize'], - align=data['align'], - width=data['width'], - label=data['label'], - unit=data['unit'], - round=data['round'], - topics=[Topic.from_dict(topic) for topic in data.get('topics', [])] + x=data["x"], + y=data["y"], + font_size=data["fontsize"], + align=data["align"], + width=data["width"], + label=data["label"], + unit=data["unit"], + round=data["round"], + topics=[Topic.from_dict(topic) for topic in data.get("topics", [])], ) class MqttBroker: - def __init__(self, broker_id: str, url: str, port: int, ws_port: int, username: str, password: str): + def __init__( + self, + broker_id: str, + url: str, + port: int, + ws_port: int, + username: str, + password: str, + ): self.broker_id = broker_id self.url = url self.port = port @@ -168,14 +211,14 @@ def __init__(self, broker_id: str, url: str, port: int, ws_port: int, username: self.password = password @staticmethod - def from_dict(data: Dict[str, Any]) -> 'MqttBroker': + def from_dict(data: Dict[str, Any]) -> "MqttBroker": return MqttBroker( - broker_id=data['brokerid'], - url=data['url'], - port=data['port'], - ws_port=data['wsport'], - username=data['username'], - password=data['password'] + broker_id=data["brokerid"], + url=data["url"], + port=data["port"], + ws_port=data["wsport"], + username=data["username"], + password=data["password"], ) @@ -186,17 +229,24 @@ def __init__(self, sensor_id: int, topic: Topic, interval: int): self.interval = interval @staticmethod - def from_dict(data: Dict[str, Any]) -> 'MqttSensor': + def from_dict(data: Dict[str, Any]) -> "MqttSensor": return MqttSensor( - sensor_id=data['sensorid'], - topic=Topic.from_dict(data['topic']), - interval=data['interval'] + sensor_id=data["sensorid"], + topic=Topic.from_dict(data["topic"]), + interval=data["interval"], ) class DeviceConfiguration: - def __init__(self, info: Info, core: Core, mqtt_buttons: List[MqttButton], mqtt_displays: List[MqttDisplay], - mqtt_brokers: List[MqttBroker], mqtt_sensors: List[MqttSensor]): + def __init__( + self, + info: Info, + core: Core, + mqtt_buttons: List[MqttButton], + mqtt_displays: List[MqttDisplay], + mqtt_brokers: List[MqttBroker], + mqtt_sensors: List[MqttSensor], + ): self.info = info self.core = core self.mqtt_buttons = mqtt_buttons @@ -205,72 +255,88 @@ def __init__(self, info: Info, core: Core, mqtt_buttons: List[MqttButton], mqtt_ self.mqtt_sensors = mqtt_sensors @staticmethod - def from_json(json_data: str) -> 'DeviceConfiguration': + def from_json(json_data: str) -> "DeviceConfiguration": data = json.loads(json_data) return DeviceConfiguration( - info=Info.from_dict(data['info']), - core=Core.from_dict(data['core']), - mqtt_buttons=[MqttButton.from_dict(button) for button in data['mqttbuttons']], - mqtt_displays=[MqttDisplay.from_dict(display) for display in data['mqttdisplays']], - mqtt_brokers=[MqttBroker.from_dict(broker) for broker in data['mqttbrokers']], - mqtt_sensors=[MqttSensor.from_dict(sensor) for sensor in data['mqttsensors']], + info=Info.from_dict(data["info"]), + core=Core.from_dict(data["core"]), + mqtt_buttons=[ + MqttButton.from_dict(button) for button in data["mqttbuttons"] + ], + mqtt_displays=[ + MqttDisplay.from_dict(display) for display in data["mqttdisplays"] + ], + mqtt_brokers=[ + MqttBroker.from_dict(broker) for broker in data["mqttbrokers"] + ], + mqtt_sensors=[ + MqttSensor.from_dict(sensor) for sensor in data["mqttsensors"] + ], ) def to_json(self) -> str: def serialize(obj): - if hasattr(obj, '__dict__'): + if hasattr(obj, "__dict__"): d = obj.__dict__.copy() # Convert the root keys if isinstance(obj, DeviceConfiguration): - d['mqttbuttons'] = [serialize(button) for button in d.pop('mqtt_buttons')] - d['mqttdisplays'] = [serialize(display) for display in d.pop('mqtt_displays')] - d['mqttbrokers'] = [serialize(broker) for broker in d.pop('mqtt_brokers')] - d['mqttsensors'] = [serialize(sensor) for sensor in d.pop('mqtt_sensors')] + d["mqttbuttons"] = [ + serialize(button) for button in d.pop("mqtt_buttons") + ] + d["mqttdisplays"] = [ + serialize(display) for display in d.pop("mqtt_displays") + ] + d["mqttbrokers"] = [ + serialize(broker) for broker in d.pop("mqtt_brokers") + ] + d["mqttsensors"] = [ + serialize(sensor) for sensor in d.pop("mqtt_sensors") + ] if isinstance(obj, Info): - d['id'] = d.pop('device_id') - d['ipaddress'] = d.pop('ip_address') - d['largedisplay'] = d.pop('large_display') + d["id"] = d.pop("device_id") + d["ipaddress"] = d.pop("ip_address") + d["largedisplay"] = d.pop("large_display") elif isinstance(obj, Connector): - d['id'] = d.pop('connector_id') - d['type'] = d.pop('connector_type') + d["id"] = d.pop("connector_id") + d["type"] = d.pop("connector_type") elif isinstance(obj, Sensor): - d['sensorid'] = d.pop('sensor_id') + d["sensorid"] = d.pop("sensor_id") elif isinstance(obj, Core): - d['autobackup'] = d.pop('auto_backup') - d['brightnesslargedisplay'] = d.pop('brightness_large_display') - d['brightnessminidisplay'] = d.pop('brightness_mini_display') - d['ledcolorfront'] = d.pop('led_color_front') - d['ledcolorwall'] = d.pop('led_color_wall') + d["autobackup"] = d.pop("auto_backup") + d["brightnesslargedisplay"] = d.pop("brightness_large_display") + d["brightnessminidisplay"] = d.pop("brightness_mini_display") + d["ledcolorfront"] = d.pop("led_color_front") + d["ledcolorwall"] = d.pop("led_color_wall") # Custom mappings for MqttButton class elif isinstance(obj, MqttButton): - d['id'] = d.pop('button_id') - d['toplabel'] = d.pop('top_label') - d['ledcolorfront'] = d.pop('led_color_front') - d['ledcolorwall'] = d.pop('led_color_wall') - d['longdelay'] = d.pop('long_delay') - d['longrepeat'] = d.pop('long_repeat') + d["id"] = d.pop("button_id") + d["toplabel"] = d.pop("top_label") + d["ledcolorfront"] = d.pop("led_color_front") + d["ledcolorwall"] = d.pop("led_color_wall") + d["longdelay"] = d.pop("long_delay") + d["longrepeat"] = d.pop("long_repeat") elif isinstance(obj, Topic): - d['brokerid'] = d.pop('broker_id') - d['eventtype'] = d.pop('event_type') + d["brokerid"] = d.pop("broker_id") + d["eventtype"] = d.pop("event_type") elif isinstance(obj, MqttDisplay): - d['fontsize'] = d.pop('font_size') - d['topics'] = [serialize(topic) for topic in d['topics']] + d["fontsize"] = d.pop("font_size") + d["topics"] = [serialize(topic) for topic in d["topics"]] elif isinstance(obj, MqttBroker): - d['brokerid'] = d.pop('broker_id') - d['wsport'] = d.pop('ws_port') + d["brokerid"] = d.pop("broker_id") + d["wsport"] = d.pop("ws_port") elif isinstance(obj, MqttSensor): - d['sensorid'] = d.pop('sensor_id') - d['topic'] = serialize(d['topic']) + d["sensorid"] = d.pop("sensor_id") + d["topic"] = serialize(d["topic"]) # Filter out None values return {k: v for k, v in d.items() if v is not None} diff --git a/custom_components/button_plus/buttonplushub.py b/custom_components/button_plus/buttonplushub.py index d835f6a..4d7c9b5 100644 --- a/custom_components/button_plus/buttonplushub.py +++ b/custom_components/button_plus/buttonplushub.py @@ -1,4 +1,5 @@ """Button+ connects several devices.""" + from __future__ import annotations import logging @@ -19,13 +20,17 @@ class ButtonPlusHub: """hub for Button+.""" - def __init__(self, hass: HomeAssistant, config: DeviceConfiguration, entry: ConfigEntry) -> None: + def __init__( + self, hass: HomeAssistant, config: DeviceConfiguration, entry: ConfigEntry + ) -> None: _LOGGER.debug(f"New hub with config {config.core}") self._hass = hass self.config = config self._name = config.core.name or config.info.device_id self._id = config.info.device_id - self._client = LocalApiClient(config.info.ip_address, aiohttp_client.async_get_clientsession(hass)) + self._client = LocalApiClient( + config.info.ip_address, aiohttp_client.async_get_clientsession(hass) + ) self.online = True self.button_entities = {} self.label_entities = {} @@ -43,14 +48,25 @@ def __init__(self, hass: HomeAssistant, config: DeviceConfiguration, entry: Conf suggested_area=self.config.core.location, name=self._name, model="Base Module", - sw_version=config.info.firmware + sw_version=config.info.firmware, ) # 1 or none display module - self.display_module = next((self.create_display_module(hass, entry, self) for _ in self.connector(ConnectorEnum.DISPLAY)), None) - self.display_bar = [(connector_id, self.create_bar_module(hass, entry, self, connector_id)) for connector_id in self.connector(ConnectorEnum.BAR)] + self.display_module = next( + ( + self.create_display_module(hass, entry, self) + for _ in self.connector(ConnectorEnum.DISPLAY) + ), + None, + ) + self.display_bar = [ + (connector_id, self.create_bar_module(hass, entry, self, connector_id)) + for connector_id in self.connector(ConnectorEnum.BAR) + ] - def create_display_module(self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub) -> None: + def create_display_module( + self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub + ) -> None: _LOGGER.debug(f"Add display module from '{hub.hub_id}'") device_registry = dr.async_get(hass) @@ -61,17 +77,22 @@ def create_display_module(self, hass: HomeAssistant, entry: ConfigEntry, hub: Bu model="Display Module", manufacturer=MANUFACTURER, suggested_area=hub.config.core.location, - identifiers={ - (DOMAIN, f"{hub.hub_id} Display Module") - }, - via_device=(DOMAIN, hub.hub_id) + identifiers={(DOMAIN, f"{hub.hub_id} Display Module")}, + via_device=(DOMAIN, hub.hub_id), ) return device - - def create_bar_module(self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub, connector_id: int) -> None: - _LOGGER.debug(f"Add bar module from '{hub.hub_id}' with connector '{connector_id}'") + def create_bar_module( + self, + hass: HomeAssistant, + entry: ConfigEntry, + hub: ButtonPlusHub, + connector_id: int, + ) -> None: + _LOGGER.debug( + f"Add bar module from '{hub.hub_id}' with connector '{connector_id}'" + ) device_registry = dr.async_get(hass) device = device_registry.async_get_or_create( @@ -81,10 +102,8 @@ def create_bar_module(self, hass: HomeAssistant, entry: ConfigEntry, hub: Button model="Bar module", manufacturer=MANUFACTURER, suggested_area=hub.config.core.location, - identifiers={ - (DOMAIN, f"{hub.hub_id} BAR Module {connector_id}") - }, - via_device=(DOMAIN, hub.hub_id) + identifiers={(DOMAIN, f"{hub.hub_id} BAR Module {connector_id}")}, + via_device=(DOMAIN, hub.hub_id), ) return device @@ -96,7 +115,6 @@ def connector(self, connector_type: ConnectorEnum): if connector.connector_type_enum() in [connector_type] ] - @property def client(self) -> LocalApiClient: """Return Button+ API client""" @@ -121,4 +139,3 @@ def add_top_label(self, button_id, entity): def add_brightness(self, identifier, entity): self.brightness_entities[identifier] = entity - diff --git a/custom_components/button_plus/config_flow.py b/custom_components/button_plus/config_flow.py index 524301c..c94983e 100644 --- a/custom_components/button_plus/config_flow.py +++ b/custom_components/button_plus/config_flow.py @@ -1,4 +1,5 @@ """Config flow for Hello World integration.""" + from __future__ import annotations import ipaddress @@ -26,11 +27,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): """Handle a config flow for Button+.""" - local_brokers = [ - "core-mosquitto", - "127.0.0.1", - "localhost" - ] + local_brokers = ["core-mosquitto", "127.0.0.1", "localhost"] def __init__(self): self.mqtt_entry = None @@ -45,13 +42,12 @@ async def async_step_user(self, user_input=None): mqtt_entries = self.hass.config_entries.async_entries(domain="mqtt") if len(mqtt_entries) < 1: - mqtt_url = f'{get_url(self.hass)}/config/integrations/integration/mqtt' + mqtt_url = f"{get_url(self.hass)}/config/integrations/integration/mqtt" return self.async_abort( reason="mqtt_not_enabled", - description_placeholders={ - "mqtt_integration_link": mqtt_url - }) + description_placeholders={"mqtt_integration_link": mqtt_url}, + ) mqtt_entry = mqtt_entries[0] broker = self.get_mqtt_endpoint(mqtt_entry.data.get("broker")) @@ -65,28 +61,25 @@ async def async_step_user(self, user_input=None): return self.async_show_form( step_id="user", - data_schema=vol.Schema({ - vol.Required("broker", default=broker): str - }), + data_schema=vol.Schema({vol.Required("broker", default=broker): str}), errors=errors, description_placeholders={ "mqtt_broker": broker, "mqtt_broker_port": broker_port, - "mqtt_user": broker_username - } + "mqtt_user": broker_username, + }, ) async def async_step_choose_entry(self, user_input=None): - errors = {} # if user_input is not None: return self.async_show_menu( step_id="choose_entry", menu_options=["fetch_website", "manual"], - description_placeholders={} + description_placeholders={}, ) async def async_step_manual(self, user_input=None): - """ Handle setting up button plus from manual IP.""" + """Handle setting up button plus from manual IP.""" errors = {} ip = None if user_input is not None: @@ -95,9 +88,13 @@ async def async_step_manual(self, user_input=None): if valid: try: _LOGGER.debug(f"Fetching button+ device at {ip}") - api_client = LocalApiClient(ip, aiohttp_client.async_get_clientsession(self.hass)) + api_client = LocalApiClient( + ip, aiohttp_client.async_get_clientsession(self.hass) + ) json_config = await api_client.fetch_config() - device_config: DeviceConfiguration = DeviceConfiguration.from_json(json_config) + device_config: DeviceConfiguration = DeviceConfiguration.from_json( + json_config + ) self.add_broker_to_config(device_config) self.add_topics_to_core(device_config) @@ -108,14 +105,14 @@ async def async_step_manual(self, user_input=None): return self.async_create_entry( title=f"{device_config.core.name}", description=f"Base module on {ip} with id {device_config.info.device_id}", - data={"config": json_config} + data={"config": json_config}, ) except JSONDecodeError as ex: # pylint: disable=broad-except _LOGGER.error( f"{DOMAIN} Could not parse json from IP {ip} : %s - traceback: %s", ex, - traceback.format_exc() + traceback.format_exc(), ) errors["base"] = f"Error connecting or reading from {ip}" @@ -123,21 +120,19 @@ async def async_step_manual(self, user_input=None): _LOGGER.error( f"{DOMAIN} Exception in login : %s - traceback: %s", ex, - traceback.format_exc() + traceback.format_exc(), ) errors["base"] = "cannot_connect" else: - errors["base"] = 'invalid_ip' + errors["base"] = "invalid_ip" return self.async_show_form( step_id="manual", data_schema=vol.Schema({CONF_IP_ADDRESS: str}), errors=errors, - description_placeholders={ - "ip": ip - } + description_placeholders={"ip": ip}, ) async def async_step_fetch_website(self, user_input=None): @@ -162,20 +157,24 @@ async def async_step_fetch_website(self, user_input=None): for device in devices: device_website_id = device.get("Id") - device_ip = device.get('IpAddress') + device_ip = device.get("IpAddress") if not device_ip: - _LOGGER.warning(f"Skipping device {device_website_id}, it has no IP so must be virtual") + _LOGGER.warning( + f"Skipping device {device_website_id}, it has no IP so must be virtual" + ) continue - _LOGGER.debug(f"loaded device from website with id: {device_website_id} and ip {device_ip}") + _LOGGER.debug( + f"loaded device from website with id: {device_website_id} and ip {device_ip}" + ) device_config = json.loads(device.get("Json")) - device_name = device_config.get('core').get('name') - device_id = device_config.get('info').get('id') + device_name = device_config.get("core").get("name") + device_id = device_config.get("info").get("id") last_entry = self.async_create_entry( title=f"{device_name}", description=f"Base module on {device_ip} with local id {device_id} and website id {device_website_id}", - data=device_config + data=device_config, ) return last_entry @@ -184,15 +183,17 @@ async def async_step_fetch_website(self, user_input=None): _LOGGER.error( f"{DOMAIN} Could not parse json from Button+ website : %s - traceback: %s", ex, - traceback.format_exc() + traceback.format_exc(), ) - errors["base"] = "Error connecting or reading from https://api.button.plus/" + errors["base"] = ( + "Error connecting or reading from https://api.button.plus/" + ) except Exception as ex: # pylint: disable=broad-except _LOGGER.error( f"{DOMAIN} Exception in login : %s - traceback: %s", ex, - traceback.format_exc() + traceback.format_exc(), ) errors["base"] = "cannot_connect" @@ -202,8 +203,10 @@ async def async_step_fetch_website(self, user_input=None): return self.async_show_form( step_id="fetch_website", - data_schema=vol.Schema({CONF_EMAIL: str, CONF_PASSWORD: str, "cookie": str}), - errors=errors + data_schema=vol.Schema( + {CONF_EMAIL: str, CONF_PASSWORD: str, "cookie": str} + ), + errors=errors, ) async def setup_api_client(self, user_input): @@ -211,7 +214,9 @@ async def setup_api_client(self, user_input): if "cookie" not in user_input: client = ApiClient(aiohttp_client.async_get_clientsession(self.hass)) - cookie = await client.get_cookie_from_login(user_input.get('email'), user_input.get('password')) + cookie = await client.get_cookie_from_login( + user_input.get("email"), user_input.get("password") + ) else: cookie = user_input.get("cookie") @@ -224,100 +229,126 @@ def validate_ip(self, ip) -> bool: except ValueError: return False - def add_broker_to_config(self, device_config: DeviceConfiguration) -> DeviceConfiguration: + def add_broker_to_config( + self, device_config: DeviceConfiguration + ) -> DeviceConfiguration: mqtt_entry = self.mqtt_entry broker_port = mqtt_entry.data.get("port") broker_username = mqtt_entry.data.get("username", "") broker_password = mqtt_entry.data.get("password", "") broker = MqttBroker( - broker_id=f"ha-button-plus", + broker_id="ha-button-plus", url=f"mqtt://{self.broker_endpoint}/", port=broker_port, ws_port=9001, username=broker_username, - password=broker_password + password=broker_password, ) device_config.mqtt_brokers.append(broker) return device_config - def add_topics_to_core(self, device_config : DeviceConfiguration) -> DeviceConfiguration: + def add_topics_to_core( + self, device_config: DeviceConfiguration + ) -> DeviceConfiguration: device_id = device_config.info.device_id min_version = "1.11" if version.parse(device_config.info.firmware) < version.parse(min_version): - _LOGGER.debug(f"Current version {device_config.info.firmware} doesn't support the brightness, it must be at least firmware version {min_version}") + _LOGGER.debug( + f"Current version {device_config.info.firmware} doesn't support the brightness, it must be at least firmware version {min_version}" + ) return - device_config.core.topics.append({ - "brokerid": "ha-button-plus", - "topic": f"buttonplus/{device_id}/brightness/large", - "payload": "", - "eventtype": EventType.BRIGHTNESS_LARGE_DISPLAY - }) - - device_config.core.topics.append({ - "brokerid": "ha-button-plus", - "topic": f"buttonplus/{device_id}/brightness/mini", - "payload": "", - "eventtype": EventType.BRIGHTNESS_MINI_DISPLAY - }) - - device_config.core.topics.append({ - "brokerid": "ha-button-plus", - "topic": f"buttonplus/{device_id}/page/status", - "payload": "", - "eventtype": EventType.PAGE_STATUS - }) - - device_config.core.topics.append({ - "brokerid": "ha-button-plus", - "topic": f"buttonplus/{device_id}/page/set", - "payload": "", - "eventtype": EventType.SET_PAGE - }) - - def add_topics_to_buttons(self, device_config : DeviceConfiguration) -> DeviceConfiguration: + device_config.core.topics.append( + { + "brokerid": "ha-button-plus", + "topic": f"buttonplus/{device_id}/brightness/large", + "payload": "", + "eventtype": EventType.BRIGHTNESS_LARGE_DISPLAY, + } + ) + + device_config.core.topics.append( + { + "brokerid": "ha-button-plus", + "topic": f"buttonplus/{device_id}/brightness/mini", + "payload": "", + "eventtype": EventType.BRIGHTNESS_MINI_DISPLAY, + } + ) + + device_config.core.topics.append( + { + "brokerid": "ha-button-plus", + "topic": f"buttonplus/{device_id}/page/status", + "payload": "", + "eventtype": EventType.PAGE_STATUS, + } + ) + + device_config.core.topics.append( + { + "brokerid": "ha-button-plus", + "topic": f"buttonplus/{device_id}/page/set", + "payload": "", + "eventtype": EventType.SET_PAGE, + } + ) + + def add_topics_to_buttons( + self, device_config: DeviceConfiguration + ) -> DeviceConfiguration: device_id = device_config.info.device_id active_connectors = [ connector.connector_id for connector in device_config.info.connectors - if connector.connector_type_enum() in [ConnectorEnum.DISPLAY, ConnectorEnum.BAR] + if connector.connector_type_enum() + in [ConnectorEnum.DISPLAY, ConnectorEnum.BAR] ] - for button in filter(lambda b: b.button_id // 2 in active_connectors, device_config.mqtt_buttons): - + for button in filter( + lambda b: b.button_id // 2 in active_connectors, device_config.mqtt_buttons + ): # Create topics for button main label - button.topics.append({ - "brokerid": "ha-button-plus", - "topic": f"buttonplus/{device_id}/button/{button.button_id}/label", - "payload": "", - "eventtype": EventType.LABEL - }) + button.topics.append( + { + "brokerid": "ha-button-plus", + "topic": f"buttonplus/{device_id}/button/{button.button_id}/label", + "payload": "", + "eventtype": EventType.LABEL, + } + ) # Create topics for button top label - button.topics.append({ - "brokerid": "ha-button-plus", - "topic": f"buttonplus/{device_id}/button/{button.button_id}/top_label", - "payload": "", - "eventtype": EventType.TOPLABEL - }) + button.topics.append( + { + "brokerid": "ha-button-plus", + "topic": f"buttonplus/{device_id}/button/{button.button_id}/top_label", + "payload": "", + "eventtype": EventType.TOPLABEL, + } + ) # Create topics for button click - button.topics.append({ - "brokerid": "ha-button-plus", - "topic": f"buttonplus/{device_id}/button/{button.button_id}/click", - "payload": "press", - "eventtype": EventType.CLICK - }) + button.topics.append( + { + "brokerid": "ha-button-plus", + "topic": f"buttonplus/{device_id}/button/{button.button_id}/click", + "payload": "press", + "eventtype": EventType.CLICK, + } + ) return device_config def get_mqtt_endpoint(self, endpoint: str) -> str: # Internal add-on is not reachable from the Button+ device so we use the hass ip if endpoint in self.local_brokers: - _LOGGER.debug(f'mqtt host is internal so use {self.hass.config.api.host} instead of {endpoint}') + _LOGGER.debug( + f"mqtt host is internal so use {self.hass.config.api.host} instead of {endpoint}" + ) return self.hass.config.api.host return endpoint diff --git a/custom_components/button_plus/coordinator.py b/custom_components/button_plus/coordinator.py index 6bb4bf6..e8bf58e 100644 --- a/custom_components/button_plus/coordinator.py +++ b/custom_components/button_plus/coordinator.py @@ -34,34 +34,26 @@ def __init__(self, hass: HomeAssistant, hub: ButtonPlusHub): self._mqtt_topic_page = f"buttonplus/{hub.hub_id}/page/+" async def _async_update_data(self): - """Create MQTT subscriptions for buttonplus """ - _LOGGER.debug(f"Initial data fetch from coordinator") + """Create MQTT subscriptions for buttonplus""" + _LOGGER.debug("Initial data fetch from coordinator") if not self._mqtt_subscribed_buttons: self.unsubscribe_mqtt = await mqtt.async_subscribe( - self._hass, - self._mqtt_topic_buttons, - self.mqtt_button_callback, - 0 + self._hass, self._mqtt_topic_buttons, self.mqtt_button_callback, 0 ) _LOGGER.debug(f"MQTT subscribed to {self._mqtt_topic_buttons}") - if not self._mqtt_subscribed_buttons: self.unsubscribe_mqtt = await mqtt.async_subscribe( self._hass, self._mqtt_topic_brightness, self.mqtt_brightness_callback, - 0 + 0, ) _LOGGER.debug(f"MQTT subscribed to {self._mqtt_topic_brightness}") - if not self._mqtt_subscribed_buttons: self.unsubscribe_mqtt = await mqtt.async_subscribe( - self._hass, - self._mqtt_topic_page, - self.mqtt_page_callback, - 0 + self._hass, self._mqtt_topic_page, self.mqtt_page_callback, 0 ) _LOGGER.debug(f"MQTT subscribed to {self._mqtt_topic_page}") @@ -69,18 +61,17 @@ async def _async_update_data(self): async def mqtt_page_callback(self, message: ReceiveMessage): # Handle the message here _LOGGER.debug(f"Received message on topic {message.topic}: {message.payload}") - match = re.search(r'/page/(\w+)', message.topic) + # match = re.search(r"/page/(\w+)", message.topic) # is 'status' or 'set' - page_type = match.group(1) + # page_type = match.group(1) # TODO: implement page control - @callback async def mqtt_brightness_callback(self, message: ReceiveMessage): # Handle the message here _LOGGER.debug(f"Received message on topic {message.topic}: {message.payload}") - match = re.search(r'/brightness/(\w+)', message.topic) + match = re.search(r"/brightness/(\w+)", message.topic) brightness_type = match.group(1) entity: NumberEntity = self.hub.brightness_entities[brightness_type] @@ -93,14 +84,11 @@ async def mqtt_brightness_callback(self, message: ReceiveMessage): async def mqtt_button_callback(self, message: ReceiveMessage): # Handle the message here _LOGGER.debug(f"Received message on topic {message.topic}: {message.payload}") - match = re.search(r'/(\d+)/click', message.topic) + match = re.search(r"/(\d+)/click", message.topic) btn_id = int(match.group(1)) if match else None entity: ButtonEntity = self.hub.button_entities[str(btn_id)] await self.hass.services.async_call( - "button", - 'press', - target={"entity_id": entity.entity_id} + "button", "press", target={"entity_id": entity.entity_id} ) - diff --git a/custom_components/button_plus/device.py b/custom_components/button_plus/device.py index 1979480..a8e3b93 100644 --- a/custom_components/button_plus/device.py +++ b/custom_components/button_plus/device.py @@ -10,7 +10,13 @@ class BarModuleDevice: - def __init__(self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub, connector_id: int) -> None: + def __init__( + self, + hass: HomeAssistant, + entry: ConfigEntry, + hub: ButtonPlusHub, + connector_id: int, + ) -> None: self.device_registry = dr.async_get(hass) self.device = self.device_registry.async_get_or_create( @@ -20,14 +26,15 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub, model="Bar module", manufacturer=MANUFACTURER, suggested_area=hub.config.core.location, - identifiers={ - (DOMAIN, f"{hub._hub_id} BAR Module {connector_id}") - }, - via_device=(DOMAIN, hub.hub_id) + identifiers={(DOMAIN, f"{hub._hub_id} BAR Module {connector_id}")}, + via_device=(DOMAIN, hub.hub_id), ) + class DisplayModuleDevice: - def __init__(self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub) -> None: + def __init__( + self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub + ) -> None: self.device_registry = dr.async_get(hass) self.device = self.device_registry.async_get_or_create( @@ -37,9 +44,6 @@ def __init__(self, hass: HomeAssistant, entry: ConfigEntry, hub: ButtonPlusHub) model="Display Module", manufacturer=MANUFACTURER, suggested_area=hub.config.core.location, - identifiers={ - (DOMAIN, f"{hub._hub_id} Display Module") - }, - via_device=(DOMAIN, hub.hub_id) + identifiers={(DOMAIN, f"{hub._hub_id} Display Module")}, + via_device=(DOMAIN, hub.hub_id), ) - diff --git a/custom_components/button_plus/light.py b/custom_components/button_plus/light.py index 1939c28..ac9cd68 100644 --- a/custom_components/button_plus/light.py +++ b/custom_components/button_plus/light.py @@ -1,4 +1,5 @@ -""" Platform for light integration. """ +"""Platform for light integration.""" + from __future__ import annotations import logging @@ -18,9 +19,9 @@ async def async_setup_entry( - hass: HomeAssistant, - config_entry: ConfigEntry, - async_add_entities: AddEntitiesCallback, + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, ) -> None: """Add switches for passed config_entry in HA.""" @@ -41,9 +42,9 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub, light_type: str): self._hub = hub self._hub_id = hub.hub_id self._light_type = light_type - self._attr_unique_id = f'light-{light_type}-{self._hub_id}-{btn_id}' + self._attr_unique_id = f"light-{light_type}-{self._hub_id}-{btn_id}" self.entity_id = f"light.{light_type}_{self._hub_id}_{btn_id}" - self._attr_name = f'light-{light_type}-{btn_id}' + self._attr_name = f"light-{light_type}-{btn_id}" self._state = False self._connector = hub.config.info.connectors[btn_id // 2] @@ -54,20 +55,26 @@ def is_on(self) -> bool | None: return self._state async def async_turn_on(self, **kwargs: Any) -> None: - """Instruct the light to turn on. """ + """Instruct the light to turn on.""" # Need to apply mqtt logic here to turn on led - _LOGGER.debug(f"Turn on {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})") + _LOGGER.debug( + f"Turn on {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})" + ) async def async_turn_off(self, **kwargs: Any) -> None: """Instruct the light to turn off.""" # Need to apply mqtt logic here to turn off led - _LOGGER.debug(f"Turn off {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})") + _LOGGER.debug( + f"Turn off {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})" + ) def update(self) -> None: """Fetch new state data for this light.""" # get latest stats from mqtt for this light # then update self._state - _LOGGER.debug(f"Update {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})") + _LOGGER.debug( + f"Update {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})" + ) @property def device_info(self): @@ -80,20 +87,29 @@ def device_info(self): match self._connector.connector_type: case 1: device_info["name"] = f"BAR Module {self._connector.connector_id}" - device_info["connections"] = {("bar_module", self._connector.connector_id)} + device_info["connections"] = { + ("bar_module", self._connector.connector_id) + } device_info["model"] = "BAR Module" - device_info["identifiers"] = {(DOMAIN, f'{self._hub.hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}')} + device_info["identifiers"] = { + ( + DOMAIN, + f"{self._hub.hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}", + ) + } case 2: - device_info["name"] = f"Display Module" + device_info["name"] = "Display Module" device_info["connections"] = {("display_module", 1)} device_info["model"] = "Display Module" - device_info["identifiers"] = {(DOMAIN, f'{self._hub.hub_id}_{self._btn_id}_display_module')} + device_info["identifiers"] = { + (DOMAIN, f"{self._hub.hub_id}_{self._btn_id}_display_module") + } return device_info class ButtonPlusWallLight(ButtonPlusLight): - """ Wall light entity representation """ + """Wall light entity representation""" def __init__(self, btn_id: int, hub: ButtonPlusHub): super().__init__(btn_id, hub, "wall") @@ -101,11 +117,11 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub): @property def name(self) -> str: """Return the display name of this light.""" - return f'LED Wall {self._btn_id}' + return f"LED Wall {self._btn_id}" class ButtonPlusFrontLight(ButtonPlusLight): - """ Wall light entity representation """ + """Wall light entity representation""" def __init__(self, btn_id: int, hub: ButtonPlusHub): super().__init__(btn_id, hub, "front") @@ -113,4 +129,4 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub): @property def name(self) -> str: """Return the display name of this light.""" - return f'LED Front {self._btn_id}' + return f"LED Front {self._btn_id}" diff --git a/custom_components/button_plus/number.py b/custom_components/button_plus/number.py index 746e5aa..4453a84 100644 --- a/custom_components/button_plus/number.py +++ b/custom_components/button_plus/number.py @@ -1,4 +1,5 @@ -""" Platform for light integration. """ +"""Platform for light integration.""" + from __future__ import annotations import logging @@ -17,10 +18,11 @@ _LOGGER = logging.getLogger(__name__) + async def async_setup_entry( - hass: HomeAssistant, - config_entry: ConfigEntry, - async_add_entities: AddEntitiesCallback, + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, ) -> None: """Add switches for passed config_entry in HA.""" @@ -28,9 +30,11 @@ async def async_setup_entry( hub: ButtonPlusHub = hass.data[DOMAIN][config_entry.entry_id] - min_version = '1.11' + min_version = "1.11" if version.parse(hub.config.info.firmware) < version.parse(min_version): - _LOGGER.info(f"Current version {hub.config.info.firmware} doesn't support the brightness, it must be at least firmware version {min_version}") + _LOGGER.info( + f"Current version {hub.config.info.firmware} doesn't support the brightness, it must be at least firmware version {min_version}" + ) return _LOGGER.debug(f"Creating number with parameters: {hub.hub_id}") @@ -45,18 +49,17 @@ async def async_setup_entry( async_add_entities(brightness) - class ButtonPlusBrightness(NumberEntity): def __init__(self, hub: ButtonPlusHub, brightness_type: str, event_type: EventType): self._hub = hub self._hub_id = hub.hub_id self._brightness_type = brightness_type self.entity_id = f"brightness.{brightness_type}_{self._hub_id}" - self._attr_name = f'brightness-{brightness_type}' + self._attr_name = f"brightness-{brightness_type}" self.event_type = event_type self._topics = hub.config.core.topics self._attr_icon = "mdi:television-ambient-light" - self._attr_unique_id = f'brightness_{brightness_type}-{self._hub_id}' + self._attr_unique_id = f"brightness_{brightness_type}-{self._hub_id}" @property def native_max_value(self) -> float: @@ -74,7 +77,9 @@ def update(self) -> None: """Fetch new state data for this light.""" # get latest stats from mqtt for this light # then update self._state - _LOGGER.debug(f"Update {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})") + _LOGGER.debug( + f"Update {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})" + ) @property def device_info(self) -> DeviceInfo: @@ -88,7 +93,7 @@ def device_info(self) -> DeviceInfo: identifiers = { (DOMAIN, f"{self._hub.hub_id} BAR Module 1"), (DOMAIN, f"{self._hub.hub_id} BAR Module 2"), - (DOMAIN, f"{self._hub.hub_id} BAR Module 3") + (DOMAIN, f"{self._hub.hub_id} BAR Module 3"), } case EventType.BRIGHTNESS_LARGE_DISPLAY: identifiers = {(DOMAIN, f"{self._hub.hub_id} Display Module")} @@ -97,20 +102,22 @@ def device_info(self) -> DeviceInfo: identifiers=identifiers, ) - - async def async_set_value(self, value: float) -> None: """Set the text value and publish to mqtt.""" label_topic = f"buttonplus/{self._hub_id}/brightness/{self._brightness_type}" _LOGGER.debug(f"ButtonPlus brightness update for {self.entity_id}") - _LOGGER.debug(f"ButtonPlus brightness update to {label_topic} with new value: {value}") - await mqtt.async_publish(hass=self.hass, topic=label_topic, payload=value, qos=0, retain=True) + _LOGGER.debug( + f"ButtonPlus brightness update to {label_topic} with new value: {value}" + ) + await mqtt.async_publish( + hass=self.hass, topic=label_topic, payload=value, qos=0, retain=True + ) self._attr_native_value = value self.async_write_ha_state() class ButtonPlusMiniBrightness(ButtonPlusBrightness): - """ Numeric entity representation """ + """Numeric entity representation""" def __init__(self, hub: ButtonPlusHub): super().__init__(hub, "mini", EventType.BRIGHTNESS_MINI_DISPLAY) @@ -118,11 +125,11 @@ def __init__(self, hub: ButtonPlusHub): @property def name(self) -> str: """Return the display name of this light.""" - return 'Brightness mini display' + return "Brightness mini display" class ButtonPlusLargeBrightness(ButtonPlusBrightness): - """ Numeric entity representation """ + """Numeric entity representation""" def __init__(self, hub: ButtonPlusHub): super().__init__(hub, "large", EventType.BRIGHTNESS_LARGE_DISPLAY) @@ -130,4 +137,4 @@ def __init__(self, hub: ButtonPlusHub): @property def name(self) -> str: """Return the display name of this light.""" - return 'Brightness large display' + return "Brightness large display" diff --git a/custom_components/button_plus/switch.py b/custom_components/button_plus/switch.py index 25912ff..0294345 100644 --- a/custom_components/button_plus/switch.py +++ b/custom_components/button_plus/switch.py @@ -1,9 +1,10 @@ -""" Platform for switch integration. """ +"""Platform for switch integration.""" + from __future__ import annotations import logging -from homeassistant.components.switch import (SwitchEntity, SwitchDeviceClass) +from homeassistant.components.switch import SwitchEntity, SwitchDeviceClass from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback @@ -19,9 +20,9 @@ async def async_setup_entry( - hass: HomeAssistant, - config_entry: ConfigEntry, - async_add_entities: AddEntitiesCallback, + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, ) -> None: """Add switches for passed config_entry in HA.""" @@ -33,7 +34,9 @@ async def async_setup_entry( if connector.connector_type_enum() in [ConnectorEnum.DISPLAY, ConnectorEnum.BAR] ] - buttons = filter(lambda b: b.button_id // 2 in active_connectors,hub.config.mqtt_buttons) + buttons = filter( + lambda b: b.button_id // 2 in active_connectors, hub.config.mqtt_buttons + ) for button in buttons: # _LOGGER.debug(f"Creating switch with parameters: {button.button_id} {button.label} {hub.hub_id}") @@ -48,10 +51,10 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub): self._hub_id = hub.hub_id self._hub = hub self._btn_id = btn_id - self._attr_unique_id = f'switch-{self._hub_id}-{btn_id}' + self._attr_unique_id = f"switch-{self._hub_id}-{btn_id}" self.entity_id = f"switch.{self._hub_id}_{btn_id}" - self._attr_name = f'switch-{btn_id}' - self._name = f'Button {btn_id}' + self._attr_name = f"switch-{btn_id}" + self._name = f"Button {btn_id}" self._device_class = SwitchDeviceClass.SWITCH self._connector = hub.config.info.connectors[btn_id // 2] @@ -70,15 +73,26 @@ def device_info(self): match self._connector.connector_type_enum(): case ConnectorEnum.BAR: - device_info["name"] = f"{self._hub._name} BAR Module {self._connector.connector_id}" - device_info["connections"] = {("bar_module", self._connector.connector_id)} + device_info["name"] = ( + f"{self._hub._name} BAR Module {self._connector.connector_id}" + ) + device_info["connections"] = { + ("bar_module", self._connector.connector_id) + } device_info["model"] = "BAR Module" - device_info["identifiers"] = {(DOMAIN, f'{self._hub.hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}')} + device_info["identifiers"] = { + ( + DOMAIN, + f"{self._hub.hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}", + ) + } case ConnectorEnum.DISPLAY: device_info["name"] = f"{self._hub._name} Display Module" device_info["connections"] = {("display_module", 1)} device_info["model"] = "Display Module" - device_info["identifiers"] = {(DOMAIN, f'{self._hub.hub_id}_{self._btn_id}_display_module')} + device_info["identifiers"] = { + (DOMAIN, f"{self._hub.hub_id}_{self._btn_id}_display_module") + } return device_info diff --git a/custom_components/button_plus/text.py b/custom_components/button_plus/text.py index 3e50d73..718bc18 100644 --- a/custom_components/button_plus/text.py +++ b/custom_components/button_plus/text.py @@ -1,7 +1,9 @@ -""" Platform for text integration. """ +"""Platform for text integration.""" + from __future__ import annotations import logging +from typing import Any from homeassistant.components.text import TextEntity from homeassistant.config_entries import ConfigEntry @@ -19,9 +21,9 @@ async def async_setup_entry( - hass: HomeAssistant, - config_entry: ConfigEntry, - async_add_entities: AddEntitiesCallback, + hass: HomeAssistant, + config_entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, ) -> None: """Add text entity for each top and main label from config_entry in HA.""" @@ -34,11 +36,14 @@ async def async_setup_entry( if connector.connector_type_enum() in [ConnectorEnum.DISPLAY, ConnectorEnum.BAR] ] - buttons = filter(lambda b: b.button_id // 2 in active_connectors, hub.config.mqtt_buttons) + buttons = filter( + lambda b: b.button_id // 2 in active_connectors, hub.config.mqtt_buttons + ) for button in buttons: _LOGGER.debug( - f"Creating Texts with parameters: {button.button_id} {button.top_label} {button.label} {hub.hub_id}") + f"Creating Texts with parameters: {button.button_id} {button.top_label} {button.label} {hub.hub_id}" + ) label_entity = ButtonPlusLabel(button.button_id, hub, button.label) top_label_entity = ButtonPlusTopLabel(button.button_id, hub, button.top_label) @@ -59,7 +64,7 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub, btn_label: str, text_type: s self._hub_id = hub.hub_id self._text_type = text_type self.entity_id = f"text.{text_type}_{self._hub_id}_{btn_id}" - self._attr_name = f'text-{text_type}-{btn_id}' + self._attr_name = f"text-{text_type}-{btn_id}" self._attr_native_value = btn_label self._connector = hub.config.info.connectors[btn_id // 2] self.unique_id = self.unique_id_gen() @@ -72,10 +77,10 @@ def unique_id_gen(self): return self.unique_id_gen_display() def unique_id_gen_bar(self): - return f'text_{self._hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}_{self._text_type}' + return f"text_{self._hub_id}_{self._btn_id}_bar_module_{self._connector.connector_id}_{self._text_type}" def unique_id_gen_display(self): - return f'text_{self._hub_id}_{self._btn_id}_display_module_{self._text_type}' + return f"text_{self._hub_id}_{self._btn_id}_display_module_{self._text_type}" @property def should_poll(self) -> bool: @@ -85,7 +90,9 @@ def update(self) -> None: """Fetch new state data for this label.""" # get latest stats from mqtt for this label # then update self._state - _LOGGER.debug(f"Update {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})") + _LOGGER.debug( + f"Update {self.name} (attr_name: {self._attr_name}) (unique: {self._attr_unique_id})" + ) @property def device_info(self) -> DeviceInfo: @@ -95,7 +102,12 @@ def device_info(self) -> DeviceInfo: match self._connector.connector_type_enum(): case ConnectorEnum.BAR: - identifiers = {(DOMAIN, f"{self._hub.hub_id} BAR Module {self._connector.connector_id}")} + identifiers = { + ( + DOMAIN, + f"{self._hub.hub_id} BAR Module {self._connector.connector_id}", + ) + } case ConnectorEnum.DISPLAY: identifiers = {(DOMAIN, f"{self._hub.hub_id} Display Module")} @@ -105,18 +117,26 @@ def device_info(self) -> DeviceInfo: async def async_set_value(self, value: str) -> None: """Set the text value and publish to mqtt.""" - parse_value: Any = template.Template(value, self.hass).async_render(parse_result=False) + parse_value: Any = template.Template(value, self.hass).async_render( + parse_result=False + ) - label_topic = f"buttonplus/{self._hub_id}/button/{self._btn_id}/{self._text_type}" + label_topic = ( + f"buttonplus/{self._hub_id}/button/{self._btn_id}/{self._text_type}" + ) _LOGGER.debug(f"ButtonPlus label update for {self.entity_id}") - _LOGGER.debug(f"ButtonPlus label update to {label_topic} with new value: {parse_value}") - await mqtt.async_publish(hass=self.hass, topic=label_topic, payload=parse_value, qos=0, retain=True) + _LOGGER.debug( + f"ButtonPlus label update to {label_topic} with new value: {parse_value}" + ) + await mqtt.async_publish( + hass=self.hass, topic=label_topic, payload=parse_value, qos=0, retain=True + ) self._attr_native_value = parse_value self.async_write_ha_state() class ButtonPlusLabel(ButtonPlusText): - """ Wall label entity representation """ + """Wall label entity representation""" def __init__(self, btn_id: int, hub: ButtonPlusHub, label: str): super().__init__(btn_id, hub, label, "label") @@ -124,11 +144,11 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub, label: str): @property def name(self) -> str: """Return the display name of this label.""" - return f'Label {self._btn_id}' + return f"Label {self._btn_id}" class ButtonPlusTopLabel(ButtonPlusText): - """ Wall label entity representation """ + """Wall label entity representation""" def __init__(self, btn_id: int, hub: ButtonPlusHub, label: str): super().__init__(btn_id, hub, label, "top_label") @@ -136,4 +156,4 @@ def __init__(self, btn_id: int, hub: ButtonPlusHub, label: str): @property def name(self) -> str: """Return the display name of this label.""" - return f'Top Label {self._btn_id}' + return f"Top Label {self._btn_id}"