mypy

pyobs/images/processors/_daobackgroundremover.py

@@ -1,6 +1,7 @@
 from typing import Tuple
 
 import numpy as np
+import numpy.typing as npt
 from astropy.stats import SigmaClip
 
 from pyobs.images import Image
@@ -20,18 +21,22 @@ def __call__(self, image: Image) -> Image:
         background = self._estimate_background(image)
         return self._remove_background(image, background)
 
-    def _estimate_background(self, image: Image):
+    def _estimate_background(self, image: Image) -> npt.NDArray[float]:
         from photutils.background import Background2D
 
         bkg = Background2D(
-            image.data, box_size=self._box_size, filter_size=self._filter_size, sigma_clip=self._sigma_clip,
-            bkg_estimator=self._bkg_estimator, mask=image.mask
+            image.data,
+            box_size=self._box_size,
+            filter_size=self._filter_size,
+            sigma_clip=self._sigma_clip,
+            bkg_estimator=self._bkg_estimator,
+            mask=image.mask,
         )
 
         return bkg.background
 
     @staticmethod
-    def _remove_background(image: Image, background: np.ndarray) -> Image:
+    def _remove_background(image: Image, background: npt.NDArray[float]) -> Image:
         output_image = image.copy()
         output_image.data = output_image.data - background
         return output_image

pyobs/images/processors/astrometry/_dotnet_request.py

@@ -1,49 +1,53 @@
-from typing import Dict, Optional
-
+from typing import Dict, Optional, Any
 import aiohttp
 
 import pyobs.utils.exceptions as exc
 
 
 class _DotNetRequest:
-    def __init__(self, request_data: Dict[str, any]):
+    def __init__(self, request_data: Dict[str, Any]):
         self._request_data = request_data
 
-        self._response_data: Optional[Dict[str, any]] = None
+        self._response_data: Optional[Dict[str, Any]] = None
         self._status_code: Optional[int] = None
 
-    async def _send_request(self, url: str, timeout: int):
+    async def _send_request(self, url: str, timeout: int) -> None:
         async with aiohttp.ClientSession() as session:
             async with session.post(url, json=self._request_data, timeout=timeout) as response:
                 self._status_code = response.status
                 self._response_data = await response.json()
 
     def _generate_request_error_msg(self) -> str:
-        if "error" not in self._response_data:
+        if self._response_data is None or "error" not in self._response_data:
             return "Could not connect to astrometry service."
 
         if self._response_data["error"] == "Could not find WCS file.":
             return "Could not determine WCS."
 
         return f"Received error from astrometry service: {self._response_data['error']}"
 
-    def _handle_request_error(self):
+    def _handle_request_error(self) -> None:
         error_msg = self._generate_request_error_msg()
         raise exc.ImageError(error_msg)
 
-    def _is_request_successful(self) -> bool:
-        return self._status_code != 200 or "error" in self._response_data
+    def _has_request_error(self) -> bool:
+        return (
+            self._status_code is None
+            or self._response_data is None
+            or self._status_code != 200
+            or "error" in self._response_data
+        )
 
-    async def send(self, url: str, timeout: int):
+    async def send(self, url: str, timeout: int) -> None:
         await self._send_request(url, timeout)
 
-        if self._is_request_successful():
+        if self._has_request_error():
             self._handle_request_error()
 
     @property
-    def request_data(self) -> Dict[str, any]:
+    def request_data(self) -> Dict[str, Any]:
         return self._request_data
 
     @property
-    def response_data(self) -> Optional[Dict[str, any]]:
+    def response_data(self) -> Optional[Dict[str, Any]]:
         return self._response_data

pyobs/images/processors/astrometry/_dotnet_request_builder.py

@@ -1,4 +1,4 @@
-from typing import Optional
+from typing import Optional, Any, Dict
 
 import pandas as pd
 from astropy.io.fits import Header
@@ -13,27 +13,29 @@ def __init__(self, source_count: int, radius: float):
         self._source_count = source_count
         self._radius = radius
 
-        self._request_data = {}
+        self._request_data: Dict[str, Any] = {}
         self._catalog = pd.DataFrame()
         self._header: Optional[Header] = None
 
-    def _filter_catalog(self):
+    def _filter_catalog(self) -> None:
         self._catalog = self._catalog.dropna(how="any")
         self._catalog = self._catalog[self._catalog["peak"] < 60000]
 
-    def _validate_catalog(self):
+    def _validate_catalog(self) -> None:
         if self._catalog is None or len(self._catalog) < 3:
             raise exc.ImageError("Not enough sources for astrometry.")
 
-    def _select_brightest_stars(self):
+    def _select_brightest_stars(self) -> None:
         self._catalog = self._catalog.sort_values("flux", ascending=False)
         self._catalog = self._catalog[: self._source_count]
 
-    def _validate_header(self):
-        if "CDELT1" not in self._header:
+    def _validate_header(self) -> None:
+        if self._header is None or "CDELT1" not in self._header:
             raise exc.ImageError("No CDELT1 found in header.")
 
-    def _build_request_data(self):
+    def _build_request_data(self) -> None:
+        if self._header is None:
+            raise exc.ImageError("No header found.")
         scale = abs(self._header["CDELT1"]) * 3600
         self._request_data = {
             "ra": self._header["TEL-RA"],

pyobs/images/processors/astrometry/_dotnet_request_logger.py

@@ -14,7 +14,7 @@ def __init__(self, logger: logging.Logger, image: Image, request_data: Dict[str,
         self._image = image
         self._request_data = request_data
 
-    def log_request_data(self):
+    def log_request_data(self) -> None:
         ra_dec = SkyCoord(ra=self._request_data["ra"] * u.deg, dec=self._request_data["dec"] * u.deg, frame="icrs")
         center_x, center_y = self._image.header["CRPIX1"], self._image.header["CRPIX2"]
 
@@ -28,7 +28,7 @@ def log_request_data(self):
             center_y,
         )
 
-    def log_request_result(self, image_wcs: WCS):
+    def log_request_result(self, image_wcs: WCS) -> None:
         center_x, center_y = self._image.header["CRPIX1"], self._image.header["CRPIX2"]
         final_ra, final_dec = image_wcs.all_pix2world(center_x, center_y, 0)
         final_ra_dec = SkyCoord(ra=final_ra * u.deg, dec=final_dec * u.deg, frame="icrs")
@@ -41,4 +41,4 @@ def log_request_result(self, image_wcs: WCS):
             self._request_data["dec"],
             center_x,
             center_y,
-        )
+        )

pyobs/images/processors/astrometry/_dotnet_response_saver.py

@@ -4,6 +4,7 @@
 from astropy.wcs import WCS
 
 from pyobs.images import Image
+import pyobs.utils.exceptions as exc
 
 
 class _ResponseImageWriter:
@@ -14,46 +15,53 @@ def __init__(self, response_data: Dict[str, Any], image: Image):
         self._image_wcs: Optional[WCS] = None
 
     @property
-    def response_data(self):
+    def response_data(self) -> Dict[str, Any]:
         return self._response_data
 
     @property
-    def image(self):
+    def image(self) -> Image:
         return self._image
 
     @property
-    def image_wcs(self):
+    def image_wcs(self) -> WCS:
         return self._image_wcs
 
-    def _write_response_into_header(self):
+    def _write_response_into_header(self) -> None:
         header_keywords_to_update = [
-            "CTYPE1", "CTYPE2",
-            "CRPIX1", "CRPIX2",
-            "CRVAL1", "CRVAL2",
-            "CD1_1", "CD1_2",
-            "CD2_1", "CD2_2",
+            "CTYPE1",
+            "CTYPE2",
+            "CRPIX1",
+            "CRPIX2",
+            "CRVAL1",
+            "CRVAL2",
+            "CD1_1",
+            "CD1_2",
+            "CD2_1",
+            "CD2_2",
         ]
 
         for keyword in header_keywords_to_update:
             self._image.header[keyword] = self._response_data[keyword]
 
-    def _delete_old_wcs_data(self):
+    def _delete_old_wcs_data(self) -> None:
         """
         astrometry.net gives a CD matrix, so we have to delete the PC matrix and the CDELT* parameters
         """
         for keyword in ["PC1_1", "PC1_2", "PC2_1", "PC2_2", "CDELT1", "CDELT2"]:
             del self._image.header[keyword]
 
-    def _generate_image_wcs(self):
+    def _generate_image_wcs(self) -> None:
         self._image_wcs = WCS(self._image.header)
 
-    def _add_plate_solution_to_catalog(self):
+    def _add_plate_solution_to_catalog(self) -> None:
+        if self._image_wcs is None:
+            raise exc.ImageError("No WCS found.")
         ras, decs = self._image_wcs.all_pix2world(self._image.catalog["x"], self._image.catalog["y"], 1)
 
         self._image.catalog["ra"] = ras
         self._image.catalog["dec"] = decs
 
-    def _add_wcs_err_success(self):
+    def _add_wcs_err_success(self) -> None:
         self._image.header["WCSERR"] = 0
 
     def __call__(self, *args, **kwargs) -> Image: