__author__ = 'sreeder'
from sqlalchemy import func ,not_, bindparam, distinct, exists
import pandas as pd
from odm2api.ODM2 import serviceBase
from odm2api.ODM2.models import *
[docs]class DetailedResult:
def __init__(self, action, result,
sc, sn,
method, variable,
processingLevel,
unit):
# result.result_id etc.
self.ResultID = result.ResultID
self.SamplingFeatureCode = sc#.SamplingFeatureCode
self.MethodCode = method.MethodCode
self.VariableCode = variable.VariableCode
self.ProcessingLevelCode = processingLevel.ProcessingLevelCode
self.UnitsName = unit.UnitsName
self.SamplingFeatureName = sn#.SamplingFeatureName
self.MethodName = method.MethodName
self.VariableNameCV = variable.VariableNameCV
self.ProcessingLevelDefinition = processingLevel.Definition
self.ValueCount = result.ValueCount
self.BeginDateTime = action.BeginDateTime
self.EndDateTime = action.EndDateTime
self.ResultObj = result
[docs]class DetailedAffiliation:
def __init__(self, affiliation, person, org):
self.AffiliationID = affiliation.AffiliationID
self.Name = person.PersonFirstName + \
" " + \
person.PersonLastName
self.Organization = "(" + org.OrganizationCode + ") " + \
org.OrganizationName
# def __repr__(self):
# return str(self.name) + " " + str(self.organization)
[docs]class ReadODM2(serviceBase):
# ################################################################################
# Exists functions
# ################################################################################
[docs] def resultExists(self, result):
"""
resultExists(self, result):
Check to see if a Result Object exists
* Pass Result Object - return a boolean value of wether the given object exists
"""
# unique Result
# FeatureActionID, ResultTypeCV, VariableID, UnitsID, ProcessingLevelID, SampledMediumCV
try:
ret = self._session.query(exists().where(Results.ResultTypeCV == result.ResultTypeCV)
.where(Results.VariableID == result.VariableID)
.where(Results.UnitsID == result.UnitsID)
.where(Results.ProcessingLevelID == result.ProcessingLevelID)
.where(Results.SampledMediumCV == result.SampledMediumCV)
)
# where(Results.FeatureActionID == result.FeatureActionID).
return ret.scalar()
except:
return None
# ################################################################################
# Annotations
# ################################################################################
[docs] def getAnnotations(self, type=None, codes=None, ids=None):
"""
def getAnnotations(self, type=None, codes = None, ids = None):
* Pass Nothing - return a list of all objects
* Pass AnnotationTypeCV - return a list of all objects of the fiven type
* Pass a list of codes - return a list of objects, one for each of the given codes
* Pass a list of ids -return a list of objects, one for each of the given ids
"""
# TODO What keywords do I use for type
a = Annotations
if type:
if type == "action":
a = ActionAnnotations
elif type == "categoricalresultvalue":
a = CategoricalResultValueAnnotations
elif type == "equipmentannotation":
a = EquipmentAnnotations
elif type == "measurementresultvalue":
a = MeasurementResultValueAnnotations
elif type == "method":
a = MethodAnnotations
elif type == "pointcoverageresultvalue":
a = PointCoverageResultValueAnnotations
elif type == "profileresultvalue":
a = ProfileResultValueAnnotations
elif type == "result":
a = ResultAnnotations
elif type == "samplingfeature":
a = SamplingFeatureAnnotations
elif type == "sectionresultvalue":
a = SectionResultValueAnnotations
elif type == "spectraresultvalue":
a = SpectraResultValueAnnotations
elif type == "timeseriesresultvalue":
a = TimeSeriesResultValueAnnotations
elif type == "trajectoryresultvalue":
a = TrajectoryResultValueAnnotations
elif type == "transectresultvalue":
a = TransectResultValueAnnotations
try:
query=self._session.query(a)
if codes:
query = query.filter(Annotations.AnnotationCode.in_(codes))
if ids:
query = query.filter(Annotations.AnnotationID.in_(ids))
return query.all()
except:
return None
# ################################################################################
# CV
# ##############################################################################
[docs] def getCVs(self, type):
"""
getCVs(self, type):
* Pass CVType - return a list of all objects of the given type
"""
CV = CVActionType
if type == "actiontype":
CV = CVActionType
elif type == "aggregationstatistic":
CV = CVAggregationStatistic
elif type == "annotationtype":
CV = CVAnnotationType
elif type == "censorcode":
CV = CVCensorCode
elif type == "dataqualitytype":
CV = CVDataQualityType
elif type == "dataset type":
CV = CVDataSetType
elif type == "Directive Type":
CV = CVDirectiveType
elif type == "Elevation Datum":
CV = CVElevationDatum
elif type == "Equipment Type":
CV = CVEquipmentType
elif type == "Medium":
CV = CVMediumType
elif type == "Method Type":
CV = CVMethodType
elif type == "Organization Type":
CV = CVOrganizationType
elif type == "Property Data Type":
CV = CVPropertyDataType
elif type == "Quality Code":
CV = CVQualityCode
elif type == "Relationship Type":
CV = CVRelationshipType
elif type == "Result Type":
CV = CVResultType
elif type == "Sampling Feature Geo-type":
CV = CVSamplingFeatureGeoType
elif type == "Sampling Feature Type":
CV = CVSamplingFeatureType
elif type == "Site Type":
CV = CVSiteType
elif type == "Spatial Offset Type":
CV = CVSpatialOffsetType
elif type == "Speciation":
CV = CVSpeciation
elif type == "Specimen Type":
CV = CVSpecimenType
elif type == "Status":
CV = CVStatus
elif type == "Taxonomic Classifier Type":
CV = CVTaxonomicClassifierType
elif type == "Units Type":
CV = CVUnitsType
elif type == "Variable Name":
CV = CVVariableName
elif type == "Variable Type":
CV = CVVariableType
else:
return None
try:
return self._session.query(CV).all()
except Exception as e:
print("Error running Query: %s" % e)
# ################################################################################
# Core
# ################################################################################
[docs] def getDetailedAffiliationInfo(self):
"""
getDetailedAffiliationInfo(self)
* Pass Nothing - Return a list of all Affiliations with detailed information, including Affiliation, People and Organization
"""
q = self._session.query(Affiliations, People, Organizations) \
.filter(Affiliations.PersonID == People.PersonID) \
.filter(Affiliations.OrganizationID == Organizations.OrganizationID)
affiliationList = []
for a, p, o in q.all():
detailedAffiliation = DetailedAffiliation(a, p, o)
affiliationList.append(detailedAffiliation)
return affiliationList
[docs] def getDetailedResultInfo(self, resultTypeCV=None, resultID=None, sfID=None):
#TODO can this be done by just getting the result object and drilling down? what is the performance comparison
"""
getDetailedResultInfo(self, resultTypeCV=None, resultID=None, sfID=None)
Get detailed information for all selected Results including , unit info, site info,
method info , ProcessingLevel info.
* Pass nothing - return a list of all objects
* Pass resultTypeCV - All objects of given type
* Pass a result ID - single object with the given result ID
* Pass a SamplingFeatureID - All objects associated with the given sampling feature.
"""
q = self._session.query(Actions, Results, SamplingFeatures.SamplingFeatureCode, SamplingFeatures.SamplingFeatureName, Methods, Variables,
ProcessingLevels, Units).filter(Results.VariableID == Variables.VariableID) \
.filter(Results.UnitsID == Units.UnitsID) \
.filter(Results.FeatureActionID == FeatureActions.FeatureActionID) \
.filter(FeatureActions.SamplingFeatureID == SamplingFeatures.SamplingFeatureID) \
.filter(FeatureActions.ActionID == Actions.ActionID) \
.filter(Actions.MethodID == Methods.MethodID) \
.filter(Results.ProcessingLevelID == ProcessingLevels.ProcessingLevelID) \
.filter(Results.ResultTypeCV == resultTypeCV) \
.order_by(Results.ResultID)
resultList = []
if sfID:
q = q.filter(SamplingFeatures.SamplingFeatureID == sfID)
if resultID:
q = q.filter(Results.ResultID==resultID)
for a, r, sc, sn, m, v, p, u in q.all():
detailedResult = DetailedResult( \
a, r, sc, sn, m, v, p, u)
resultList.append(detailedResult)
return resultList
"""
Taxonomic Classifiers
"""
[docs] def getTaxonomicClassifiers(self):
"""
getTaxonomicClassifiers(self):
* Pass nothing - return a list of all objects
"""
return self._session.query(TaxonomicClassifiers).all()
"""
Variable
"""
[docs] def getVariables(self, ids=None, codes=None, sitecode=None, results= False):
"""
getVariables(self, ids=None, codes=None, sitecode=None, results= False):
* Pass nothing - returns full list of variable objects
* Pass a list of VariableID - returns a single variable object
* Pass a list of VariableCode - returns a single variable object
* Pass a SiteCode - returns a list of Variable objects that are collected at the given site.
* Pass whether or not you want to return the sampling features that have results associated with them
"""
if sitecode:
try:
vars = [x[0] for x in
self._session.query(distinct(Results.VariableID))
.filter(Results.FeatureActionID == FeatureActions.FeatureActionID)
.filter(FeatureActions.SamplingFeatureID == SamplingFeatures.SamplingFeatureID)
.filter(SamplingFeatures.SamplingFeatureCode == sitecode).all()
]
if ids:
ids = list(set(ids).intersection(vars))
else:
ids = vars
except:
pass
if results:
try:
vars = [x[0] for x in self._session.query(distinct(Results.VariableID)).all()]
if ids:
ids = list(set(ids).intersection(vars))
else:
ids = vars
except:
pass
query = self._session.query(Variables)
if ids: query = query.filter(Variables.VariableID.in_(ids))
if codes: query = query.filter(Variables.VariableCode.in_(codes))
try:
return query.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
Method
"""
[docs] def getMethods(self, ids=None, codes=None, type=None):
"""
getMethods(self, ids=None, codes=None, type=None):
* Pass nothing - returns full list of method objects
* Pass a list of MethodIDs - returns a single method object for each given id
* Pass a list of MethodCode - returns a single method object for each given code
* Pass a MethodType - returns a list of method objects of the given MethodType
"""
q = self._session.query(Methods)
if ids: q = q.filter(Methods.MethodID.in_(ids))
if codes: q = q.filter(Methods.MethodCode.in_(codes))
if type: q = q.filter_by(MethodTypeCV=type)
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
ProcessingLevel
"""
[docs] def getProcessingLevels(self, ids=None, codes=None):
"""
getProcessingLevels(self, ids=None, codes=None)
* Pass nothing - returns full list of ProcessingLevel objects
* Pass a list of ProcessingLevelID - returns a single processingLevel object for each given id
* Pass a list of ProcessingLevelCode - returns a single processingLevel object for each given code
"""
q = self._session.query(ProcessingLevels)
if ids: q = q.filter(ProcessingLevels.ProcessingLevelsID.in_(ids))
if codes: q = q.filter(ProcessingLevels.ProcessingLevelCode.in_(codes))
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
Sampling Feature
"""
[docs] def getSamplingFeatures(self, ids=None, codes=None, uuids=None, type=None, wkt=None, results=False):
"""
getSamplingFeatures(self, ids=None, codes=None, uuids=None, type=None, wkt=None, results=False):
* Pass nothing - returns a list of all sampling feature objects with each object of type specific to that sampling feature
* Pass a list of SamplingFeatureID - returns a single sampling feature object for the given ids
* Pass a list of SamplingFeatureCode - returns a single sampling feature object for the given code
* Pass a list of SamplingFeatureUUID - returns a single sampling feature object for the given UUID's
* Pass a SamplingFeatureType - returns a list of sampling feature objects of the type passed in
* Pass a SamplingFeature Well Known Text - return a list of sampling feature objects
* Pass whether or not you want to return only the sampling features that have results associated with them
"""
if results:
try:
fas = [x[0] for x in self._session.query(distinct(Results.FeatureActionID)).all()]
except:
return None
sf = [x[0] for x in self._session.query(distinct(FeatureActions.SamplingFeatureID))
.filter(FeatureActions.FeatureActionID.in_(fas)).all()]
if ids:
ids = list(set(ids).intersection(sf))
else:
ids = sf
q = self._session.query(SamplingFeatures)
if type: q = q.filter_by(SamplingFeatureTypeCV=type)
if ids: q = q.filter(SamplingFeatures.SamplingFeatureID.in_(ids))
if codes: q = q.filter(SamplingFeatures.SamplingFeatureCode.in_(codes))
if uuids: q = q.filter(SamplingFeatures.SamplingFeatureUUID.in_(uuids))
if wkt: q = q.filter_by(FeatureGeometryWKT=wkt)
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
Action
"""
[docs] def getActions(self, ids=None, type=None, sfid=None):
"""
getActions(self, ids=None, type=None, sfid=None)
* Pass nothing - returns a list of all Actions
* Pass a list of Action ids - returns a list of Action objects
* Pass a ActionTypeCV - returns a list of Action objects of that type
* Pass a SamplingFeature ID - returns a list of Action objects associated with that Sampling feature ID, Found through featureAction table
"""
a = Actions
if type == "equipment":
a = EquipmentActions
elif type == "calibration":
a = CalibrationActions
elif type == "maintenance":
a = MaintenanceActions
q = self._session.query(a)
if ids: q = q.filter(a.ActionID.in_(ids))
if sfid:
q = q.join(FeatureActions).filter(FeatureActions.SamplingFeatureID == sfid)
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
Unit
"""
[docs] def getUnits(self, ids=None, name=None, type=None):
"""
getUnits(self, ids=None, name=None, type=None)
* Pass nothing - returns a list of all units objects
* Pass a list of UnitsID - returns a single units object for the given id
* Pass UnitsName - returns a single units object
* Pass a type- returns a list of all objects of the given type
"""
q = self._session.query(Units)
if ids: q = q.filter(Units.UnitsID.in_(ids))
if name: q = q.filter(Units.UnitsName.ilike(name))
if type: q = q.filter(Units.UnitsTypeCV.ilike(type))
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
Organization
"""
[docs] def getOrganizations(self, ids=None, codes=None):
"""
getOrganizations(self, ids=None, codes=None)
* Pass nothing - returns a list of all organization objects
* Pass a list of OrganizationID - returns a single organization object
* Pass a list of OrganizationCode - returns a single organization object
"""
q = self._session.query(Organizations)
if ids: q = q.filter(Organizations.OrganizationID.in_(ids))
if codes: q = q.filter(Organizations.OrganizationCode.in_(codes))
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
Person
"""
[docs] def getPeople(self, ids=None, firstname=None, lastname=None):
"""
getPeople(self, ids=None, firstname=None, lastname=None)
* Pass nothing - returns a list of all People objects
* Pass a list of PeopleID - returns a single People object
* Pass a First Name - returns a single People object
* Pass a Last Name - returns a single People object
"""
q = self._session.query(People)
if ids: q = q.filter(People.PersonID.in_(ids))
if firstname: q = q.filter(People.PersonFirstName.ilike(firstname))
if lastname: q = q.filter(People.PersonLastName.ilike(lastname))
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
[docs] def getAffiliations(self, ids=None, personfirst=None, personlast=None, orgcode=None):
"""Retrieve a list of Affiliation objects.
If no arguments are passed to the function, or their values are None,
all Affiliation objects in the database will be returned.
Args:
ids (list, optional): List of AffiliationIDs.
personfirst (str, optional): Person First Name.
personlast (str, optional): Person Last Name.
orgcode (str, optional): Organization Code.
Returns:
list: List of Affiliation objects
Examples:
>>> ReadODM2.getAffiliations(ids=[39,40])
>>> ReadODM2.getAffiliations(personfirst='John',
... personlast='Smith')
>>> ReadODM2.getAffiliations(orgcode='Acme')
"""
q = self._session.query(Affiliations)
if ids: q = q.filter(Affiliations.AffiliationID.in_(ids))
if orgcode: q = q.join(Affiliations.OrganizationObj).filter(
Organizations.OrganizationCode.ilike(orgcode))
if personfirst: q = q.join(Affiliations.PersonObj).filter(
People.PersonFirstName.ilike(personfirst))
if personlast: q = q.join(Affiliations.PersonObj).filter(
People.PersonLastName.ilike(personlast))
try:
return q.all()
except Exception as e:
print("Error running Query: %s"%e)
return None
"""
Results
"""
[docs] def getResults(self, ids=None, type=None, uuids=None, actionid=None, simulationid=None, sfid=None,
variableid=None, siteid=None):
# TODO what if user sends in both type and actionid vs just actionid
"""Retrieve a list of Result objects.
If no arguments are passed to the function, or their values are None,
all Result objects in the database will be returned.
Args:
ids (list, optional): List of ResultIDs.
type (str, optional): Type of Result from
`controlled vocabulary name <http://vocabulary.odm2.org/resulttype/>`_.
uuids (list, optional): List of UUIDs string.
actionid (int, optional): ActionID.
simulationid (int, optional): SimulationID.
sfid (int, optional): SamplingFeatureID.
variableid (int, optional): VariableID.
siteid (int, optional): SiteID.
Returns:
list: List of Result objects
Examples:
>>> ReadODM2.getResults(ids=[39,40])
>>> ReadODM2.getResults(type='Time series coverage')
>>> ReadODM2.getResults(sfid=65)
>>> ReadODM2.getResults(uuids=['a6f114f1-5416-4606-ae10-23be32dbc202',
... '5396fdf3-ceb3-46b6-aaf9-454a37278bb4'])
>>> ReadODM2.getResults(simulationid=50)
>>> ReadODM2.getResults(siteid=6)
>>> ReadODM2.getResults(variableid=7)
>>> ReadODM2.getResults(actionid=20)
"""
query = self._session.query(Results)
if type: query = query.filter_by(ResultTypeCV=type)
if variableid: query = query.filter_by(VariableID=variableid)
if ids: query = query.filter(Results.ResultID.in_(ids))
if uuids: query = query.filter(Results.ResultUUID.in_(uuids))
if simulationid: query = query.join(FeatureActions)\
.join(Actions)\
.join(Simulations)\
.filter_by(SimulationID=simulationid)
if actionid: query = query.join(FeatureActions).filter_by(ActionID=actionid)
if sfid: query = query.join(FeatureActions).filter_by(SamplingFeatureID=sfid)
if siteid:
sfids = [x[0] for x in self._session.query(distinct(SamplingFeatures.SamplingFeatureID))
.select_from(RelatedFeatures)
.join(RelatedFeatures.SamplingFeatureObj)
.filter(RelatedFeatures.RelatedFeatureID == siteid)
#.filter(RelatedFeatures.RelationshipTypeCV == "Was Collected at")
.all()]
query = query.join(FeatureActions).filter(FeatureActions.SamplingFeatureID.in_(sfids))
try:
return query.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
"""
Datasets
"""
[docs] def getDataSets(self, codes=None, uuids=None):
"""
getDataSets(self, codes=None, uuids=None)
* Pass nothing - returns a list of all DataSet objects
* Pass a list of DataSetCode - returns a single DataSet object for each code
* Pass a list of UUIDS - returns a single DataSet object for each UUID
"""
q = self._session.query(DataSets)
if codes:
q = q.filter(DataSets.DataSetCode.in_(codes))
if uuids:
q.q.filter(DataSets.DataSetUUID.in_(uuids))
try:
return q.all()
except Exception as e:
print("Error running Query %s" % e)
return None
# ################################################################################
# Data Quality
# ################################################################################
[docs] def getDataQuality(self):
"""
getDataQuality(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(DataQuality).all()
# TODO DataQuality Schema Queries
[docs] def getReferenceMaterials(self):
"""
getReferenceMaterials(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(ReferenceMaterials).all()
[docs] def getReferenceMaterialValues(self):
"""
getReferenceMaterialValues(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(ReferenceMaterialValues).all()
[docs] def getResultNormalizationValues(self):
"""
getResultNormalizationValues(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(ResultNormalizationValues).all()
[docs] def getResultsDataQuality(self):
"""
getResultsDataQuality(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(ResultsDataQuality).all()
# ################################################################################
# Equipment
# ################################################################################
# TODO Equipment Schema Queries
[docs] def getEquipment(self, codes=None, type=None, sfid=None, actionid=None):
"""
getEquipment(self, codes=None, type=None, sfid=None, actionid=None)
* Pass nothing - returns a list of all Equipment objects
* Pass a list of EquipmentCodes- return a list of all Equipment objects that match each of the codes
* Pass a EquipmentType - returns a single Equipment object
* Pass a SamplingFeatureID - returns a single Equipment object
* Pass an ActionID - returns a single Equipment object
"""
e = self._session.query(Equipment)
if sfid: e = e.join(EquipmentUsed) \
.join(Actions) \
.join(FeatureActions) \
.filter(FeatureActions.SamplingFeatureID == sfid)
if codes: e = e.filter(Equipment.EquipmentCode.in_(codes))
if actionid: e = e.join(EquipmentUsed).join(Actions) \
.filter(Actions.ActionID == actionid)
return e.all()
[docs] def CalibrationReferenceEquipment(self):
"""
CalibrationReferenceEquipment(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(CalibrationReferenceEquipment).all()
[docs] def CalibrationStandards(self):
"""
CalibrationStandards(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(CalibrationStandards).all()
[docs] def DataloggerFileColumns(self):
"""
DataloggerFileColumns(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(DataLoggerFileColumns).all()
[docs] def DataLoggerFiles(self):
"""
DataLoggerFiles(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(DataLoggerFiles).all()
[docs] def DataloggerProgramFiles(self):
"""
DataloggerProgramFiles(self)
* Pass Nothing - return a list of all objects
"""
return self._session.query(DataLoggerProgramFiles).all()
[docs] def EquipmentModels(self):
"""
EquipmentModels(self)
* Pass Nothing - return a list of all objects
"""
return self._session.query(EquipmentModels).all()
[docs] def EquipmentUsed(self):
"""
EquipmentUsed(self)
* Pass Nothing - return a list of all objects
"""
return self._session.query(EquipmentUsed).all()
[docs] def InstrumentOutputVariables(self, modelid=None, variableid=None):
"""
InstrumentOutputVariables(self, modelid=None, variableid=None)
* Pass Nothing - return a list of all objects
* Pass ModelID
* Pass VariableID
"""
i = self._session.query(InstrumentOutputVariables)
if modelid: i = i.filter_by(ModelID=modelid)
if variableid: i = i.filter_by(VariableID=variableid)
return i.all()
# ################################################################################
# Extension Properties
# ################################################################################
[docs] def getExtensionProperties(self, type=None):
"""
getExtensionProperties(self, type=None)
* Pass nothing - return a list of all objects
* Pass type- return a list of all objects of the given type
"""
# Todo what values to use for extensionproperties type
e = ExtensionProperties
if type == "action":
e = ActionExtensionPropertyValues
elif type == "citation":
e = CitationExtensionPropertyValues
elif type == "method":
e = MethodExtensionPropertyValues
elif type == "result":
e = ResultExtensionPropertyValues
elif type == "samplingfeature":
e = SamplingFeatureExtensionPropertyValues
elif type == "variable":
e = VariableExtensionPropertyValues
try:
return self._session.query(e).all()
except Exception as e:
print("Error running Query: %s" % e)
return None
# ################################################################################
# External Identifiers
# ################################################################################
[docs] def getExternalIdentifiers(self, type=None):
"""
getExternalIdentifiers(self, type=None)
* Pass nothing - return a list of all objects
* Pass type- return a list of all objects of the given type
"""
e = ExternalIdentifierSystems
if type.lowercase == "citation":
e = CitationExternalIdentifiers
elif type == "method":
e = MethodExternalIdentifiers
elif type == "person":
e = PersonExternalIdentifiers
elif type == "referencematerial":
e = ReferenceMaterialExternalIdentifiers
elif type == "samplingfeature":
e = SamplingFeatureExternalIdentifiers
elif type == "spatialreference":
e = SpatialReferenceExternalIdentifiers
elif type == "taxonomicclassifier":
e = TaxonomicClassifierExternalIdentifiers
elif type == "variable":
e = VariableExternalIdentifiers
try:
return self._session.query(e).all()
except Exception as e:
print("Error running Query: %s" % e)
return None
# ################################################################################
# Lab Analyses
# ################################################################################
# TODO functions for Lab Analyses
[docs] def getDirectives(self):
"""
getDirectives(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(Directives).all()
[docs] def getActionDirectives(self):
"""
getActionDirectives(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(ActionDirectives).all()
[docs] def getSpecimenBatchPositions(self):
"""
getSpecimenBatchPositions(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(SpecimenBatchPositions).all()
# ################################################################################
# Provenance
# ################################################################################
# TODO functions for Provenance
[docs] def getAuthorLists(self):
"""
getAuthorLists(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(AuthorLists).all()
[docs] def getDatasetCitations(self):
"""
getDatasetCitations(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(DataSetCitations).all()
[docs] def getDerivationEquations(self):
"""
getDerivationEquations(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(DerivationEquations).all()
[docs] def getMethodCitations(self):
"""
getMethodCitations(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(MethodCitations).all()
[docs] def getResultDerivationEquations(self):
"""
getResultDerivationEquations(self)
* Pass nothing - return a list of all objects
"""
return self._session.query(ResultDerivationEquations).all()
# ################################################################################
# Results
# ################################################################################
"""
ResultValues
"""
[docs] def getResultValues(self, resultids, starttime=None, endtime=None):
"""
getResultValues(self, resultids, starttime=None, endtime=None)
* Pass in a list of ResultID - Returns a pandas dataframe object of type that is specific to the result type -
The resultids must be associated with the same value type
* Pass a ResultID and a date range - returns a pandas dataframe object of type that is specific to the result type with values between the input date range
* Pass a starttime - Returns a dataframe with the values after the given start time
* Pass an endtime - Returns a dataframe with the values before the given end time
"""
type= self._session.query(Results).filter_by(ResultID=resultids[0]).first().ResultTypeCV
ResultType = TimeSeriesResults
if "categorical" in type.lower():ResultType = CategoricalResultValues
elif "measurement" in type.lower():ResultType = MeasurementResultValues
elif "point" in type.lower():ResultType = PointCoverageResultValues
elif "profile" in type.lower():ResultType = ProfileResultValues
elif "section" in type.lower():ResultType = SectionResults
elif "spectra" in type.lower():ResultType = SpectraResultValues
elif "time" in type.lower():ResultType = TimeSeriesResultValues
elif "trajectory" in type.lower():ResultType = TrajectoryResultValues
elif "transect" in type.lower():ResultType = TransectResultValues
# q.filter(Affiliations.AffiliationID.in_(ids))
q = self._session.query(ResultType).filter(ResultType.ResultID.in_(resultids))
if starttime: q = q.filter(ResultType.ValueDateTime >= starttime)
if endtime: q = q.filter(ResultType.ValueDateTime <= endtime)
try:
vals = q.order_by(ResultType.ValueDateTime)
# df = pd.DataFrame([dv.list_repr() for dv in vals.all()])
# df.columns = vals[0].get_columns()
query = q.statement.compile(dialect=self._session_factory.engine.dialect)
df = pd.read_sql_query(sql=query,
con=self._session_factory.engine,
params=query.params)
return df
except Exception as e:
print("Error running Query: %s" % e)
return None
# ################################################################################
# SamplingFeatures
# ################################################################################
"""
Site
"""
[docs] def getSpatialReferences(self, srsCodes=None):
"""
getSpatialReferences(self, srsCodes=None)
* Pass nothing - return a list of all Spatial References
* Pass in a list of SRS Codes-
"""
q = self._session.query(SpatialReferences)
if srsCodes: q.filter(SpatialReferences.SRSCode.in_(srsCodes))
try:
return q.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
# ################################################################################
# Simulation
# ################################################################################
[docs] def getSimulations(self, name=None, actionid=None):
"""
getSimulations(self, name=None, actionid=None)
* Pass nothing - get a list of all converter simuation objects
* Pass a SimulationName - get a single simulation object
* Pass an ActionID - get a single simulation object
"""
s = self._session.query(Simulations)
if name: s = s.filter(Simulations.SimulationName.ilike(name))
if actionid: s = s.filter_by(ActionID=actionid)
try:
return s.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
[docs] def getModels(self, codes=None):
"""
getModels(self, codes=None)
* Pass nothing - return a list of all Model Objects
* Pass a list of ModelCodes - get a list of converter objects related to the converter having ModeCode
"""
m = self._session.query(Models)
if codes: m = m.filter(Models.ModelCode.in_(codes))
try:
return m.all()
except Exception as e:
print("Error running Query: %s" % e)
return None
