Added influx measurement log

pyobs_cloudcover/measurement_log/influx.py

@@ -0,0 +1,31 @@
+import datetime
+
+import influxdb_client
+from influxdb_client.client.write_api import SYNCHRONOUS
+
+
+from pyobs_cloudcover.cloud_coverage_info import CloudCoverageInfo
+from pyobs_cloudcover.measurement_log.measurement_log import MeasurementLog
+
+
+class Influx(MeasurementLog):
+    def __init__(self, url: str, bucket: str, org: str, token: str) -> None:
+        self._client = influxdb_client.InfluxDBClient(
+            url=url,
+            token=token,
+            org=org
+        )
+        self._bucket = bucket
+        self._org = org
+
+    def __call__(self, measurement: CloudCoverageInfo, obs_time: datetime.datetime) -> None:
+        data = influxdb_client.Point("measurement")
+
+        data.time(obs_time)
+
+        data.field("total-cover", measurement.total_cover)
+        data.field("zenith-cover", measurement.zenith_cover)
+        data.field("change", measurement.change)
+
+        with self._client.write_api(write_options=SYNCHRONOUS) as write_api:
+            write_api.write(bucket=self._bucket, org=self._org, record=data)

pyobs_cloudcover/measurement_log/measurement_log.py

@@ -0,0 +1,10 @@
+import abc
+import datetime
+
+from pyobs_cloudcover.cloud_coverage_info import CloudCoverageInfo
+
+
+class MeasurementLog(object, metaclass=abc.ABCMeta):
+    @abc.abstractmethod
+    def __call__(self, measurement: CloudCoverageInfo, obs_time: datetime.datetime) -> None:
+        ...

pyproject.toml

@@ -9,6 +9,7 @@ packages = [{include = "pyobs_cloudcover"}]
 [tool.poetry.dependencies]
 python = ">=3.11,<3.12"
 pyobs-core = {version = ">=1.13.0", extras = ["full"]}
+influxdb-client = "^1.42.0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7.4.0"

tests/unit/measurment_log/test_influx.py

@@ -0,0 +1,34 @@
+import datetime
+from unittest.mock import Mock
+
+import influxdb_client
+import numpy as np
+
+from pyobs_cloudcover.cloud_coverage_info import CloudCoverageInfo
+from pyobs_cloudcover.measurement_log.influx import Influx
+
+from influxdb_client.client.write_api import WriteApi
+
+
+def test_call():
+    influxdb_client.client.write_api.WriteApi.write = Mock()
+
+    measurement = CloudCoverageInfo(np.array([]), 0, 1, 2)
+    obs_time = datetime.datetime(2020, 1, 1, 0, 0, 0)
+
+    measurement_log = Influx("", "bucket", "org", "token")
+
+    measurement_log(measurement, obs_time)
+
+    call_args = influxdb_client.client.write_api.WriteApi.write.call_args_list
+
+    kwargs = call_args[0].kwargs
+    assert kwargs["bucket"] == "bucket"
+    assert kwargs["org"] == "org"
+
+    data: influxdb_client.Point = kwargs["record"]
+    assert data._time == obs_time
+    assert data._fields["total-cover"] == 0
+    assert data._fields["zenith-cover"] == 1
+    assert data._fields["change"] == 2
+