# -*- mode: python; coding: utf-8 -*-
# Copyright 2020 the AAS WorldWide Telescope project
# Licensed under the MIT License.
"""
Support for loading images from an AstroPix feed.
TODO: update metadata tomfoolery to match the new things that I've learned. Cf.
the ``wwtdatatool wtml report`` utility and the djangoplicity implementation.
NOTE: AstroPix seems to have its image parity backwards. Standard JPEGs are
reported with ``wcs_scale[0] < 0``, which is *positive* AKA bottoms-up AKA FITS
parity. If we pass their parameters more-or-less straight through to WWT, we get
the right appearance onscreen.
"""
__all__ = """
AstroPixImageSource
AstroPixCandidateInput
""".split()
import codecs
from datetime import datetime
import json
import numpy as np
import os.path
import requests
import shutil
from urllib.parse import quote as urlquote
from ..image import ImageLoader
from . import CandidateInput, ImageSource, NotActionableError
EXTENSION_REMAPPING = {
"jpeg": "jpg",
}
[docs]
class AstroPixImageSource(ImageSource):
"""
An ImageSource that obtains its inputs from a query to the AstroPix service.
"""
_json_query_url = None
[docs]
@classmethod
def get_config_key(cls):
return "astropix"
[docs]
@classmethod
def deserialize(cls, data):
inst = cls()
inst._json_query_url = data["json_query_url"]
return inst
[docs]
def query_candidates(self):
with requests.get(self._json_query_url, stream=True) as resp:
feed_data = json.load(resp.raw)
for item in feed_data:
yield AstroPixCandidateInput(item)
[docs]
def fetch_candidate(self, unique_id, cand_data_stream, cachedir):
with codecs.getreader("utf8")(cand_data_stream) as text_stream:
info = json.load(text_stream)
lower_id = info["image_id"].lower()
global_id = info["publisher_id"] + "_" + lower_id
if info["resource_url"] and len(info["resource_url"]):
source_url = info["resource_url"]
else:
# Original image not findable. Get the best version available from
# AstroPix.
#
# GROSS: saving this code since I won't be testing this super
# thoroughly ... but it looks like now the original is being made
# available consistently?
##size = int(info['image_max_boundry'])
##if size >= 24000:
## best_astropix_size = 24000
##elif size >= 12000:
## best_astropix_size = 12000
##elif size >= 6000:
## best_astropix_size = 6000
##elif size >= 3000:
## best_astropix_size = 3000
##elif size >= 1600:
## best_astropix_size = 1600
##elif size > 1024: # transition point to sizes that are always generated
## best_astropix_size = 1280
##elif size > 500:
## best_astropix_size = 1024
##elif size > 320:
## best_astropix_size = 500
##else:
## best_astropix_size = 320
source_url = (
"http://astropix.ipac.caltech.edu/archive/%s/%s/%s_original.jpg"
% (
urlquote(info["publisher_id"]),
urlquote(lower_id),
urlquote(global_id),
)
)
# Now ready to download the image.
ext = source_url.rsplit(".", 1)[-1].lower()
ext = EXTENSION_REMAPPING.get(ext, ext)
with requests.get(source_url, stream=True) as resp:
if not resp.ok:
raise Exception(f"error downloading {source_url}: {resp.status_code}")
with open(os.path.join(cachedir, "image." + ext), "wb") as f:
shutil.copyfileobj(resp.raw, f)
[docs]
def process(self, unique_id, cand_data_stream, cachedir, builder):
# Set up the metadata.
with codecs.getreader("utf8")(cand_data_stream) as text_stream:
info = json.load(text_stream)
if info["resource_url"] and len(info["resource_url"]):
ext = info["resource_url"].rsplit(".", 1)[-1].lower()
ext = EXTENSION_REMAPPING.get(ext, ext)
else:
ext = "jpg"
img_path = os.path.join(cachedir, "image." + ext)
md = AstroPixMetadata(info)
# Load up the image.
img = ImageLoader().load_path(img_path)
# Do the processing.
builder.tile_base_as_study(img)
builder.make_thumbnail_from_other(img)
builder.imgset.set_position_from_wcs(
md.as_wcs_headers(img.width, img.height),
img.width,
img.height,
place=builder.place,
)
builder.set_name(info["title"])
builder.imgset.credits_url = md.get_credit_url()
builder.cascade()
ASTROPIX_FLOAT_ARRAY_KEYS = [
"wcs_reference_dimension", # NB: should be ints, but sometimes expressed with decimal points
"wcs_reference_pixel",
"wcs_reference_value",
"wcs_scale",
]
ASTROPIX_FLOAT_SCALAR_KEYS = [
"wcs_rotation",
]
class AstroPixMetadata(object):
"""
Metadata derived from AstroPix query results.
"""
image_id = None
publisher_id = None
resource_url = None
wcs_coordinate_frame = None # ex: 'ICRS'
wcs_equinox = None # ex: 'J2000'
wcs_projection = None # ex: 'TAN'
wcs_reference_dimension = None # ex: [7416.0, 4320.0]
wcs_reference_value = None # ex: [187, 12.3]
wcs_reference_pixel = (
None # ex: [1000.4, 1000.7]; from examples, this seems to be 1-based
)
wcs_rotation = None # ex: -0.07 (deg, presumably)
wcs_scale = None # ex: [-6e-7, 6e-7]
def __init__(self, json_dict):
# Some massaging for consistency:
for k in ASTROPIX_FLOAT_ARRAY_KEYS:
if k in json_dict:
json_dict[k] = list(map(float, json_dict[k]))
for k in ASTROPIX_FLOAT_SCALAR_KEYS:
if k in json_dict:
json_dict[k] = float(json_dict[k])
for k, v in json_dict.items():
setattr(self, k, v)
def as_wcs_headers(self, width, height):
"""
The metadata here are essentially AVM headers. As described in
`Builder.apply_avm_info()`, the data that we've seen in the wild are a
bit wonky with regards to parity: the metadata essentially correspond to
FITS-like parity, and we need to flip them to JPEG-like parity. See also
very similar code in `djangoplicity.py`.
"""
headers = {}
# headers['RADECSYS'] = self.wcs_coordinate_frame # causes Astropy warnings
headers["CTYPE1"] = "RA---" + self.wcs_projection
headers["CTYPE2"] = "DEC--" + self.wcs_projection
headers["CRVAL1"] = self.wcs_reference_value[0]
headers["CRVAL2"] = self.wcs_reference_value[1]
# See Calabretta & Greisen (2002; DOI:10.1051/0004-6361:20021327), eqn 186
crot = np.cos(self.wcs_rotation * np.pi / 180)
srot = np.sin(self.wcs_rotation * np.pi / 180)
lam = self.wcs_scale[1] / self.wcs_scale[0]
pc1_1 = crot
pc1_2 = -lam * srot
pc2_1 = srot / lam
pc2_2 = crot
# If we couldn't get the original image, the pixel density used for
# the WCS parameters may not match the image resolution that we have
# available. In such cases, we need to remap the pixel-related
# headers. From the available examples, `wcs_reference_pixel` seems to
# be 1-based in the same way that `CRPIXn` are. Since in FITS, integer
# pixel values correspond to the center of each pixel box, a CRPIXn of
# [0.5, 0.5] (the lower-left corner) should not vary with the image
# resolution. A CRPIXn of [W + 0.5, H + 0.5] (the upper-right corner)
# should map to [W' + 0.5, H' + 0.5] (where the primed quantities are
# the new width and height).
factor0 = width / self.wcs_reference_dimension[0]
factor1 = height / self.wcs_reference_dimension[1]
headers["CRPIX1"] = (self.wcs_reference_pixel[0] - 0.5) * factor0 + 0.5
headers["CRPIX2"] = (self.wcs_reference_pixel[1] - 0.5) * factor1 + 0.5
# Now finalize and apply the parity flip
cdelt1 = self.wcs_scale[0] / factor0
cdelt2 = self.wcs_scale[1] / factor1
headers["CD1_1"] = cdelt1 * pc1_1
headers["CD1_2"] = -cdelt1 * pc1_2
headers["CD2_1"] = cdelt2 * pc2_1
headers["CD2_2"] = -cdelt2 * pc2_2
headers["CRPIX2"] = height + 1 - headers["CRPIX2"]
return headers
def get_credit_url(self):
if self.reference_url:
return self.reference_url
return "http://astropix.ipac.caltech.edu/image/%s/%s" % (
urlquote(self.publisher_id),
urlquote(self.image_id),
)