Merge pull request #6 from panoptes/add-image-and-obs-status

Add image and observation status enums

CHANGELOG.md

@@ -1,6 +1,33 @@
 # Changelog
 
-## Version 0.1 (development)
+## Version 0.1.3
+
+- Cleanup of images and observations.
+
+## Version 0.1.2
+
+- Notebook updates.
+
+## Version 0.1.1
+
+- Notebook and usability improvements.
+
+## Version 0.1.0
+
+- Minor bump to correspond with new GCP project and working library..
+
+## Version 0.0.9
+
+- Fixing the library to work with new GCP project.
+
+## Version 0.0.8
+
+- Import error fix.
+
+## Version 0.0.7
+
+- Update to pydantic v2 including pydantic-settings.
+- CLI improvements.
 
 ## Version 0.0.6
 

src/panoptes/data/images.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from datetime import datetime
+from enum import IntEnum, auto
 from pathlib import Path
 from typing import Union
 
@@ -12,6 +13,25 @@
 from panoptes.data.settings import PATH_MATCHER
 
 
+class ImageStatus(IntEnum):
+    """The status of an image."""
+    ERROR = auto()
+    MASKED = auto()
+    UNKNOWN = auto()
+    RECEIVING = auto()
+    RECEIVED = auto()
+    UNSOLVED = auto()
+    PROCESSING = auto()
+    CALIBRATING = auto()
+    CALIBRATED = auto()
+    SOLVING = auto()
+    SOLVED = auto()
+    MATCHING = auto()
+    MATCHED = auto()
+    EXTRACTING = auto()
+    EXTRACTED = auto()
+
+
 @dataclass
 class ImagePathInfo:
     """Parse the location path for an image.

src/panoptes/data/observations.py

@@ -1,140 +1,33 @@
 import shutil
 import warnings
-from dataclasses import dataclass
-from datetime import datetime
 from pathlib import Path
-from typing import Union
+from enum import IntEnum, auto
 
 import pandas as pd
 from astropy.nddata import Cutout2D, CCDData
-from astropy.time import Time
 from astropy.utils.data import download_file
 from astropy.wcs import FITSFixedWarning
-from dateutil.parser import parse as parse_date
+from panoptes.utils.images import fits as fits_utils
 from tqdm.auto import tqdm
 
-from panoptes.utils.images import fits as fits_utils
-from panoptes.utils.time import flatten_time
-from panoptes.data.settings import PATH_MATCHER, CloudSettings
+from panoptes.data.settings import CloudSettings
 
 warnings.filterwarnings('ignore', category=FITSFixedWarning)
 
 
-@dataclass
-class ObservationPathInfo:
-    """Parse the location path for an image.
-
-    This is a small dataclass that offers some convenience methods for dealing
-    with a path based on the image id.
-
-    This would usually be instantiated via `path`:
-
-    ..doctest::
-
-        >>> from panoptes.data.observations import ObservationPathInfo
-        >>> bucket_path = 'gs://panoptes-images-background/PAN012/Hd189733/358d0f/20180824T035917/20180824T040118.fits'
-        >>> path_info = ObservationPathInfo(path=bucket_path)
-
-        >>> path_info.id
-        'PAN012_358d0f_20180824T035917_20180824T040118'
-
-        >>> path_info.unit_id
-        'PAN012'
-
-        >>> path_info.sequence_id
-        'PAN012_358d0f_20180824T035917'
-
-        >>> path_info.image_id
-        'PAN012_358d0f_20180824T040118'
-
-        >>> path_info.as_path(base='/tmp', ext='.jpg')
-        '/tmp/PAN012/358d0f/20180824T035917/20180824T040118.jpg'
-
-        >>> ObservationPathInfo(path='foobar')
-        Traceback (most recent call last):
-          ...
-        ValueError: Invalid path received: self.path='foobar'
-
-
-    """
-    path: Union[str, Path] = None
-    unit_id: str = None
-    camera_id: str = None
-    field_name: str = None
-    sequence_time: Union[str, datetime, Time] = None
-    image_time: Union[str, datetime, Time] = None
-
-    def __post_init__(self):
-        """Parse the path when provided upon initialization."""
-        if self.path is not None:
-            path_match = PATH_MATCHER.match(self.path)
-            if path_match is None:
-                raise ValueError(f'Invalid path received: {self.path}')
-
-            self.unit_id = path_match.group('unit_id')
-            self.camera_id = path_match.group('camera_id')
-            self.field_name = path_match.group('field_name')
-            self.sequence_time = Time(parse_date(path_match.group('sequence_time')))
-            self.image_time = Time(parse_date(path_match.group('image_time')))
-
-    @property
-    def id(self):
-        """Full path info joined with underscores"""
-        return self.get_full_id()
-
-    @property
-    def sequence_id(self) -> str:
-        """The sequence id."""
-        return f'{self.unit_id}_{self.camera_id}_{flatten_time(self.sequence_time)}'
-
-    @property
-    def image_id(self) -> str:
-        """The matched image id."""
-        return f'{self.unit_id}_{self.camera_id}_{flatten_time(self.image_time)}'
-
-    def as_path(self, base: Union[Path, str] = None, ext: str = None) -> Path:
-        """Return a Path object."""
-        image_str = flatten_time(self.image_time)
-        if ext is not None:
-            image_str = f'{image_str}.{ext}'
-
-        full_path = Path(self.unit_id, self.camera_id, flatten_time(self.sequence_time), image_str)
-
-        if base is not None:
-            full_path = base / full_path
-
-        return full_path
-
-    def get_full_id(self, sep='_') -> str:
-        """Returns the full path id with the given separator."""
-        return f'{sep}'.join([
-            self.unit_id,
-            self.camera_id,
-            flatten_time(self.sequence_time),
-            flatten_time(self.image_time)
-        ])
-
-    @classmethod
-    def from_fits(cls, fits_file):
-        header = fits_utils.getheader(fits_file)
-        return cls.from_fits_header(header)
-
-    @classmethod
-    def from_fits_header(cls, header):
-        try:
-            new_instance = cls(path=header['FILENAME'])
-        except ValueError:
-            sequence_id = header['SEQID']
-            image_id = header['IMAGEID']
-            unit_id, camera_id, sequence_time = sequence_id.split('_')
-            _, _, image_time = image_id.split('_')
-
-            new_instance = cls(unit_id=unit_id,
-                               camera_id=camera_id,
-                               sequence_time=Time(parse_date(sequence_time)),
-                               image_time=Time(parse_date(image_time)))
-
-        return new_instance
+class ObservationStatus(IntEnum):
+    """The status of an observation."""
+    ERROR = auto()
+    NOT_ENOUGH_FRAMES = auto()
+    UNKNOWN = auto()
+    CREATED = auto()
+    RECEIVING = auto()
+    RECEIVED = auto()
+    PROCESSING = auto()
+    CALIBRATING = auto()
+    CALIBRATED = auto()
+    MATCHING = auto()
+    MATCHED = auto()
 
 
 class ObservationInfo: