<commit_before>"""Utility functions for survey planning and scheduling.
"""
from __future__ import print_function, division

import datetime

import numpy as np

import pytz

import astropy.time
import astropy.coordinates
import astropy.units as u

import desisurvey.config


_telescope_location = None


def get_overhead_time(current_pointing, new_pointing, readout=True):
    """
    Compute the instrument overhead time between exposures.

    Use a model of the time required to slew and focus, in parallel with
    reading out the previous exposure.

    The calculation will be automatically broadcast over an array of new
    pointings and return an array of overhead times.

    Parameters
    ----------
    current_pointing : astropy.coordinates.SkyCoord or None
        Current pointing of the telescope.  Do not include any slew overhead
        when None.
    new_pointing : astropy.coordinates.SkyCoord
        New pointing(s) of the telescope.
    readout : bool
        Include parallel readout time of previous exposure in overhead.

    Returns
    -------
    astropy.units.Quantity
        Overhead time(s) for each new_pointing.
    """
    config = desisurvey.config.Configuration()
    if current_pointing is not None:
        # Calculate the amount that each axis needs to move in degrees.
        # The ra,dec attributes of a SkyCoord are always in the ranges
        # [0,360] and [-90,+90] degrees.
        delta_dec = np.fabs(
            (new_pointing.dec - current_pointing.dec).to(u.deg).value)
        delta_ra = np.fabs(
            (new_pointing.ra - current_pointing.ra).to(u.deg).value)
        # Handle wrap around in RA
        delta_ra = 180 - np.fabs(delta_ra - 180)
        # The slew time is determined by the axis motor with the most travel.
        max_travel = np.maximum(delta_ra, delta_dec) * u.deg
        moving = max_travel > 0
        overhead = max_travel / config.slew_rate()
        overhead[moving] += config.slew_overhead()
    else:
        overhead = np.zeros(new_pointing.shape) * u.s
    # Add the constant focus time.
    overhead += config.focus_time()
    if readout:
        # Overhead must be at least the readout time.
        overhead = np.maximum(overhead, config.readout_time())
    return overhead


def get_location():
    """Return the telescope's earth location.

    The location object is cached after the first call, so there is no need
    to cache this function's return value externally.

    Returns
    -------
    astropy.coordinates.EarthLocation
    """
    global _telescope_location
    if _telescope_location is None:
        config = desisurvey.config.Configuration()
        _telescope_location = astropy.coordinates.EarthLocation.from_geodetic(
            lat=config.location.latitude(),
            lon=config.location.longitude(),
            height=config.location.elevation())
    return _telescope_location


def get_observer(when, alt=None, az=None):
    """Return the AltAz frame for the telescope at the specified time(s).

    Refraction corrections are not applied (for now).

    The returned object is automatically broadcast over input arrays.

    Parameters
    ----------
    when : astropy.time.Time
        One or more times when the AltAz transformations should be calculated.
    alt : astropy.units.Quantity or None
        Local altitude angle(s)
    az : astropy.units.Quantity or None
        Local azimuth angle(s)

    Returns
    -------
    astropy.coordinates.AltAz
        AltAz frame object suitable for transforming to/from local horizon
        (alt, az) coordinates.
    """
    if alt is not None and az is not None:
        kwargs = dict(alt=alt, az=az)
    elif alt is not None or az is not None:
        raise ValueError('Must specify both alt and az.')
    else:
        kwargs = {}
    return astropy.coordinates.AltAz(
        location=get_location(), obstime=when, pressure=0, **kwargs)


def get_airmass(when, ra, dec):
    """Return the airmass of (ra,dec) at the specified observing time.

    Uses the Rosenberg 1966 formula: 1/X = cosZ + 0.025 * exp(-11*cosZ).

    The value of cosZ is clipped at zero, so observations below the horizon
    return the horizon value.

    Parameters
    ----------
    when : astropy.time.Time
        Observation time, which specifies the local zenith.
    ra : astropy.units.Quantity
        Target RA angle(s)
    dec : astropy.units.Quantity
        Target DEC angle(s)

    Returns
    -------
    array or float
        Value of the airmass for each input (ra,dec).
    """
    target = astropy.coordinates.SkyCoord(ra=ra, dec=dec)
    zenith = get_observer(when, alt=90 * u.deg, az=0 * u.deg
                          ).transform_to(astropy.coordinates.ICRS)
    cosZ = np.clip(np.cos(target.separation(zenith)), 0., 1.)
    return 1. / (cosZ + 0.025 * np.exp(-11 * cosZ))


def is_monsoon(night):
    """Test if this night's observing falls in the monsoon shutdown.

    Uses the monsoon date range defined in the
    :class:`desisurvey.config.Configuration`.  Based on (month, day) comparisons
    rather than day-of-year comparisons, so the monsoon always starts on the
    same calendar date, even in leap years.

    Parameters
    ----------
    night : date
        Converted to a date using :func:`desisurvey.utils.get_date`.

    Returns
    -------
    bool
        True if this night's observing falls during the monsoon shutdown.
    """
    date = get_date(night)

    # Fetch our configuration.
    config = desisurvey.config.Configuration()
    start = config.monsoon_start()
    stop = config.monsoon_stop()
<commit_msg>Do not return an astropy Quantity from get_airmass()<commit_after>"""Utility functions for survey planning and scheduling.
"""
from __future__ import print_function, division

import datetime

import numpy as np

import pytz

import astropy.time
import astropy.coordinates
import astropy.units as u

import desisurvey.config


_telescope_location = None


def get_overhead_time(current_pointing, new_pointing, readout=True):
    """
    Compute the instrument overhead time between exposures.

    Use a model of the time required to slew and focus, in parallel with
    reading out the previous exposure.

    The calculation will be automatically broadcast over an array of new
    pointings and return an array of overhead times.

    Parameters
    ----------
    current_pointing : astropy.coordinates.SkyCoord or None
        Current pointing of the telescope.  Do not include any slew overhead
        when None.
    new_pointing : astropy.coordinates.SkyCoord
        New pointing(s) of the telescope.
    readout : bool
        Include parallel readout time of previous exposure in overhead.

    Returns
    -------
    astropy.units.Quantity
        Overhead time(s) for each new_pointing.
    """
    config = desisurvey.config.Configuration()
    if current_pointing is not None:
        # Calculate the amount that each axis needs to move in degrees.
        # The ra,dec attributes of a SkyCoord are always in the ranges
        # [0,360] and [-90,+90] degrees.
        delta_dec = np.fabs(
            (new_pointing.dec - current_pointing.dec).to(u.deg).value)
        delta_ra = np.fabs(
            (new_pointing.ra - current_pointing.ra).to(u.deg).value)
        # Handle wrap around in RA
        delta_ra = 180 - np.fabs(delta_ra - 180)
        # The slew time is determined by the axis motor with the most travel.
        max_travel = np.maximum(delta_ra, delta_dec) * u.deg
        moving = max_travel > 0
        overhead = max_travel / config.slew_rate()
        overhead[moving] += config.slew_overhead()
    else:
        overhead = np.zeros(new_pointing.shape) * u.s
    # Add the constant focus time.
    overhead += config.focus_time()
    if readout:
        # Overhead must be at least the readout time.
        overhead = np.maximum(overhead, config.readout_time())
    return overhead


def get_location():
    """Return the telescope's earth location.

    The location object is cached after the first call, so there is no need
    to cache this function's return value externally.

    Returns
    -------
    astropy.coordinates.EarthLocation
    """
    global _telescope_location
    if _telescope_location is None:
        config = desisurvey.config.Configuration()
        _telescope_location = astropy.coordinates.EarthLocation.from_geodetic(
            lat=config.location.latitude(),
            lon=config.location.longitude(),
            height=config.location.elevation())
    return _telescope_location


def get_observer(when, alt=None, az=None):
    """Return the AltAz frame for the telescope at the specified time(s).

    Refraction corrections are not applied (for now).

    The returned object is automatically broadcast over input arrays.

    Parameters
    ----------
    when : astropy.time.Time
        One or more times when the AltAz transformations should be calculated.
    alt : astropy.units.Quantity or None
        Local altitude angle(s)
    az : astropy.units.Quantity or None
        Local azimuth angle(s)

    Returns
    -------
    astropy.coordinates.AltAz
        AltAz frame object suitable for transforming to/from local horizon
        (alt, az) coordinates.
    """
    if alt is not None and az is not None:
        kwargs = dict(alt=alt, az=az)
    elif alt is not None or az is not None:
        raise ValueError('Must specify both alt and az.')
    else:
        kwargs = {}
    return astropy.coordinates.AltAz(
        location=get_location(), obstime=when, pressure=0, **kwargs)


def get_airmass(when, ra, dec):
    """Return the airmass of (ra,dec) at the specified observing time.

    Uses the Rosenberg 1966 formula: 1/X = cosZ + 0.025 * exp(-11*cosZ).

    The value of cosZ is clipped at zero, so observations below the horizon
    return the horizon value.

    Parameters
    ----------
    when : astropy.time.Time
        Observation time, which specifies the local zenith.
    ra : astropy.units.Quantity
        Target RA angle(s)
    dec : astropy.units.Quantity
        Target DEC angle(s)

    Returns
    -------
    array or float
        Value of the airmass for each input (ra,dec).
    """
    target = astropy.coordinates.SkyCoord(ra=ra, dec=dec)
    zenith = get_observer(when, alt=90 * u.deg, az=0 * u.deg
                          ).transform_to(astropy.coordinates.ICRS)
    cosZ = np.clip(np.cos(target.separation(zenith).to(u.rad).value), 0., 1.)
    return 1. / (cosZ + 0.025 * np.exp(-11 * cosZ))


def is_monsoon(night):
    """Test if this night's observing falls in the monsoon shutdown.

    Uses the monsoon date range defined in the
    :class:`desisurvey.config.Configuration`.  Based on (month, day) comparisons
    rather than day-of-year comparisons, so the monsoon always starts on the
    same calendar date, even in leap years.

    Parameters
    ----------
    night : date
        Converted to a date using :func:`desisurvey.utils.get_date`.

    Returns
    -------
    bool
        True if this night's observing falls during the monsoon shutdown.
    """
    date = get_date(night)

    # Fetch our configuration.
    config = desisurvey.config.Configuration()
    start = config.monsoon_start()
    stop = config.monsoon_stop()

{'fraction_non_alphanumeric': 0.08870181657066903, 'fraction_numerical': 0.006911829862649535, 'mean_word_length': 2.5591296121097447, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '318059', 'n_tokens_mistral': 3366, 'n_tokens_neox': 3145, 'n_words': 1105}