Merge pull request #31 from AnotherDaniel/develop

Pacify the linter

.github/workflows/docker-publish.yml

@@ -28,7 +28,7 @@ jobs:
 
     steps:
       - name: Checkout repository
-        uses: actions/checkout@v3
+        uses: actions/checkout@v4
 
       - name: Setup Docker buildx
         uses: docker/setup-buildx-action@v2

.github/workflows/python-app.yml

@@ -18,9 +18,9 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-    - uses: actions/checkout@v3
+    - uses: actions/checkout@v4
     - name: Set up Python 3.10
-      uses: actions/setup-python@v3
+      uses: actions/setup-python@v4
       with:
         python-version: "3.10"
     - name: Install dependencies
@@ -33,7 +33,7 @@ jobs:
         # stop the build if there are Python syntax errors or undefined names
         flake8 src/ plugins/ tests/ --count --select=E9,F63,F7,F82 --show-source --statistics
         # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
-        flake8 src/ plugins/ tests/ --count --exit-zero --max-complexity=10 --max-line-length=200 --statistics
+        flake8 src/ plugins/ tests/ --count --exit-zero --max-complexity=30 --max-line-length=200 --statistics
     - name: Test with pytest
       run: |
         pytest tests

plugins/sinks/ha_mqtt/ha_mqtt.py

@@ -5,9 +5,8 @@
 import ha_mqtt_discoverable
 from ha_mqtt_discoverable import Settings
 from ha_mqtt_discoverable.sensors import SensorInfo, Sensor, DeviceInfo
-from utils.smasensors import *
-from utils.smahelpers import *
-
+from utils.smasensors import get_sensor_dict
+from utils.smahelpers import status_string
 
 # store for all the ha_mqtt sensor objects
 sensors = defaultdict(Sensor)
@@ -113,7 +112,7 @@ def get_sensor(name, device_info):
         if result is None:
             logging.debug(f"HA-MQTT sensor definition for {key} not found or disabled")
             return None
-            
+
         sensor_info = SensorInfo(unique_id=name,
                                  name=result.get('name'),
                                  unit_of_measurement=result.get('unit_of_measurement'),
@@ -131,16 +130,16 @@ def get_sensor(name, device_info):
 def publish(sensor, value):
     if sensor is None:
         return
-    
+
     publish_value = value
     # get rid of unit, in case our value is a value/unit tuple
     if isinstance(value, tuple):
         publish_value = value[0]
-    
+
     # if there's not unit, we should be able to look-up a string value for the parameter number
     if not sensor._entity.unit_of_measurement and status_string(publish_value):
         publish_value = status_string(publish_value)
-    
+
     sensor.set_state(publish_value)
 
 

plugins/sinks/mqtt/mqtt.py

@@ -48,7 +48,7 @@ def execute(config, get_items, register_callback, do_stop):
     except ValueError as exc:
         logging.fatal(f"MQTT broker configuration error: {str(exc)}")
         return
-    except ConnectionError as exc:
+    except ConnectionError:
         logging.fatal(f"MQTT broker not reachable at address: {config.get(
             'server', 'address')}: {str(config.get('server', 'port'))}")
         return
@@ -108,7 +108,7 @@ def attempt_reconnect(client, delay=2, max_delay=300):
         logging.debug("Attempting to reconnect to MQTT Broker...")
         try:
             client.reconnect()
-        except:
+        except ConnectionError:
             logging.debug(
                 "Reconnecting to MQTT Broker failed. Waiting to retry...")
             time.sleep(delay)

plugins/sources/EVCharger/evcharger.py

@@ -29,6 +29,7 @@ def env_vars(config):
     if os.environ.get('EVCHARGER_PREFIX'):
         config['behavior']['sensorPrefix'] = os.environ.get('EVCHARGER_PREFIX')
 
+
 def execute(config, add_data, dostop):
     env_vars(config)
 
