Fixed lgtm warnings

README.md

@@ -1,17 +1,26 @@
+[![GitHub Release](https://img.shields.io/github/release/toreamun/amshan-homeassistant?style=for-the-badge)](https://github.com/toreamun/amshan-homeassistant/releases)
+[![Language grade: Python](https://img.shields.io/lgtm/grade/python/g/toreamun/amshan-homeassistant.svg?logo=lgtm&logoWidth=18&style=for-the-badge)](https://lgtm.com/projects/g/toreamun/amshan-homeassistant/context:python)
+[![License](https://img.shields.io/github/license/toreamun/amshan-homeassistant?style=for-the-badge)](LICENSE)
+
 [![hacs_badge](https://img.shields.io/badge/HACS-Default-orange.svg?style=for-the-badge)](https://github.com/custom-components/hacs)
+![Project Maintenance](https://img.shields.io/badge/maintainer-Tore%20Amundsen%20%40toreamun-blue.svg?style=for-the-badge)
 [![buy me a coffee](https://img.shields.io/badge/If%20you%20like%20it-Buy%20me%20a%20coffee-orange.svg?style=for-the-badge)](https://www.buymeacoffee.com/toreamun)
 
 # AMS HAN Home Assistant integration
+
 Integrate HAN-port of Aidon, Kaifa and Kamstrum meters used in Norway with Home Assistant.
 
 This integration supports connecting to MBUS device using serial port or TCP-IP address/port.
 
 ## Conecting MBUS device
+
 You need to have a MBUS slave device connected to to the HAN (Home Area Network) port of your meter. The HAN-port is a RJ45 socket where only pin 1 and 2 is used. Connect wires from pin 1 and 2 to the MBUS slave device. Then connect the MBUS device to your computer. Most devices uses USB to become a serial device when connected. You can then relay (i.e. using net2ser and socat) the signal to TCP/IP if your device is connected to a remote computer.
 
 ## MBUS device
+
 This integration has been tested with several simple USB devices sold on e-bay. Search for MBUS USB slave. Not that some devices uses EVEN parity (default is ODD) when connecting.
 
 ## Setup
+
 Search for AMSHAN on Configuration/Integrations page after installing (most simple is to use HACS).
 Please not that some MBUS serial devices uses EVEN parity (the default is ODD).

custom_components/amshan/__init__.py

@@ -2,8 +2,7 @@
 from asyncio import AbstractEventLoop, BaseProtocol, Queue
 from datetime import datetime
 import logging
-import typing
-from typing import Any, Dict, Mapping, NamedTuple, Union
+from typing import Any, Dict, Mapping, NamedTuple, Union, cast
 
 from amshan import obis_map
 from amshan.meter_connection import (
@@ -137,27 +136,29 @@ async def async_close(
 
 
 def setup_meter_connection(
-    loop: AbstractEventLoop, config: Mapping[str, Any], measure_queue: "Queue[bytes]",
+    loop: AbstractEventLoop,
+    config: Mapping[str, Any],
+    measure_queue: "Queue[bytes]",
 ) -> ConnectionManager:
     """Initialize ConnectionManager using configured connection type."""
     connection_factory = get_connection_factory(loop, config, measure_queue)
     return ConnectionManager(connection_factory)
 
 
 def get_connection_factory(
-    loop: AbstractEventLoop, config: Mapping[str, Any], measure_queue: "Queue[bytes]",
+    loop: AbstractEventLoop,
+    config: Mapping[str, Any],
+    measure_queue: "Queue[bytes]",
 ) -> AsyncConnectionFactory:
     """Get connection factory based on configured connection type."""
 
     async def tcp_connection_factory() -> MeterTransportProtocol:
         connection = await loop.create_connection(
-            lambda: typing.cast(
-                BaseProtocol, SmartMeterFrameContentProtocol(measure_queue)
-            ),
+            lambda: cast(BaseProtocol, SmartMeterFrameContentProtocol(measure_queue)),
             host=config[CONF_TCP_HOST],
             port=config[CONF_TCP_PORT],
         )
-        return typing.cast(MeterTransportProtocol, connection)
+        return cast(MeterTransportProtocol, connection)
 
     async def serial_connection_factory() -> MeterTransportProtocol:
         connection = await serial_asyncio.create_serial_connection(
@@ -172,7 +173,7 @@ async def serial_connection_factory() -> MeterTransportProtocol:
             rtscts=config[CONF_SERIAL_RTSCTS],
             dsrdtr=config[CONF_SERIAL_DSRDTR],
         )
-        return typing.cast(MeterTransportProtocol, connection)
+        return cast(MeterTransportProtocol, connection)
 
     # select tcp or serial connection factory
     connection_factory = (
@@ -200,6 +201,4 @@ def from_measure_data(
         cls, measure_data: Dict[str, Union[str, int, float, datetime]]
     ) -> "MeterInfo":
         """Create MeterInfo from measure_data dictionary."""
-        return cls(
-            *[typing.cast(str, measure_data[key]) for key in METER_DATA_INFO_KEYS]
-        )
+        return cls(*[cast(str, measure_data[key]) for key in METER_DATA_INFO_KEYS])

custom_components/amshan/config_flow.py

@@ -1,15 +1,14 @@
 """Config flow for AMS HAN meter integration."""
-import asyncio
-from asyncio import AbstractEventLoop, Queue
+from asyncio import AbstractEventLoop, Queue, wait_for
 from enum import Enum
 import logging
 import socket
-import typing
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, cast
 from serial import SerialException
 
-from amshan import autodecoder, obis_map
-from homeassistant import config_entries, core, exceptions
+from amshan import obis_map
+from amshan.autodecoder import AutoDecoder
+from homeassistant import config_entries
 from homeassistant.helpers.typing import HomeAssistantType
 import voluptuous as vol
 
@@ -115,7 +114,7 @@ async def async_step_serial_connection(
         """Handle the network connection step."""
         if user_input:
             meter_info = await self._validator.async_validate_connection_input(
-                typing.cast(HomeAssistantType, self.hass).loop,
+                cast(HomeAssistantType, self.hass).loop,
                 ConnectionType.SERIAL,
                 user_input,
             )
@@ -139,7 +138,7 @@ async def async_step_network_connection(
         """Handle the network connection step."""
         if user_input:
             meter_info = await self._validator.async_validate_connection_input(
-                typing.cast(HomeAssistantType, self.hass).loop,
+                cast(HomeAssistantType, self.hass).loop,
                 ConnectionType.NETWORK,
                 user_input,
             )
@@ -175,10 +174,10 @@ def _set_base_error(self, error_key: str) -> None:
 
     async def _async_get_meter_info(self, measure_queue: "Queue[bytes]") -> MeterInfo:
         """Decode meter data stream and return meter information if available."""
-        decoder = autodecoder.AutoDecoder()
+        decoder = AutoDecoder()
 
         for _ in range(MAX_FRAME_SEARCH_COUNT):
-            measure = await asyncio.wait_for(measure_queue.get(), MAX_FRAME_WAIT_TIME)
+            measure = await wait_for(measure_queue.get(), MAX_FRAME_WAIT_TIME)
             decoded_measure = decoder.decode_frame_content(measure)
             if decoded_measure:
                 if (

custom_components/amshan/sensor.py

@@ -2,7 +2,6 @@
 from asyncio import Queue
 from datetime import datetime
 import logging
-import typing
 from typing import (
     Any,
     Callable,
@@ -14,6 +13,7 @@
     Optional,
     Set,
     Union,
+    cast,
 )
 
 from amshan.autodecoder import AutoDecoder
@@ -92,13 +92,25 @@ class NorhanEntity(Entity):
             None, None, None, None, "Meter type"
         ),
         obis_map.NEK_HAN_FIELD_OBIS_LIST_VER_ID: EntitySetup(
-            None, None, None, None, "OBIS List version identifier",
+            None,
+            None,
+            None,
+            None,
+            "OBIS List version identifier",
         ),
         obis_map.NEK_HAN_FIELD_ACTIVE_POWER_IMPORT: EntitySetup(
-            POWER_WATT, None, None, ICON_POWER_IMPORT, "Active power import (Q1+Q4)",
+            POWER_WATT,
+            None,
+            None,
+            ICON_POWER_IMPORT,
+            "Active power import (Q1+Q4)",
         ),
         obis_map.NEK_HAN_FIELD_ACTIVE_POWER_EXPORT: EntitySetup(
-            POWER_WATT, None, None, ICON_POWER_EXPORT, "Active power export (Q2+Q3)",
+            POWER_WATT,
+            None,
+            None,
+            ICON_POWER_EXPORT,
+            "Active power export (Q2+Q3)",
         ),
         obis_map.NEK_HAN_FIELD_REACTIVE_POWER_IMPORT: EntitySetup(
             UNIT_KILO_VOLT_AMPERE_REACTIVE,
@@ -203,13 +215,15 @@ def on_new_measure(
                 self._measure_data = measure_data
                 if _LOGGER.isEnabledFor(logging.DEBUG):
                     _LOGGER.debug(
-                        "Update sensor %s with state %s", self.unique_id, self.state,
+                        "Update sensor %s with state %s",
+                        self.unique_id,
+                        self.state,
                     )
                 self.async_write_ha_state()
 
         # subscribe to update events for this meter
         self._async_remove_dispatcher = async_dispatcher_connect(
-            typing.cast(HomeAssistantType, self.hass),
+            cast(HomeAssistantType, self.hass),
             self._new_measure_signal_name,
             on_new_measure,
         )
@@ -365,7 +379,8 @@ def _add_entities(self, entities: List[NorhanEntity]):
         new_measures = [x.measure_id for x in entities]
         self._known_measures.update(new_measures)
         _LOGGER.debug(
-            "Register new entities for measures: %s", new_measures,
+            "Register new entities for measures: %s",
+            new_measures,
         )
         self._async_add_entities(list(entities), True)
 
@@ -383,7 +398,9 @@ def _create_entities(
                         f"{DOMAIN}_measure_available_meterid_{meter_id}"
                     )
                 entity = NorhanEntity(
-                    measure_id, measure_data, self._new_measure_signal_name,
+                    measure_id,
+                    measure_data,
+                    self._new_measure_signal_name,
                 )
                 new_enitities.append(entity)
             else: