almanac package

Submodules

almanac.apogee module

almanac.apogee.create_apogee_id_lookup()

Create a lookup dictionary mapping APOGEE IDs to SDSS IDs.

Return type:

Dict[str, int]

almanac.apogee.get_almanac_data(observatory, mjd, fibers=False, meta=False, only_assign_sdss_id_to_first_exposure_per_configuration=True)

Return comprehensive almanac data for all exposures taken from a given observatory on a given MJD.

Parameters:

• observatory (str) – The observatory name (e.g. “apo”).

• mjd (int) – The Modified Julian Date.

• fibers (bool) – Whether to include fiber mapping information.

• xmatch – Whether to perform cross-matching with catalog database.

• kwargs – Additional keyword arguments passed to other functions.

Returns:

Tuple containing: - observatory name - MJD - A list of exposures - Table of exposure data - dictionary of sequence indices - dictionary of fiber mappings

almanac.apogee.get_arclamp_sequences(exposures)

Return a list of tuples indicating the start and end exposure numbers for a sequence of arc lamp exposures.

Parameters:

exposures (List[Exposure]) – A list of Exposure instances.

Return type:

List[Tuple[int, int]]

Returns:

List of tuples containing (start_exposure, end_exposure) for each arc lamp sequence.

almanac.apogee.get_expected_number_of_exposures(observatory, mjd)

Query the SDSS database to get the expected exposures for a given observatory and MJD. This is useful for identifying missing exposures.

Return type:

int

almanac.apogee.get_exposures(observatory, mjd)

Generate exposures taken from a given observatory on a given MJD.

Parameters:

• observatory (str) – The observatory name (e.g. “apo”).

• mjd (int) – The Modified Julian Date.

Yields:

Exposure instances for each unique exposure found on disk.

Return type:

Generator[Exposure, None, None]

almanac.apogee.get_science_sequences(exposures)

Return a list of tuples indicating the start and end exposure numbers for a sequence of science exposures.

Parameters:

exposures (List[Exposure]) – A list of Exposure instances.

Return type:

List[Tuple[int, int]]

Returns:

List of tuples containing (start_exposure, end_exposure) for each science sequence.

almanac.apogee.get_sequences(exposures, image_type, fields, ignore_interrupting_darks=False)

Get exposure number ranges for sequences of a specific image type.

Parameters:

• exposures (List[Exposure]) – Astropy Table containing exposure metadata.

• image_type (Literal['blackbody', 'dark', 'object', 'domeflat', 'arclamp', 'twilightflat', 'internalflat', 'quartzflat', 'missing']) – The image type to search for (e.g., “Object”, “ArcLamp”).

• fields (Tuple[str, ...]) – Tuple of column names to group exposures by.

• ignore_interrupting_darks (bool) – [optional] Whether to allow (and ignore) dark exposures interrupting sequences.

Return type:

List[Tuple[int, int]]

Returns:

List of tuples containing (start_exposure, end_exposure) for each sequence.

almanac.apogee.get_unique_exposure_paths(paths)

Process a list of file paths to find unique exposures and determine which chips are available.

Parameters:

paths (List[str]) – List of file paths to APOGEE exposure files.

Return type:

List[str]

Returns:

List of exposure paths.

almanac.apogee.organize_exposures(exposures)

Identify any missing exposures (based on non-contiguous exposure numbers) and fill them with missing image types.

Parameters:

exposures (List[Exposure]) – A list of Exposure instances.

Return type:

List[Exposure]

Returns:

A list of organized Exposure instances.

almanac.apogee.parse_target_identifier(target)

almanac.cli module

almanac.cli.check_paths_and_format(input_path, output_path, given_format, overwrite)

almanac.config module

class almanac.config.Config(sdssdb=<factory>, database_connect_time_warning=3, sdssdb_exposure_min_mjd=<factory>, logging_level=20, platelist_dir='/uufs/chpc.utah.edu/common/home/sdss09/software/svn.sdss.org/data/sdss/platelist/trunk/plates/', sdsscore_dir='/uufs/chpc.utah.edu/common/home/sdss50/software/git/sdss/sdsscore/main/', apogee_dir='/uufs/chpc.utah.edu/common/home/sdss/sdsswork/data/apogee/', mapper_dir='/uufs/chpc.utah.edu/common/home/sdss50/sdsswork/data/mapper/', display_field_names=<factory>)

Bases: object

apogee_dir: str = '/uufs/chpc.utah.edu/common/home/sdss/sdsswork/data/apogee/'

database_connect_time_warning: int = 3

display_field_names: List[str]

logging_level: int = 20

mapper_dir: str = '/uufs/chpc.utah.edu/common/home/sdss50/sdsswork/data/mapper/'

platelist_dir: str = '/uufs/chpc.utah.edu/common/home/sdss09/software/svn.sdss.org/data/sdss/platelist/trunk/plates/'

sdsscore_dir: str = '/uufs/chpc.utah.edu/common/home/sdss50/software/git/sdss/sdsscore/main/'

sdssdb: DatabaseConfig

sdssdb_exposure_min_mjd: ObservatoryMJD

class almanac.config.ConfigManager

Bases: object

A utility class to save and load dataclass configurations using YAML.

static load(cls, file_path)

Loads a dataclass object from a YAML file.

static save(config, file_path)

Saves a dataclass object to a YAML file.

class almanac.config.DatabaseConfig(user='sdss_user', host='operations.sdss.org', port=5432, domain='operations.sdss.*')

Bases: object

domain: str = 'operations.sdss.*'

host: str = 'operations.sdss.org'

port: int = 5432

user: str = 'sdss_user'

class almanac.config.ObservatoryMJD(apo=59558, lco=59558)

Bases: object

apo: int = 59558

lco: int = 59558

almanac.config.get_config_path()

almanac.database module

almanac.display module

class almanac.display.BufferedHandler

Bases: Handler

Custom logging handler that buffers log records

emit(record)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

flush_to_console(console=None)

Flush buffered records to console

class almanac.display.ObservationsDisplay(mjd_min, mjd_max, observatories=('apo', 'lco'))

Bases: object

add_observation(date, observatory)

Add an observation for a specific date and observatory

color_apo = 'dodger_blue3'

color_both = 'purple4'

color_lco = 'green4'

color_missing = 'red'

color_no_data = 'bright_black'

color_outside_range = 'black'

color_unknown = 'white'

create_contributions_grid_for_year(year)

Create the contributions grid for a specific year

create_display()

Create the complete display with title and yearly grids

get_day_color(day_index)

Return the color for a given day based on completion status

class almanac.display.Task(name, total=100, completed=0, status=TaskStatus.PENDING)

Bases: object

completed: int = 0

name: str

status: TaskStatus = 'pending'

total: int = 100

class almanac.display.TaskDisplay

Bases: object

add_task(name, total=100)

Return type:

str

advance(task_id, amount=1)

complete(task_id)

fail(task_id)

start_task(task_id)

update_task(task_id, total=None, name=None)

Update task properties like total or name.

