Merge pull request #4 from GermanHydrogen/dev

Integrated application

.github/workflows/tests.yaml

@@ -22,8 +22,7 @@ jobs:
           python-version: '3.11'
           cache: 'poetry'
 
-      - run: poetry install --all-extras
-      - run: poetry run maturin develop --release
+      - run: poetry run maturin develop --release --all-features
       - run: poetry run pytest -v
 
   mypy:
@@ -46,6 +45,5 @@ jobs:
           python-version: '3.11'
           cache: 'poetry'
 
-      - run: poetry install --all-extras
-      - run: poetry run maturin develop --release
+      - run: poetry run maturin develop --release --all-features
       - run: poetry run mypy ./pyobs_cloudcover

example.yaml

@@ -0,0 +1,52 @@
+class: pyobs_cloudcover.application.Application
+
+image_sender: "allskycam"
+
+model:
+  class: pyobs_cloudcover.world_model.SimpleModel
+  a0: 4.81426598e-03
+  F: 2.00000000e+00
+  R: 1.06352627e+03
+  c_x: 7.57115607e+02
+  c_y: 5.11194838e+02
+
+server:
+  url: "localhost"
+  port: 8080
+
+measurement_log:
+  url: ""
+  bucket: ""
+  org: ""
+  token: ""
+
+pipelines:
+  night:
+    alt_interval:
+      start: None
+      end: -18
+
+    options:
+      preprocessor:
+        mask_filepath: "tests/integration/mask.npy"
+        bin_size: 2
+        bkg_sigma_clip: 3.0
+        bkg_box_size: (5, 5)
+
+      catalog:
+        filepath: "tests/integration/catalog.csv"
+        filter:
+          alt: 30.0
+          v_mag: 7.0
+          distance: 0.0
+
+      reverse_matcher:
+        sigma_threshold: 3.0
+        window_size: 6.0
+
+      cloud_map:
+        radius: 50.0
+
+      coverage_info:
+        cloud_threshold: 0.5
+        zenith_radius: 20

pyobs_cloudcover/application.py

@@ -0,0 +1,57 @@
+import datetime
+from typing import Dict, Any
+
+from pyobs.events import NewImageEvent, Event
+from pyobs.modules import Module
+from pyobs.object import get_object
+
+from pyobs_cloudcover.measurement_log.influx import Influx
+from pyobs_cloudcover.pipeline.pipeline_controller_factory import PipelineControllerFactory
+from pyobs_cloudcover.web_api.server import Server
+from pyobs_cloudcover.web_api.server_factory import ServerFactory
+from pyobs_cloudcover.world_model import WorldModel
+
+
+class Application(Module):
+    def __init__(self,
+                 image_sender: str,
+                 model: Dict[str, Any],
+                 server: Dict[str, Any],
+                 measurement_log: Dict[str, Any],
+                 pipelines: Dict[str, Dict[str, Any]]) -> None:
+
+        super(Application, self).__init__()
+
+        self._image_sender = image_sender
+
+        world_model: WorldModel = get_object(model, WorldModel)
+
+        server_factory = ServerFactory(self.observer, world_model)
+        self._server = server_factory(server)
+
+        self._measurement_log = Influx(**measurement_log)
+
+        pipeline_controller_factory = PipelineControllerFactory(self.observer, world_model)
+        self._pipeline_controller = pipeline_controller_factory(pipelines)
+
+    async def open(self) -> None:
+        await super(Application, self).open()
+        await self._server.start()
+
+        await self.comm.register_event(NewImageEvent, self.process_new_image)
+
+    async def process_new_image(self, event: Event, sender: str) -> None:
+        if not isinstance(event, NewImageEvent):
+            return
+
+        if sender != self._image_sender:
+            return
+
+        image = await self.vfs.read_image(event.filename)
+
+        obs_time = datetime.datetime.strptime(image.header["DATE-OBS"], "%Y-%m-%dT%H:%M:%S.%f")
+
+        measurement = self._pipeline_controller(image.data, obs_time)
+
+        if measurement is not None:
+            self._server.set_measurement(measurement)

pyobs_cloudcover/application_factory.py

@@ -0,0 +1,6 @@
+
+
+class ApplicationFactory(object):
+    def __init__(self) -> None:
+        ...
+

pyobs_cloudcover/cloud_coverage_info.py

@@ -1,3 +1,4 @@
+import datetime
 from typing import Optional
 
 import numpy as np
@@ -6,8 +7,9 @@
 
 class CloudCoverageInfo(object):
     def __init__(self, cloud_cover_map: npt.NDArray[np.float_],
-                 total_cover: float, zenith_cover: float, change: Optional[float]) -> None:
+                 total_cover: float, zenith_cover: float, change: Optional[float], obs_time: datetime.datetime) -> None:
         self.cloud_cover_map = cloud_cover_map
         self.total_cover = total_cover
         self.zenith_cover = zenith_cover
         self.change = change
+        self.obs_time = obs_time

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) -> None:
+        data = influxdb_client.Point("measurement")
+
+        data.time(measurement.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) -> None:
+        ...

pyobs_cloudcover/pipeline/night/catalog/catalog_constructor.py

@@ -4,7 +4,7 @@
 
 from pyobs_cloudcover.pipeline.night.catalog.altaz_catalog_loader import AltAzCatalogLoader
 from pyobs_cloudcover.pipeline.night.catalog.pixel_catalog import PixelCatalog
-from pyobs_cloudcover.pipeline.night.world_model.world_model import WorldModel
+from pyobs_cloudcover.world_model import WorldModel
 
 
 class CatalogConstructor(object):

pyobs_cloudcover/pipeline/night/catalog/catalog_constructor_factory.py

@@ -0,0 +1,22 @@
+from astroplan import Observer
+
+from pyobs_cloudcover.pipeline.night.catalog.altaz_catalog_loader import AltAzCatalogLoader
+from pyobs_cloudcover.pipeline.night.catalog.catalog_constructor import CatalogConstructor
+from pyobs_cloudcover.pipeline.night.catalog.catalog_constructor_options import CatalogConstructorOptions
+from pyobs_cloudcover.world_model import WorldModel
+
+
+class CatalogConstructorFactory(object):
+    def __init__(self, options: CatalogConstructorOptions, model: WorldModel, observer: Observer) -> None:
+        self._options = options
+        self._model = model
+        self._observer = observer
+
+    def __call__(self) -> CatalogConstructor:
+        altaz_catalog_loader = AltAzCatalogLoader.from_csv(self._options.filepath)
+        catalog_constructor = CatalogConstructor(
+            altaz_catalog_loader, self._model, self._observer,
+            self._options.alt_filter, self._options.v_mag_filter, self._options.distance_filter
+        )
+
+        return catalog_constructor

pyobs_cloudcover/pipeline/night/catalog/catalog_constructor_options.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+from typing import Dict, Any, Optional
+
+
+class CatalogConstructorOptions(object):
+    def __init__(self, filepath: str, alt_filter: float = 20.0, v_mag_filter: float = 7.5, distance_filter: float = 0.0) -> None:
+        self.filepath = filepath
+        self.alt_filter = alt_filter
+        self.v_mag_filter = v_mag_filter
+        self.distance_filter = distance_filter
+
+    @classmethod
+    def from_dict(cls, options: Dict[str, Any]) -> CatalogConstructorOptions:
+        filepath: str = str(options["filepath"])
+
+        if "filter" not in options:
+            return cls(filepath)
+
+        filter_options: Dict[str, Any] = options["filter"]
+
+        filter_kwargs: Dict[str, float] = {}
+
+        if "alt" in filter_options:
+            filter_kwargs["alt_filter"] = filter_options["alt"]
+
+        if "v_mag" in filter_options:
+            filter_kwargs["v_mag_filter"] = filter_options["v_mag"]
+
+        if "distance" in filter_options:
+            filter_kwargs["distance_filter"] = filter_options["distance"]
+
+        return cls(filepath, **filter_kwargs)

pyobs_cloudcover/pipeline/night/catalog/pixel_catalog.py

@@ -1,13 +1,12 @@
 from __future__ import annotations
 
-import datetime
 from typing import cast
 
 import numpy as np
 import numpy.typing as npt
 
 from pyobs_cloudcover.pipeline.night.catalog.altaz_catalog import AltAzCatalog
-from pyobs_cloudcover.pipeline.night.world_model.world_model import WorldModel
+from pyobs_cloudcover.world_model import WorldModel
 
 
 class PixelCatalog(object):

pyobs_cloudcover/pipeline/night/cloud_coverage_calculator/cloud_info_calculator_factory.py

@@ -0,0 +1,23 @@
+from pyobs_cloudcover.pipeline.night.cloud_coverage_calculator.cloud_info_calculator_options import \
+    CloudInfoCalculatorOptions
+from pyobs_cloudcover.pipeline.night.cloud_coverage_calculator.coverage_calculator import CoverageCalculator
+from pyobs_cloudcover.pipeline.night.cloud_coverage_calculator.coverage_change_calculator import \
+    CoverageChangeCalculator
+from pyobs_cloudcover.pipeline.night.cloud_coverage_calculator.coverage_info_calculator import CoverageInfoCalculator
+from pyobs_cloudcover.pipeline.night.cloud_coverage_calculator.zenith_masker import ZenithMasker
+from pyobs_cloudcover.world_model import WorldModel
+
+
+class CloudInfoCalculatorFactory(object):
+    def __init__(self, options: CloudInfoCalculatorOptions, model: WorldModel):
+        self._options = options
+        self._model = model
+
+    def __call__(self) -> CoverageInfoCalculator:
+        coverage_calculator = CoverageCalculator(self._options.cloud_threshold)
+        coverage_change_calculator = CoverageChangeCalculator()
+        zenith_masker = ZenithMasker(self._options.altitude_limit, self._model)
+        cloud_coverage_info_calculator = CoverageInfoCalculator(coverage_calculator, coverage_change_calculator,
+                                                                zenith_masker)
+
+        return cloud_coverage_info_calculator

pyobs_cloudcover/pipeline/night/cloud_coverage_calculator/cloud_info_calculator_options.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from typing import Dict, Any
+
+
+class CloudInfoCalculatorOptions(object):
+    def __init__(self, cloud_threshold: float, zenith_radius: float) -> None:
+        self.cloud_threshold = cloud_threshold
+        self.altitude_limit = 90 - zenith_radius
+
+    @classmethod
+    def from_dict(cls, options: Dict[str, Any]) -> CloudInfoCalculatorOptions:
+        return CloudInfoCalculatorOptions(**options)

pyobs_cloudcover/pipeline/night/cloud_coverage_calculator/coverage_info_calculator.py

@@ -1,3 +1,4 @@
+import datetime
 from copy import copy
 
 import numpy as np
@@ -16,9 +17,9 @@ def __init__(self, coverage_calculator: CoverageCalculator, coverage_change_calc
         self._coverage_change_calculator = coverage_change_calculator
         self._zenith_masker = zenith_masker
 
-    def __call__(self, cloud_map: npt.NDArray[np.float_]) -> CloudCoverageInfo:
+    def __call__(self, cloud_map: npt.NDArray[np.float_], obs_time: datetime.datetime) -> CloudCoverageInfo:
         change = self._coverage_change_calculator(cloud_map)
         coverage = self._coverage_calculator(cloud_map)
         zenith_coverage = self._coverage_calculator(self._zenith_masker(cloud_map))
 
-        return CloudCoverageInfo(copy(cloud_map), coverage, zenith_coverage, change)
+        return CloudCoverageInfo(copy(cloud_map), coverage, zenith_coverage, change, obs_time)