# -*- coding: utf-8 -*-
"""
Based on the skycalc_cli package, but heavily modified.
The original code was taken from ``skycalc_cli`` version 1.1.
Credit for ``skycalc_cli`` goes to ESO
"""
import logging
import warnings
import hashlib
import json
from os import environ
from datetime import datetime
from pathlib import Path
from collections.abc import Mapping
from importlib import import_module
import httpx
from astropy.io import fits
CACHE_DIR_FALLBACK = ".astar/skycalc_ipy"
[docs]def get_cache_dir() -> Path:
"""Establish the cache directory.
There are three possible locations for the cache directory:
1. As set in `os.environ["SKYCALC_IPY_CACHE_DIR"]`
2. As set in the `scopesim_data` package.
3. The `data` directory in this package.
"""
try:
dir_cache = Path(environ["SKYCALC_IPY_CACHE_DIR"])
except KeyError:
try:
sim_data = import_module("scopesim_data")
dir_cache = Path(getattr(sim_data, "dir_cache_skycalc"))
except (ImportError, AttributeError):
dir_cache = Path.home() / CACHE_DIR_FALLBACK
if not dir_cache.is_dir():
dir_cache.mkdir(parents=True)
return dir_cache
[docs]class ESOQueryBase:
"""Base class for queries to ESO skycalc server."""
REQUEST_TIMEOUT = 2 # Time limit (in seconds) for server response
BASE_URL = "https://etimecalret-002.eso.org/observing/etc"
def __init__(self, url, params):
self.url = url
self.params = params
def _send_request(self) -> httpx.Response:
try:
with httpx.Client(base_url=self.BASE_URL,
timeout=self.REQUEST_TIMEOUT) as client:
response = client.post(self.url, json=self.params)
response.raise_for_status()
except httpx.RequestError as err:
logging.exception("An error occurred while requesting %s.",
err.request.url)
raise err
except httpx.HTTPStatusError as err:
logging.error("Error response %s while requesting %s.",
err.response.status_code, err.request.url)
raise err
return response
[docs] def get_cache_filenames(self, prefix: str, suffix: str) -> str:
"""Produce filename from hass of parameters.
Using three underscores between the key-value pairs and two underscores
between the key and the value.
"""
akey = "___".join(f"{k}__{v}" for k, v in self.params.items())
ahash = hashlib.sha256(akey.encode("utf-8")).hexdigest()
fname = f"{prefix}_{ahash}.{suffix}"
return fname
[docs]class AlmanacQuery(ESOQueryBase):
"""
A class for querying the SkyCalc Almanac.
Parameters
----------
indic : dict, SkyCalcParams
A dictionary / :class:``SkyCalcParams`` object containing the following
keywords: ``ra``, ``dec``, ``date`` or ``mjd``
- ``ra`` : [deg] a float in the range [0, 360]
- ``dec`` : [deg] a float in the range [-90, 90]
And either ``data`` or ``mjd``:
- ``date`` : a datetime string in the format 'yyyy-mm-ddThh:mm:ss'
- ``mjd`` : a float with the modified julian date. Note that ``mjd=0``
corresponds to the date "1858-11-17"
"""
def __init__(self, indic):
# FIXME: This basically checks isinstance(indic, ui.SkyCalc), but we
# can't import that because it would create a circual import.
# TODO: Find a better way to do this!!
if hasattr(indic, "defaults"):
indic = indic.values
super().__init__("/api//skycalc_almanac", {})
# Left: users keyword (skycalc_cli),
# Right: skycalc Almanac output keywords
self.alm_parameters = {
"airmass": "target_airmass",
"msolflux": "sun_aveflux",
"moon_sun_sep": "moon_sun_sep",
"moon_target_sep": "moon_target_sep",
"moon_alt": "moon_alt",
"moon_earth_dist": "moon_earth_dist",
"ecl_lon": "ecl_lon",
"ecl_lat": "ecl_lat",
"observatory": "observatory",
}
# The Almanac needs:
# coord_ra : float [deg]
# coord_dec : float [deg]
# input_type : ut_time | local_civil_time | mjd
# mjd : float
# coord_year : int
# coord_month : int
# coord_day : int
# coord_ut_hour : int
# coord_ut_min : int
# coord_ut_sec : float
self._set_date(indic)
self._set_radec(indic, "ra")
self._set_radec(indic, "dec")
if "observatory" in indic:
self.params["observatory"] = indic["observatory"]
def _set_date(self, indic):
if "date" in indic and indic["date"] is not None:
if isinstance(indic["date"], str):
isotime = datetime.strptime(indic["date"], "%Y-%m-%dT%H:%M:%S")
else:
isotime = indic["date"]
updated = {
"input_type": "ut_time",
"coord_year": isotime.year,
"coord_month": isotime.month,
"coord_day": isotime.day,
"coord_ut_hour": isotime.hour,
"coord_ut_min": isotime.minute,
"coord_ut_sec": isotime.second,
}
elif "mjd" in indic and indic["mjd"] is not None:
updated = {
"input_type": "mjd",
"mjd": float(indic["mjd"]),
}
else:
raise ValueError("No valid date or mjd given for the Almanac")
self.params.update(updated)
def _set_radec(self, indic, which):
try:
self.params[f"coord_{which}"] = float(indic[which])
except KeyError as err:
logging.exception("%s coordinate not given for the Almanac.",
which)
raise err
except ValueError as err:
logging.exception("Wrong %s format for the Almanac.", which)
raise err
def _get_jsondata(self, file_path: Path):
if file_path.exists():
return json.load(file_path.open(encoding="utf-8"))
response = self._send_request()
if not response.text:
raise ValueError("Empty response.")
jsondata = response.json()["output"]
# Use a fixed date so the stored files are always identical for
# identical requests.
jsondata["execution_datetime"] = "2017-01-07T00:00:00 UTC"
try:
json.dump(jsondata, file_path.open("w", encoding="utf-8"))
# json.dump(self.params, open(fn_params, 'w'))
except (PermissionError, FileNotFoundError) as err:
# Apparently it is not possible to save here.
raise err
return jsondata
def __call__(self):
"""
Query the ESO Skycalc server with the parameters in self.params.
Returns
-------
almdata : dict
Dictionary with the relevant parameters for the date given
"""
cache_dir = get_cache_dir()
cache_name = self.get_cache_filenames("almanacquery", "json")
cache_path = cache_dir / cache_name
jsondata = self._get_jsondata(cache_path)
# Find the relevant (key, value)
almdata = {}
for key, value in self.alm_parameters.items():
prefix = value.split("_", maxsplit=1)[0]
if prefix in {"sun", "moon", "target"}:
subsection = prefix
elif prefix == "ecl":
subsection = "target"
else:
subsection = "observation"
try:
almdata[key] = jsondata[subsection][value]
except (KeyError, ValueError):
logging.warning("Key '%s/%s' not found in Almanac response.",
subsection, value)
return almdata
[docs] def query(self):
"""Deprecated feature. Class is now callable, use that instead."""
warnings.warn("The .query() method is deprecated and will be removed "
"in a future release. Please simply call the instance.",
DeprecationWarning, stacklevel=2)
return self()
[docs]class SkyModel(ESOQueryBase):
"""
Class for querying the Advanced SkyModel at ESO.
Contains all the parameters needed for querying the ESO SkyCalc server.
The parameters are contained in :attr:`.params` and the returned FITS file
is in :attr:`.data` in binary form. This must be saved to disk before it
can be read with the :meth:`.write` method.
Parameter and their default values and comments can be found at:
https://www.eso.org/observing/etc/bin/gen/form?INS.MODE=swspectr+INS.NAME=SKYCALC
"""
def __init__(self):
self.data = None
self.data_url = "/tmp/"
self.deleter_script_url = "/api/rmtmp"
self._last_status = ""
self.tmpdir = ""
params = {
# Airmass. Alt and airmass are coupled through the plane parallel
# approximation airmass=sec(z), z being the zenith distance
# z=90-Alt
"airmass": 1.0, # float range [1.0,3.0]
# Season and Period of Night
"pwv_mode": "pwv", # string grid ['pwv','season']
# integer grid [0,1,2,3,4,5,6] (0=all year, 1=dec/jan,2=feb/mar...)
"season": 0,
# third of night integer grid [0,1,2,3] (0=all year, 1,2,3 = third
# of night)
"time": 0,
# Precipitable Water Vapor PWV
# mm float grid [-1.0,0.5,1.0,1.5,2.5,3.5,5.0,7.5,10.0,20.0]
"pwv": 3.5,
# Monthly Averaged Solar Flux
"msolflux": 130.0, # s.f.u float > 0
# Scattered Moon Light
# Moon coordinate constraints: |z - zmoon| <= rho <= |z + zmoon|
# where rho = moon/target separation, z = 90-target altitude and
# zmoon = 90-moon altitude.
# string grid ['Y','N'] flag for inclusion of scattered moonlight.
"incl_moon": "Y",
# degrees float range [0.0,360.0] Separation of Sun and Moon as
# seen from Earth ("moon phase")
"moon_sun_sep": 90.0,
# degrees float range [0.0,180.0] Moon-Target Separation ( rho )
"moon_target_sep": 45.0,
# degrees float range [-90.0,90.0] Moon Altitude over Horizon
"moon_alt": 45.0,
# float range [0.91,1.08] Moon-Earth Distance (mean=1)
"moon_earth_dist": 1.0,
# Starlight
# string grid ['Y','N'] flag for inclusion of scattered starlight
"incl_starlight": "Y",
# Zodiacal light
# string grid ['Y','N'] flag for inclusion of zodiacal light
"incl_zodiacal": "Y",
# degrees float range [-180.0,180.0] Heliocentric ecliptic
# longitude
"ecl_lon": 135.0,
# degrees float range [-90.0,90.0] Ecliptic latitude
"ecl_lat": 90.0,
# Molecular Emission of Lower Atmosphere
# string grid ['Y','N'] flag for inclusion of lower atmosphere
"incl_loweratm": "Y",
# Emission Lines of Upper Atmosphere
# string grid ['Y','N'] flag for inclusion of upper stmosphere
"incl_upperatm": "Y",
# Airglow Continuum (Residual Continuum)
# string grid ['Y','N'] flag for inclusion of airglow
"incl_airglow": "Y",
# Instrumental Thermal Emission This radiance component represents
# an instrumental effect. The emission is provided relative to the
# other model components. To obtain the correct absolute flux, an
# instrumental response curve must be applied to the resulting
# model spectrum See section 6.2.4 in the documentation
# http://localhost/observing/etc/doc/skycalc/
# The_Cerro_Paranal_Advanced_Sky_Model.pdf
# string grid ['Y','N'] flag for inclusion of instrumental thermal
# radiation
"incl_therm": "N",
"therm_t1": 0.0, # K float > 0
"therm_e1": 0.0, # float range [0,1]
"therm_t2": 0.0, # K float > 0
"therm_e2": 0.0, # float range [0,1]
"therm_t3": 0.0, # float > 0
"therm_e3": 0.0, # K float range [0,1]
# Wavelength Grid
"vacair": "vac", # vac or air
"wmin": 300.0, # nm float range [300.0,30000.0] < wmax
"wmax": 2000.0, # nm float range [300.0,30000.0] > wmin
# string grid ['fixed_spectral_resolution','fixed_wavelength_step']
"wgrid_mode": "fixed_wavelength_step",
# nm/step float range [0,30000.0] wavelength sampling step dlam
# (not the res.element)
"wdelta": 0.1,
# float range [0,1.0e6] RESOLUTION is misleading, it is rather
# lam/dlam where dlam is wavelength step (not the res.element)
"wres": 20000,
# Convolve by Line Spread Function
"lsf_type": "none", # string grid ['none','Gaussian','Boxcar']
"lsf_gauss_fwhm": 5.0, # wavelength bins float > 0
"lsf_boxcar_fwhm": 5.0, # wavelength bins float > 0
"observatory": "paranal", # paranal
}
super().__init__("/api/skycalc", params)
[docs] def fix_observatory(self):
"""
Convert the human readable observatory name into its ESO ID number.
The following observatory names are accepted: ``lasilla``, ``paranal``,
``armazones`` or ``3060m``, ``highanddry`` or ``5000m``
"""
# FIXME: DO WE ALWAYS WANT TO RAISE WHEN IT'S NOT ONE OF THOSE???
if self.params["observatory"] not in {
"paranal",
"lasilla",
"armazones",
"3060m",
"5000m",
}:
return # nothing to do
if self.params["observatory"] == "lasilla":
self.params["observatory"] = "2400"
elif self.params["observatory"] == "paranal":
self.params["observatory"] = "2640"
elif (
self.params["observatory"] == "3060m"
or self.params["observatory"] == "armazones"
):
self.params["observatory"] = "3060"
elif (
self.params["observatory"] == "5000m"
or self.params["observatory"] == "highanddry"
):
self.params["observatory"] = "5000"
else:
raise ValueError(
"Wrong Observatory name, please refer to the documentation."
)
return # for consistency
def __getitem__(self, item):
return self.params[item]
def __setitem__(self, key, value):
self.params[key] = value
if key == "observatory":
self.fix_observatory()
def _retrieve_data(self, url):
try:
self.data = fits.open(url)
# Use a fixed date so the stored files are always identical for
# identical requests.
self.data[0].header["DATE"] = "2017-01-07T00:00:00"
except Exception as err:
logging.exception(
"Exception raised trying to get FITS data from %s", url)
raise err
[docs] def write(self, local_filename, **kwargs):
"""Write data to file."""
try:
self.data.writeto(local_filename, **kwargs)
except (IOError, FileNotFoundError):
logging.exception("Exception raised trying to write fits file.")
[docs] def getdata(self):
"""Deprecated feature, just use the .data attribute."""
warnings.warn("The .getdata method is deprecated and will be removed "
"in a future release. Use the identical .data attribute "
"instead.", DeprecationWarning, stacklevel=2)
return self.data
def _delete_server_tmpdir(self, tmpdir):
try:
with httpx.Client(base_url=self.BASE_URL,
timeout=self.REQUEST_TIMEOUT) as client:
response = client.get(self.deleter_script_url,
params={"d": tmpdir})
deleter_response = response.text.strip().lower()
if deleter_response != "ok":
logging.error("Could not delete server tmpdir %s: %s",
tmpdir, deleter_response)
except httpx.HTTPError:
logging.exception("Exception raised trying to delete tmp dir %s",
tmpdir)
def _update_params(self, updated: Mapping) -> None:
par_keys = self.params.keys()
new_keys = updated.keys()
self.params.update((key, updated[key]) for key in par_keys & new_keys)
logging.debug("Ignoring invalid keywords: %s", new_keys - par_keys)
def __call__(self, **kwargs):
"""Send server request."""
if kwargs:
logging.info("Setting new parameters: %s", kwargs)
self._update_params(kwargs)
self.fix_observatory()
cache_dir = get_cache_dir()
cache_name = self.get_cache_filenames("skymodel", "fits")
cache_path = cache_dir / cache_name
if cache_path.exists():
self.data = fits.open(cache_path)
return
response = self._send_request()
try:
res = response.json()
status = res["status"]
tmpdir = res["tmpdir"]
except (KeyError, ValueError) as err:
logging.exception(
"Exception raised trying to decode server response.")
raise err
self._last_status = status
if status == "success":
try:
# retrive and save FITS data (in memory)
self._retrieve_data(
self.BASE_URL + self.data_url + tmpdir + "/skytable.fits")
except httpx.HTTPError as err:
logging.exception("Could not retrieve FITS data from server.")
raise err
try:
self.data.writeto(cache_path)
# with fn_params.open("w", encoding="utf-8") as file:
# json.dump(self.params, file)
except (PermissionError, FileNotFoundError):
# Apparently it is not possible to save here.
pass
self._delete_server_tmpdir(tmpdir)
else: # print why validation failed
logging.error("Parameter validation error: %s", res["error"])
[docs] def call(self):
"""Deprecated feature, just call the instance."""
warnings.warn("The .call() method is deprecated and will be removed "
"in a future release. Please simply call the instance.",
DeprecationWarning, stacklevel=2)
self()
[docs] def callwith(self, newparams):
"""Deprecated feature, just call the instance."""
warnings.warn("The .callwith(args) method is deprecated and will be "
"removed in a future release. Please simply call the "
"instance with optional kwargs instead.",
DeprecationWarning, stacklevel=2)
self(**newparams)
[docs] def printparams(self, keys=None):
"""
List the values of all, or a subset, of parameters.
Parameters
----------
keys : sequence of str, optional
List of keys to print. If None, all keys will be printed.
"""
for key in keys or self.params.keys():
print(f" {key}: {self.params[key]}")