"""
The coverage RD element.
Most code to obtain coverage information from actual resources is in
gavo.user.info.
"""
#c Copyright 2008-2023, the GAVO project <gavo@ari.uni-heidelberg.de>
#c
#c This program is free software, covered by the GNU GPL. See the
#c COPYING file in the source distribution.
from gavo import base
from gavo import stc
from gavo import utils
from gavo.rscdef import rdinj
from gavo.utils import pgsphere
[docs]class FloatInterval(base.AtomicAttribute):
"""An attribute keeping a floating-point attribute.
The literal of these is as in VOTable: a whitespace-separated pair
for decimal floating-point literals.
"""
typeDesc_ = "A float interval (i.e., a space-separated pair of floats)"
[docs] def parse(self, value):
try:
lower, upper = [float(f) for f in value.split()]
if lower>upper:
raise base.LiteralParseError(self.name_, value,
hint="In intervals, the lower value must be smaller or equal"
" to the upper one.")
return lower, upper
except (ValueError, TypeError):
raise base.LiteralParseError(self.name_, value,
hint="Intervals are represented as two floating-point literals"
" separated by whitespace.")
[docs] def unparse(self, value):
return "%.10g %.10g"%value
[docs]class FloatOrDateInterval(FloatInterval):
"""An attribute keeping a temporal interval as MJD.
On literal input, we also allow ISO-like timestamps. You cannot
mix MJD and ISO, though.
"""
typeDesc_ = "A float or ISO timestamp interval (values separated by space)"
[docs] def parse(self, value):
try:
return FloatInterval.parse(self, value)
except base.LiteralParseError:
try:
lower, upper = [stc.dateTimeToMJD(utils.parseISODT(f))
for f in value.split()]
if lower>upper:
raise base.LiteralParseError(self.name_, value,
hint="In intervals, the lower value must be smaller or equal"
" to the upper one.")
return lower, upper
except (ValueError, TypeError):
raise base.LiteralParseError(self.name_, value,
hint="Time intervals are represented as two floating-point or"
" ISO timestamp literals separated by whitespace.")
[docs]class Ranges(base.ListOfAtomsAttribute):
"""An attribute definition for a list of intervals.
"""
typeDesc_ = "A sequence of intervals (a space-separated pair of floats"
def __init__(self, name, intervalAttr, default=[], **kwargs):
base.ListOfAtomsAttribute.__init__(
self, name, default, intervalAttr, **kwargs)
[docs] def feedFromArray(self, instance, arr):
"""defines the intervals from a flat array by constructing
each interval from consecuive pairs.
"""
self.feedObject(instance,
list(zip(arr[0::2], arr[1::2])))
[docs]class Updater(base.Structure):
"""Information on where and how to update a piece of coverage information.
"""
name_ = "updater"
# this stuff is interpreted by user.limits.iterCoverageItems
_sourceTable = base.ReferenceAttribute("sourceTable",
default=base.NotGiven,
description="A table from which to compute coverage by default.",
copyable=True)
_spaceTable = base.ReferenceAttribute("spaceTable",
default=base.NotGiven,
description="A table from which to compute spatial coverage (overrides"
" sourceTable).",
copyable=True)
_timeTable = base.ReferenceAttribute("timeTable",
default=base.NotGiven,
description="A table from which to compute temporal coverage (overrides"
" sourceTable)",
copyable=True)
_spectralTable = base.ReferenceAttribute("spectralTable",
default=base.NotGiven,
description="A table from which to compute spectral coverage (overrides"
" sourceTable)",
copyable=True)
_mocOrder = base.IntAttribute("mocOrder",
default=6,
description="Maximal HEALpix order to use in coverage MOCs (6 is about"
" a degree resolution, each additional point doubles resolution).",
copyable=True)
[docs]class Coverage(base.Structure):
"""The coverage of a resource.
For now, this is attached to the complete resource rather than the table,
since this is where it sits in VOResource. DaCHS *could* be a bit
more flexible, allowing different coverages per publish element. It
is not right now, though.
Note: Technically, this will introduce or amend the coverage meta
element. The information given here will be masked if you define
a coverage meta on the service or table level. Just do not do that.
"""
name_ = "coverage"
_updater = base.StructAttribute("updater",
childFactory=Updater,
description="Rules for automatic computation or updating of"
" coverage information.",
default=base.NotGiven)
_spectral = Ranges("spectral",
FloatInterval(
"spectral", description="A wavelength interval"),
description="Interval(s) of spectral coverage, in Joules of"
" BARYCENTER vacuum messenger particle energy.")
_temporal = Ranges("temporal",
FloatOrDateInterval(
"temporal", description="A time interval"),
description="Interval(s) of temporal coverage, in MJD"
" (for TT BARYCENTER).")
_spatial = base.UnicodeAttribute("spatial",
default=base.NotGiven,
description="A MOC in ASCII representation giving the ICRS coverage"
" of the resource")
_fallbackTo = base.UnicodeAttribute("fallbackTo",
default=base.NotGiven,
description="Take coverage information from this RD in case"
" none is given locally. It is not an error if no such"
" coverage information exists either. Use this when a service"
" re-exposes data from another RD, as in sitewide siap2 and obscore,"
" so you only have to compute the coverage once.",
copyable=True)
[docs] def completeElement(self, ctx):
super().completeElement(ctx)
fallbackRecords = {}
if self.fallbackTo:
with base.getTableConn() as conn:
try:
fallbackRecords = rdinj._getRecordsFor(
self.fallbackTo, "dc.rdmeta", conn)[0]
except IndexError:
# no coverage from fallbackTo, either; keep the empty dict.
pass
if self.spatial is base.NotGiven:
inj = ctx.getInjected("spatial_coverage",
fallbackRecords.get("spatial"))
if inj:
self.feed("spatial", inj)
if not self.spectral:
inj = ctx.getInjected("spectral_coverage",
fallbackRecords.get("spectral"))
if inj:
self._spectral.feedFromArray(self, inj)
if not self.temporal:
inj = ctx.getInjected("temporal_coverage",
fallbackRecords.get("temporal"))
if inj:
self._temporal.feedFromArray(self, inj)
if self.spatial is base.NotGiven:
self.parsedMOC = None
else:
self.parsedMOC = pgsphere.SMoc.fromASCII(self.spatial)
[docs] def onParentComplete(self):
if self.spatial is not base.NotGiven:
self.parent.addMeta("coverage.spatial", self.spatial)
for interval in self.temporal:
self.parent.addMeta("coverage.temporal",
self._temporal.itemAttD.unparse(interval))
for interval in self.spectral:
self.parent.addMeta("coverage.spectral",
self._spectral.itemAttD.unparse(interval))