From 1a8ed755ad16f558a2451e2aad4bf1eb5a373721 Mon Sep 17 00:00:00 2001 From: Sean-Morrison Date: Mon, 21 Oct 2024 10:57:32 -0500 Subject: [PATCH] Add code to set SDSSC2BV and update mapping --- notebooks/20231011_sdss5_targeting.ipynb | 8 ++- .../20231011_sdss5_targeting_creation.ipynb | 3 +- python/sdss_semaphore/__init__.py | 4 ++ python/sdss_semaphore/targeting.py | 59 +++++++++++++++++-- 4 files changed, 67 insertions(+), 7 deletions(-) diff --git a/notebooks/20231011_sdss5_targeting.ipynb b/notebooks/20231011_sdss5_targeting.ipynb index ace415a..02f94d2 100644 --- a/notebooks/20231011_sdss5_targeting.ipynb +++ b/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)" ] }, { diff --git a/notebooks/20231011_sdss5_targeting_creation.ipynb b/notebooks/20231011_sdss5_targeting_creation.ipynb index 3d9931a..3184f24 100644 --- a/notebooks/20231011_sdss5_targeting_creation.ipynb +++ b/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\")" ] } diff --git a/python/sdss_semaphore/__init__.py b/python/sdss_semaphore/__init__.py index bef1c38..cca4f45 100644 --- a/python/sdss_semaphore/__init__.py +++ b/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.""" diff --git a/python/sdss_semaphore/targeting.py b/python/sdss_semaphore/targeting.py index 9a3f3ce..0bed9b4 100644 --- a/python/sdss_semaphore/targeting.py +++ b/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)