"""
Writing annotations in RDs.
This module provides the glue between annotations (typically in SIL)
and the rest of the RDs. It provides the ResAnnotation struct, which
contains the SIL, and the makeAttributeAnnotation function that is a factory
for attribute annotations.
"""
#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.
import functools
import itertools
from gavo import base
from gavo.dm import annotations
from gavo.dm import common
from gavo.dm import sil
from gavo.utils import stanxml
from gavo.utils import algotricks
stanxml.registerPrefix("vo-dml", "http://www.ivoa.net/xml/VODML/v1",
stanxml.schemaURL("vo-dml.xsd"))
[docs]class SynthesizedRoles(base.Structure):
"""DM annotation copied and adapted to a new table.
This is a stand-in for DataModelRoles in tables not parsed from
XMLs. Their DM structure is defined through references the columns and
params make to the annotations their originals had.
These have no attributes but just arrange for the new annotations
to be generated.
"""
name_ = "_synthesizedRoles"
[docs] def synthesizeAnnotations(self, rd, ctx):
# don't overwrite existing annotations (usually, they
# will be the result of a previous run when multiple
# annotations are present).
if self.parent.annotations:
return
annotationMap = {}
# First, construct a map from column/param annotations in
# the old annotations to new columns and params. As a
# side effect, the dmRoles of the new items are cleared.
for item in itertools.chain(self.parent.params, self.parent.columns):
if item.parent!=self.parent:
# Don't annotate things that don't belong to us. We need to
# be careful there because in the current architecture, we
# would be changing dmRoles
continue
if isinstance(item.dmRoles, list):
# item has been processed before, probably as part of another
# DM declaration. No need to repeat that.
continue
for annotation in item.dmRoles.oldRoles:
# annotation is a weakref here; skip the annotation if the
# referenced thing has gone away
resolved = annotation()
if resolved is not None:
annotationMap[resolved] = item
item.dmRoles = []
# Then, collect the annotations we have to copy
oldInstances = set(ann.instance() for ann in annotationMap)
# tell the instances to copy themselves, replacing references
# accordingly.
newInstances = []
for oldInstance in oldInstances:
newInstances.append(
oldInstance.copyWithAnnotationMap(
annotationMap, self.parent, None))
self.parent.annotations = newInstances
[docs] def completeElement(self, ctx):
ctx.addExitFunc(self.synthesizeAnnotations)
super().completeElement(ctx)
[docs]class DataModelRoles(base.Structure):
"""an annotation of a table in terms of data models.
The content of this element is a Simple Instance Language clause.
"""
# We used to defer the SIL parsing until the end of the XML parse
# in order to be relaxed about forward references. That, however,
# is not a good idea because later elements in the RD may copy the table,
# and its annotations will be incomplete then. Instead, we're now
# building the annotation when the element is complete and worry
# about cross-table forward references when they turn up.
#
# There's an additional complication in that we may want to
# access parsed annotations while parsing other annotations
# (e.g., when processing foreign keys).
# To allow the thing to "parse itself" in such situations, we do
# all the crazy magic with the _buildAnnotation function.
name_ = "dm"
_sil = base.DataContent(description="SIL (simple instance language)"
" annotation.", copyable=True)
[docs] def getAnnotation(self, roleName, container, instance):
return annotations.GroupRefAnnotation(roleName, self.parse(), instance)
[docs] def onParentComplete(self):
self.parent.annotations.append(self.parse())
[docs] def parse(self):
"""returns a parsed version of the embedded annotation.
The parse really is only run once, but you should only call it once
the embedding table is complete, because this needs to resolve table
params and columns (and possibly other dm elements).
"""
if hasattr(self, "_parsedAnnotation"):
return self._parsedAnnotation
try:
self._parsedAnnotation = sil.getAnnotation(
self.content_, getAnnotationMaker(self.parent))
except Exception as ex:
base.ui.notifyInfo("While parsing the SIL below: %s"%str(ex))
base.ui.notifyInfo(self.content_)
raise base.ui.logOldExc(base.StructureError(str(ex),
pos=self.getSourcePosition()))
return self._parsedAnnotation
[docs] def copy(self, newParent, ctx):
# we use the general mechanism used for recovering annotations from
# columns and params in tables here so we're independent of
# changes in columns.
return SynthesizedRoles(newParent).finishElement(ctx)
[docs]class DataModelRolesAttribute(base.StructListAttribute):
"""an attribute allowing data model annotation using SIL.
It will also give an annotations attribute on the instance, and a
getAnnotationsOfType method letting you pull out a specific annotation.
For situation where an existing annotation should be copied from
annotations coming in through columns and/or params, the attribute
also gives an updateAnnotationFromChildren method. It will do
nothing if some annotation already exists.
"""
def __init__(self):
base.StructListAttribute.__init__(self,
"dm",
childFactory=DataModelRoles,
description="Annotations for data models.",
copyable=True)
[docs] def iterParentMethods(self):
def iterAnnotationsOfType(instance, typeName):
"""returns the first annotation of the type passed.
"""
for ann in instance.annotations:
if ann.type==typeName:
yield ann
yield ("iterAnnotationsOfType", iterAnnotationsOfType)
def updateAnnotationFromChildren(instance):
roleSynthesizer = SynthesizedRoles(instance)
roleSynthesizer.synthesizeAnnotations(None, None)
yield ("updateAnnotationFromChildren", updateAnnotationFromChildren)
[docs]def makeAttributeAnnotation(container, instance, attName, attValue):
"""returns a typed annotation for attValue within container.
When attValue is a literal, this is largely trivial. If it's a reference,
this figures out what it points to and creates an annotation of
the appropriate type (e.g., ColumnAnnotation, ParamAnnotation, etc).
container in current DaCHS should be a TableDef or something similar;
this function expects at least a getByName function and an rd attribute.
instance is the root of the current annotation. Complex objects should
keep a (weak) reference to that. We don't have parent links in
our dm trees, and without a reference to the root there's no
way we can go "up".
This is usually used as a callback from within sil.getAnnotation and
expects Atom and Reference instances as used there.
"""
if isinstance(attValue, sil.Atom):
return common.AtomicAnnotation(attName, attValue, instance=instance)
elif isinstance(attValue, sil.Reference):
# try name-resolving first (resolveId only does id resolving on
# unadorned strings)
try:
res = container.getByName(attValue)
except base.NotFoundError:
if container.rd:
res = base.resolveId(container.rd, attValue, instance=container)
else:
raise
if not hasattr(res, "getAnnotation"):
raise base.StructureError("Element %s cannot be referenced"
" within a data model."%repr(res))
return res.getAnnotation(attName, container, instance)
else:
assert False
[docs]def getAnnotationMaker(container):
"""wraps makeAttributeAnnotationMaker such that names are resolved
within container.
"""
return functools.partial(makeAttributeAnnotation, container)
def _splitPath(path:str):
"""returns a pair of (prefix)/(stuff-without-slash) for path.
"""
splitpoint = path.rfind("/")
if splitpoint==-1:
return "", path
else:
return path[:splitpoint], path[splitpoint+1:]
def _makeDMVal(stcObj):
"""returns a value palatable to makeAttributeAnnotation from stcObj
(which essentially will be an set.ColRef or some literal).
"""
if hasattr(stcObj, "dest"):
return sil.Reference(stcObj.dest)
else:
return sil.Atom(stcObj)
[docs]def buildAdhocSTC(table, vals):
"""builds a modern adhoc STC annotation in table based on a mapping
of attribute paths to annotation targets in vals.
This is only necessary as long as we want to support <stc> specs. The
keys here are, in effect defined in rscdef.tabledef; see _utypeToModern.
"""
if vals.pop("!!!flavor!!!", None)!="SPHERICAL":
# anything else would need a lot more thought, and that's certainly not
# worth it. STC-S should die.
return
# Sigh. STC1 split off the J/B from the epoch value. Let's cope in
# some way.
if vals.pop("!!!yearDef!!!", "J")!="J":
raise NotImplementedError("We can only deal with Julian epochs")
if "/space/frame/epoch" in vals:
if isinstance(vals["/space/frame/epoch"], str):
vals["/space/frame/epoch"] = "J"+vals["/space/frame/epoch"]
annMaker = getAnnotationMaker(table)
# astid is the id of the STC1 AST that has produced the original utypes.
# We want to record this because in votablewrite we want to re-use the
# COOSYS that already exists when there is an STC1 annotation.
astid = None
objs = {}
for path in algotricks.pathSort(vals.keys()):
prefix, attname = _splitPath(path)
parent = objs[prefix] if path else None
if path in buildAdhocSTC.typesForPaths:
# it's an intermediate object: create it
objs[path] = common.ObjectAnnotation(
attname, buildAdhocSTC.typesForPaths[path], objs.get(""))
if parent is not None:
parent.add(objs[path])
else:
# it's an attribute; translate stc's references and literals
parent.add(annMaker(objs[""], attname, _makeDMVal(vals[path])))
if astid is None:
# if value is a ColRef, whatever it refers to will define our
# STC1 AST
try:
destCol = table.getElementForName(vals[path].dest)
astid = str(id(destCol.stc))
objs[""].add(annMaker(objs[""], "astid", sil.Atom(astid)))
except (base.NotFoundError, AttributeError):
# not a ColRef, apparently, try next val
pass
table.annotations.append(objs[""])
buildAdhocSTC.typesForPaths = {
"": "votable:Coords",
"/space": "votable:SphericalCoordinate",
"/space/frame": "votable:SpaceFrame",
"/time": "votable:TimeCoordinate",
"/time/frame": "votable:TimeFrame",
}