@@ -45,7 +46,7 @@ def execute(config, add_data, dostop):
     session.mount('http://', adapter)
     session.mount('https://', adapter)
 
-    loginurl = f"{config.get('server','protocol')}://{config.get('server','address')}/api/v1/token"
+    loginurl = f"{config.get('server', 'protocol')}://{config.get('server', 'address')}/api/v1/token"
     postdata = {'grant_type': 'password',
                 'username': config.get('server', 'username'),
                 'password': config.get('server', 'password'),
@@ -58,7 +59,7 @@ def execute(config, add_data, dostop):
     # Login & Extract Access-Token
     try:
         response = session.post(loginurl, data=postdata, timeout=5, verify=verify_tls)
-    except requests.exceptions.ConnectTimeout as e:
+    except requests.exceptions.ConnectTimeout:
         logging.fatal(f"Charger not reachable via HTTP: {config.get('server', 'address')}")
         return
     except requests.exceptions.ConnectionError as e:
@@ -76,13 +77,14 @@ def execute(config, add_data, dostop):
     headers = {"Authorization": "Bearer " + token}
 
     # Request Device Info
-    url = f"{config.get('server','protocol')}://{config.get('server','address')}/api/v1/plants/Plant:1/devices/IGULD:SELF"
+    url = f"{config.get('server', 'protocol')}://{config.get('server', 'address')
+                                                  }/api/v1/plants/Plant:1/devices/IGULD:SELF"
     response = session.get(url, headers=headers, verify=verify_tls)
     dev = response.json()
 
     DeviceInfo = {}
     DeviceInfo['name'] = dev['product']
-    DeviceInfo['configuration_url'] = f"{config.get('server','protocol')}://{config.get('server', 'address')}"
+    DeviceInfo['configuration_url'] = f"{config.get('server', 'protocol')}://{config.get('server', 'address')}"
     DeviceInfo['identifiers'] = dev['serial']
     DeviceInfo['model'] = f"{dev['vendor']}-{dev['product']}"
     DeviceInfo['manufacturer'] = dev['vendor']
@@ -97,7 +99,7 @@ def execute(config, add_data, dostop):
             add_data(dname, value)
 
         try:
-            url = f"{config.get('server','protocol')}://{config.get('server', 'address')}/api/v1/measurements/live"
+            url = f"{config.get('server', 'protocol')}://{config.get('server', 'address')}/api/v1/measurements/live"
             response = session.post(url, headers=headers, data='[{"componentId":"IGULD:SELF"}]', verify=verify_tls)
 
             # Check if a new acccess token is neccesary (TODO use refresh token)
@@ -112,7 +114,7 @@ def execute(config, add_data, dostop):
 
             for d in data:
                 # name is the generic parameter/measurement name
-                name = f"{d['channelId'].replace('Measurement.','').replace('[]', '')}"
+                name = f"{d['channelId'].replace('Measurement.', '').replace('[]', '')}"
                 # dname is name, prefixed with device prefix and serial number
                 dname = f"{config.get('behavior', 'sensorPrefix')}{DeviceInfo['identifiers']}.{name}"
                 if "value" in d['values'][0]:
@@ -143,7 +145,7 @@ def execute(config, add_data, dostop):
             time.sleep(int(config.get('behavior', 'updateFreq')))
 
         except TimeoutError:
-            logging.warning(f"Got TimeoutError - retrying")
+            logging.warning("Got TimeoutError - retrying")
             pass
 
         except Exception as e:

plugins/sources/TripowerX/tripowerx.py

@@ -30,6 +30,7 @@ def env_vars(config):
     if os.environ.get('TRIPOWERX_PREFIX'):
         config['behavior']['sensorPrefix'] = os.environ.get('TRIPOWERX_PREFIX')
 
+
 def execute(config, add_data, dostop):
     env_vars(config)
 
@@ -49,7 +50,7 @@ def execute(config, add_data, dostop):
     session.mount('http://', adapter)
     session.mount('https://', adapter)
 
-    loginurl = f"{config.get('server','protocol')}://{config.get('server','address')}/api/v1/token"
+    loginurl = f"{config.get('server', 'protocol')}://{config.get('server', 'address')}/api/v1/token"
     postdata = {'grant_type': 'password',
                 'username': config.get('server', 'username'),
                 'password': config.get('server', 'password'),
@@ -62,7 +63,7 @@ def execute(config, add_data, dostop):
     # Login & Extract Access-Token
     try:
         response = session.post(loginurl, data=postdata, timeout=5, verify=verify_tls)
-    except requests.exceptions.ConnectTimeout as e:
+    except requests.exceptions.ConnectTimeout:
         logging.fatal(f"Inverter not reachable via HTTP: {config.get('server', 'address')}")
         return
     except requests.exceptions.ConnectionError as e:
@@ -80,13 +81,14 @@ def execute(config, add_data, dostop):
     headers = {"Authorization": "Bearer " + token}
 
     # Request Device Info
-    url = f"{config.get('server','protocol')}://{config.get('server','address')}/api/v1/plants/Plant:1/devices/IGULD:SELF"
+    url = f"{config.get('server', 'protocol')}://{config.get('server', 'address')
+                                                  }/api/v1/plants/Plant:1/devices/IGULD:SELF"
     response = session.get(url, headers=headers, verify=verify_tls)
     dev = response.json()
 
     DeviceInfo = {}
     DeviceInfo['name'] = dev['product']
-    DeviceInfo['configuration_url'] = f"{config.get('server','protocol')}://{config.get('server', 'address')}"
+    DeviceInfo['configuration_url'] = f"{config.get('server', 'protocol')}://{config.get('server', 'address')}"
     DeviceInfo['identifiers'] = dev['serial']
     DeviceInfo['model'] = f"{dev['vendor']}-{dev['product']}"
     DeviceInfo['manufacturer'] = dev['vendor']
@@ -101,7 +103,7 @@ def execute(config, add_data, dostop):
             add_data(dname, value)
 
         try:
-            url = f"{config.get('server','protocol')}://{config.get('server', 'address')}/api/v1/measurements/live"
+            url = f"{config.get('server', 'protocol')}://{config.get('server', 'address')}/api/v1/measurements/live"
             response = session.post(url, headers=headers, data='[{"componentId":"IGULD:SELF"}]', verify=verify_tls)
 
             # Check if a new acccess token is neccesary (TODO use refresh token)
@@ -116,7 +118,7 @@ def execute(config, add_data, dostop):
 
             for d in data:
                 # name is the generic parameter/measurement name
-                name = f"{d['channelId'].replace('Measurement.','').replace('[]', '')}"
+                name = f"{d['channelId'].replace('Measurement.', '').replace('[]', '')}"
                 # dname is name, prefixed with device prefix and serial number
                 dname = f"{config.get('behavior', 'sensorPrefix')}{DeviceInfo['identifiers']}.{name}"
                 if "value" in d['values'][0]:
@@ -147,7 +149,7 @@ def execute(config, add_data, dostop):
             time.sleep(int(config.get('behavior', 'updateFreq')))
 
         except TimeoutError:
-            logging.warning(f"Got TimeoutError - retrying")
+            logging.warning("Got TimeoutError - retrying")
             pass
 
         except Exception as e:
@@ -230,7 +232,7 @@ def execute(config, add_data, dostop):
         'name': "cos φ(V), status",
         'suggested_display_precision': 2,
     },
-        {
+    {
         'key': "Operation.BckStt",
         'enabled': "true",
         'name': "Backup mode status",

plugins/sources/demo/demo.py

@@ -40,4 +40,4 @@ def execute(config, add_data, dostop):
         'name': "Demo value",
         'entity_category': "diagnostic",
     },
-]
+]

src/smahub.py

@@ -204,7 +204,6 @@ async def main(args):
             logging.info(f"Holding startup, waiting for debug connection on port: {args.debug_port}")
             debugpy.wait_for_client()
 
-
     # Main business logic
     load_plugins(args.source_dir, sources)
     load_plugins(args.sink_dir, sinks)
@@ -240,10 +239,13 @@ async def main(args):
     parser.add_argument('-V', '--verboser', action='store_true', help='Enable even more verbose output')
     parser.add_argument('-d', '--debug', action='store_true', help='Enable debug server')
     parser.add_argument('-p', '--debug-port', action='store', help='Debug server port', default=5678)
-    parser.add_argument('-D', '--debug-hold', action='store_true', help='Hold startup to wait for debug connection; requires --debug')
+    parser.add_argument('-D', '--debug-hold', action='store_true',
+                        help='Hold startup to wait for debug connection; requires --debug')
     parser.add_argument('--version', action='version', version=f'%(prog)s {version}')
-    parser.add_argument('--source-dir', type=str, default='plugins/sources', help='Path to the directory containing source plugins')
-    parser.add_argument('--sink-dir', type=str, default='plugins/sinks', help='Path to the directory containing sink plugins')
+    parser.add_argument('--source-dir', type=str, default='plugins/sources',
+                        help='Path to the directory containing source plugins')
+    parser.add_argument('--sink-dir', type=str, default='plugins/sinks',
+                        help='Path to the directory containing sink plugins')
 
     # Parse command-line arguments
     args = parser.parse_args()

src/utils/smahelpers.py

@@ -107,33 +107,39 @@
     16777213: "Information not available",
 }
 
-def status_string(id): 
+
+def status_string(id):
     if id in TRIPOWER_STATUS_DICT:
         return TRIPOWER_STATUS_DICT[id]
     else:
         return ""
 
-# DEPRECATED in favor of smasensors.py dynamic sensor data registry!    
+# DEPRECATED in favor of smasensors.py dynamic sensor data registry!
+
+
 def parameter_unit(name):
     if name in TRIPOWER_PARAM_DICT:
         return TRIPOWER_PARAM_DICT[name][2]
     else:
         return ""
 
+
 # DEPRECATED in favor of smasensors.py dynamic sensor data registry!
 def parameter_description(name):
     if name in TRIPOWER_PARAM_DICT:
         return TRIPOWER_PARAM_DICT[name][1]
     else:
         return ""
 
+
 # DEPRECATED in favor of smasensors.py dynamic sensor data registry!
 def parameter_group(name):
     if name in TRIPOWER_PARAM_DICT:
         return TRIPOWER_PARAM_DICT[name][0]
     else:
         return ""
 
+
 # DEPRECATED in favor of smasensors.py dynamic sensor data registry!
 def unit_of_measurement(name):
     if name.endswith("TmpVal"):

src/utils/smasensors.py

@@ -4,6 +4,7 @@
 
 SENSOR_REGISTRY = {}
 
+
 def register_sensor_dict(device, sensor_info):
     """
     Register a sensor dictionary by associating its name with its corresponding attributes.
@@ -21,6 +22,7 @@ def register_sensor_dict(device, sensor_info):
     if device not in SENSOR_REGISTRY:
         SENSOR_REGISTRY[device] = sensor_info
 
+
 def get_sensor_dict(device):
     """
     Retrieve a dictionary of device sensor attributes.
@@ -37,6 +39,7 @@ def get_sensor_dict(device):
     """
     return SENSOR_REGISTRY.get(device, None)
 
+
 def get_parameter_unit(device_name, target_key):
     """
     Retrieve the unit_of_measurement of a particular key in a device's sensor list.
@@ -52,4 +55,4 @@ def get_parameter_unit(device_name, target_key):
     for sensor_dict in sensor_list:
         if sensor_dict['key'] == target_key:
             return sensor_dict.get('unit_of_measurement', None)
-    return None
+    return None