class almanac.display.TaskStatus(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

COMPLETED = 'completed'

FAILED = 'failed'

PENDING = 'pending'

RUNNING = 'running'

almanac.display.display_exposures(exposures, sequences=None, console=None, header_style='bold cyan', column_names=None, sequence_styles=('green', 'yellow'), missing_style='blink bold red', title_style='bold blue')

Display exposure information using Rich table formatting.

Parameters:

• exposures (List[Exposure]) – List of Exposure objects containing exposure data

• sequences (Optional[Dict[str, List[Tuple[int, int]]]]) – Dictionary mapping sequence names to lists of (start, end) tuples (default: None)

• console (Optional[Console]) – Rich Console instance (default: None, creates new one)

• header_style (str) – Style for table headers (default: “bold cyan”)

• sequence_styles (Tuple[str, ...]) – Tuple of styles to cycle through for sequences (default: (“green”, “yellow”))

• missing_style (str) – Style for missing/error entries (default: “red”)

• title_style (str) – Style for the table title (default: “bold blue”)

Return type:

None

almanac.display.mjd_to_datetime(mjd)

Convert MJD to datetime - mock implementation

almanac.io module

almanac.io.convert_value_for_hdf5(value, target_dtype)

Convert a Python value to be compatible with HDF5 storage.

almanac.io.delete_hdf5_entry(fp, group_name)

almanac.io.extract_field_data(models, field_name)

Extract data for a specific field from all models.

Return type:

List[Any]

almanac.io.get_default_array(field_spec, num_records)

almanac.io.get_hdf5_dtype(pydantic_type, sample_value=None)

Map Pydantic field types to appropriate HDF5/NumPy dtypes.

Parameters:

• pydantic_type – The Pydantic field type annotation

• sample_value – A sample value to help determine string lengths, etc.

Returns:

Appropriate NumPy dtype for HDF5

almanac.io.get_or_create_group(fp, group_name)

almanac.io.update(fp, observatory, mjd, exposures, sequences, fibers=False, verbose=False, compression=True)

almanac.io.write_almanac(output, results, fibers=False, verbose=False, compression=True)

Write the results of an Almanac query to an HDF5 file.

Parameters:

• output (str) – Path to the output HDF5 file.

• results (List[Tuple[str, int, List[Exposure], Dict[str, List[Any]]]]) – List of tuples containing (observatory, mjd, exposures, sequences). - observatory: str, e.g., “apo” or “lco” - mjd: int, Modified Julian Date - exposures: List[Exposure], list of Exposure models - sequences: Dict[str, List[Any]], dictionary of sequences by image type

• fibers (bool) – Whether to include fiber data in the output.

• verbose (bool) – Whether to print progress information.

• compression (bool) – Compression algorithm to use for datasets. If True, uses ‘gzip’.

almanac.io.write_models_to_hdf5_group(models, hdf5_group, chunk_size=1000, compression=None)

Write a list of Pydantic models to an HDF5 group as separate datasets per field.

Parameters:

• models (List[BaseModel]) – List of Pydantic model instances (all same type)

• hdf5_group (Group) – HDF5 group to write datasets to

• chunk_size (int) – Chunk size for HDF5 datasets (for performance)

• compression (str) – Compression algorithm (‘gzip’, ‘lzf’, ‘szip’, None)

almanac.logger module

almanac.logger.get_formatter()

almanac.logger.get_logger()

almanac.utils module

almanac.utils.adjusted_fiber_index_to_fiber_id(adjusted_fiber_index)

Adjusted fiber index runs from 1 to 600 (inclusive) to account for APO and LCO.

Separate from whether it is APO or LCO, the fiber index runs in the opposite direction as what we call fiber_id.

almanac.utils.datetime_to_mjd(date)

Convert date string to Modified Julian Date.

Parameters:

date (str) – Date string in format “YYYY-MM-DD”

Return type:

int

Returns:

Modified Julian Date as integer

almanac.utils.get_current_mjd()

Get current Modified Julian Date as integer.

Return type:

int

Returns:

Current MJD as integer

almanac.utils.get_observatories(apo, lco)

Get observatory names based on boolean flags.

Parameters:

• apo (bool) – Whether to include APO observatory

• lco (bool) – Whether to include LCO observatory

Return type:

Tuple[str, ...]

Returns:

Tuple of observatory names (“apo”, “lco”, or both)

almanac.utils.group_contiguous(v)

almanac.utils.mjd_to_datetime(mjd)

Convert Modified Julian Date to datetime object.

Parameters:

mjd (float) – Modified Julian Date

Return type:

datetime

Returns:

Datetime object

almanac.utils.parse_mjds(mjd, mjd_start, mjd_end, date, date_start, date_end, earliest_mjd=0, return_nones=False)

Parse MJD and date parameters to determine observation date range.

Parameters:

• mjd (Optional[int]) – Single MJD value (can be negative for relative to current)

• mjd_start (Optional[int]) – Start MJD for range (can be negative for relative to current)

• mjd_end (Optional[int]) – End MJD for range (can be negative for relative to current)

• date (Optional[str]) – Single date string in “YYYY-MM-DD” format

• date_start (Optional[str]) – Start date string in “YYYY-MM-DD” format

• date_end (Optional[str]) – End date string in “YYYY-MM-DD” format

• earliest_mjd (int) – Earliest allowed MJD value (default: 0)

• return_nones (bool) – If True, return None for mjd range if no dates are specified (default: False)

Returns:

• MJD values (single int, range, or tuple)

• Start MJD (int)

• End MJD (int)

Return type:

Tuple containing

Raises:

• ValueError – If more than one time specification method is provided

• RuntimeError – If no valid time specification is found

almanac.utils.timestamp_to_mjd(v)

Convert Unix timestamp to Modified Julian Date (MJD).

Parameters:

v (float) – Unix timestamp in seconds

Return type:

float

Returns:

Modified Julian Date as float

Module contents