Add code to set SDSSC2BV and update mapping

notebooks/20231011_sdss5_targeting.ipynb

@@ -33,7 +33,7 @@
     "\n",
     "# Let's load the Astra allVisits file.\n",
     "# At the time of writing (2023-10-11) this contains targeting flags for BOSS spectra (v6_1_1) and APOGEE DR17 (no APOGEE SDSS-V).\n",
-    "image = fits.open(os.path.expandvars(\"$MWM_ASTRA/0.4.4/summary/mwmAllVisit-0.4.4.fits\"))"
+    "image = fits.open(os.path.expandvars(\"$MWM_ASTRA/ipl-3/summary/mwmAllVisit-0.5.0.fits\"))"
    ]
   },
   {
@@ -80,7 +80,11 @@
    "outputs": [],
    "source": [
     "# Let's play with flags for spectra observed by BOSS.\n",
-    "flags = TargetingFlags(image[1].data[\"SDSS5_TARGET_FLAGS\"])"
+    "try:\n",
+    "    SDSSC2BV = image[1].header.get('SDSSC2BV',default=1)\n",
+    "except:\n",
+    "    SDSSC2BV = 1\n",
+    "flags = TargetingFlags(image[1].data[\"SDSS5_TARGET_FLAGS\"], sdssc2bv = SDSSC2BV)"
    ]
   },
   {

notebooks/20231011_sdss5_targeting_creation.ipynb

@@ -108,7 +108,7 @@
     "    try:\n",
     "        flags_dict[sdss_id]\n",
     "    except KeyError:\n",
-    "        flags_dict[sdss_id] = TargetingFlags()\n",
+    "        flags_dict[sdss_id] = TargetingFlags(sdssc2bv=2)\n",
     "        \n",
     "    flags_dict[sdss_id].set_bit_by_carton_pk(0, carton_pk) # 0 since this is the only object\n",
     "    manual_counts.setdefault(carton_pk, set())\n",
@@ -158,6 +158,7 @@
     "        fits.Column(name=\"SDSS5_TARGET_FLAGS\", array=flags.array, format=f\"{F}B\", dim=f\"({F})\")\n",
     "    ])\n",
     "])\n",
+    "hdul[1].header['SDSSC2BV'] = 2\n",
     "hdul.writeto(\"output.fits\")"
    ]
   }

python/sdss_semaphore/__init__.py

@@ -82,6 +82,10 @@ def dtype(self):
     def n_bits(self):
         raise NotImplementedError(f"`n_bits` must be defined in subclass")
 
+    @property
+    def MAPPING_BASENAME(self):
+        raise NotImplementedError(f"`MAPPING_BASENAME` must be defined in subclass")
+
     @cached_class_property
     def mapping(self) -> dict:
         """A dictionary containing bit positions as keys, and dictionaries of flag attributes as values."""

python/sdss_semaphore/targeting.py

@@ -1,7 +1,9 @@
 import numpy as np
-from typing import Tuple
-from os import getenv
+from typing import Union, Tuple, Iterable, Optional
+import os
+import warnings
 from sdss_semaphore import BaseFlags, cached_class_property
+import importlib.resources as resources
 
 class BaseTargetingFlags(BaseFlags):
 
@@ -132,14 +134,63 @@ def bit_position_from_carton_pk(self):
         """
         return { attrs["carton_pk"]: bit for bit, attrs in self.mapping.items() }
 
+    @property
+    def version(self):
+        raise NotImplementedError(f"`version` must be defined in subclass")
+
+    @cached_class_property
+    def _ver_name(self):
+        raise NotImplementedError(f"`ver_name` must be defined in subclass")
+
+    @cached_class_property
+    def _MAPPING_BASENAME(self):
+        raise NotImplementedError(f"`_MAPPING_BASENAME` must be defined in subclass")
+
+    @classmethod
+    def set_version(cls, version: int):
+        """Set the SDSSC2BV value and reload the mapping."""
+        MAPPING_BASENAME = cls._MAPPING_BASENAME.format(version = version)
+        try: #python 3.9+
+            path = resources.files(__name__.split('.')[0]).joinpath('etc',f'{MAPPING_BASENAME}')
+        except: #python 3.7,3.8
+            with resources.path(__name__.split('.')[0]+'.etc', f'{MAPPING_BASENAME}') as path:
+                path = str(path)
+
+        if not os.path.exists(path):
+            warnings.warn(
+                f"{cls._ver_name} = {version} is invalid, defaulting to {cls._ver_name} = {cls.version}",
+                InvalidVersionWarning)
+            cls.MAPPING_BASENAME = cls._MAPPING_BASENAME.format(version = cls.version)
+            return
+        cls.version = version
+        cls.MAPPING_BASENAME = MAPPING_BASENAME
+
+        # Clear the cached mapping so it will be reloaded
+        if hasattr(cls, '_mapping_'):
+            del cls._mapping_
+
+class InvalidVersionWarning(Warning):
+    """Custom warning for invalid version values."""
+    pass
+
 class TargetingFlags(BaseTargetingFlags):
 
     """Communicating with SDSS-V targeting flags."""
 
     dtype, n_bits = (np.uint8, 8)
-    MAPPING_BASENAME = f"sdss5_target_{getenv('SDSSC2VB', default=2)}_with_groups.csv"
-
+    version = 2
+    _ver_name = 'SDSSC2BV'
+    _MAPPING_BASENAME = "sdss5_target_{version}_with_groups.csv"
+
     # TODO: Metadata about mapping version should be stored in the MAPPING_BASENAME file
     #       and be assigned as a cached class property once the file is loaded.
 
     # TODO: Update this file once we have finalised the format and content.
+
+    def __init__(self, array: Optional[Union[np.ndarray, Iterable[Iterable[int]], Iterable[bytearray], Iterable['BaseFlags']]] = None,
+                 sdssc2bv: Optional[int] = None) -> None:
+        """Initialize TargetingFlags with an optional sdssc2bv value."""
+        if sdssc2bv is None:
+            sdssc2bv = int(os.getenv('SDSSC2BV', default=2))
+        self.set_version(sdssc2bv)
+        super().__init__(array)