Merge pull request #24 from linien-org/develop

Author: bleykauf

.pre-commit-config.yaml

@@ -1,9 +1,8 @@
 repos:
   - repo: https://github.com/psf/black
-    rev: 22.3.0
+    rev: 22.10.0
     hooks:
       - id: black
-        language_version: python3.7
 
   - repo: https://github.com/pycqa/isort
     rev: 5.10.1

linien_influxdb/app.py

@@ -1,18 +1,19 @@
 from configparser import ConfigParser
 from time import sleep
+from typing import Any, Dict, List
 
 import click
 from influxdb_client import InfluxDBClient, Point
 from linien.client.connection import LinienClient, MHz, Vpp
 
 
 class LinienConnection:
-    def __init__(self, host, username="root", password="root"):
+    def __init__(self, host: str, username: str = "root", password: str = "root"):
         self.client = LinienClient(
             {"host": host, "username": username, "password": password}
         )
 
-    def get_parameters(self, parameters):
+    def get_parameters(self, parameters: List[str]) -> Dict[str, Any]:
 
         # get the `signal_stats` parameter only once and only if required
         stats_suffices = ["_min", "_max", "_mean", "_std"]
@@ -59,7 +60,7 @@ def get_parameters(self, parameters):
     is_flag=True,
     help="Do not write the logged data to influxdb.",
 )
-def main(config, print_only):
+def main(config: str, print_only: bool):
     """Logs Linien parameters according to the configuration in CONFIG ini file."""
     config_parser = ConfigParser(
         converters={"list": lambda x: [i.strip() for i in x.split(",")]}
@@ -70,12 +71,12 @@ def main(config, print_only):
         bucket = config_parser["influx2"]["bucket"]
         measurement = config_parser["influx2"]["measurement"]
     interval = config_parser["linien"].getfloat("interval")
-    parameters = config_parser["linien"].getlist("parameters")
+    requested_parameters = config_parser["linien"].getlist("parameters")
     host = config_parser["linien"]["host"]
     username = config_parser["linien"]["username"]
     password = config_parser["linien"]["password"]
 
-    if len(parameters) == 1 and parameters[0] == "":
+    if len(requested_parameters) == 1 and requested_parameters[0] == "":
         raise ValueError(
             "No parameters requested. Add at least one parameter in the .ini file."
         )
@@ -87,15 +88,16 @@ def main(config, print_only):
             write_api = client.write_api()
             while True:
                 point = Point(measurement)
-                parameters = connection.get_parameters(parameters=parameters)
+                parameters = connection.get_parameters(parameters=requested_parameters)
                 print(parameters)
-                for key, value in parameters.items():
+                for key, value in requested_parameters.items():
                     point.field(key, value)
                 write_api.write(bucket=bucket, record=point)
                 sleep(interval)
     else:
         while True:
-            parameters = connection.get_parameters(parameters=parameters)
+            parameters = connection.get_parameters(parameters=requested_parameters)
+            print(parameters)
             sleep(interval)
 
 

linien_influxdb/example_config.ini

@@ -11,5 +11,5 @@ measurement=linien-status
 host=rp-f0xxxxx.local
 username=root
 password=root
-interval=10 # log interval in seconds
+interval=10
 parameters=lock, modulation_amplitude, modulation_frequency, p, i, d, monitor_signal_mean, error_signal_mean, error_signal_std, control_signal_mean