Commit 0a9da36b authored by Alessandro Frigeri's avatar Alessandro Frigeri
Browse files

Updated the classes with dataclass

parent 1e77a7f7

moondb/__pycache__/__init__.cpython-37.pyc

deleted100644 → 0
−176 B

File deleted.

View file

moondb/__pycache__/core.cpython-37.pyc

deleted100644 → 0
−6.18 KiB

File deleted.

View file

moondb/__pycache__/helpers.cpython-37.pyc

deleted100644 → 0
−3.54 KiB

File deleted.

View file

moondb/core.py

+197 −33
Original line number Diff line number Diff line
@@ -5,46 +5,53 @@ 
# MoonDB Python module

# https://realpython.com/api-integration-in-python/



import requests

import sys

import json 

import urllib.parse

import urllib3,socket

import logging



from dataclasses import dataclass

from collections import namedtuple



import requests

#import attr



class Mission:

   def __init__(self,name):

      self.name = name

   def __str__(self):

      return self.name



@dataclass

class Landmark:

   def __init__(self,name):

      self.name = name

   """The Landmark class"""

   name: str

   GPNFID: int

   GPNFURL: str

   latitude: float

   longitude: float



   def asWkt(self):

      point = "POINT ({} {})"

      return point.format(self.longitude,self.latitude)



@dataclass

class Specimen:

   def __init__(self):

       self.specimenCode = ""

       self.specimenName = ""

       self.parentSpecimen = ""

       self.childSpecimens = ""

       self.specimenType = ""

       self.samplingTechnique = ""

       self.mission = ""

       self.landmark = ""

       self.lunarStation = ""

       self.returnContainer = ""

       self.weight = ""

       self.pristinity = ""

       self.pristinityDate = ""

       self.description = ""

   def say_hello(self):

      print('Hello Moon!')

   def __str__(self):

      return self.specimenName

   def __repr__(self):

      return self.specimenName

   specimenCode: str

   specimenName: str

   parentSpecimen: str

   childSpecimens: list

   specimenType: str

   samplingTechnique: str

   mission: str

   landmark: str

   lunarStation: str

   returnContainer: str

   weight: str

   pristinity: str

   pristinityDate: str

   description: str





class SpecimenType:

   def __init__(self,name):
@@ -92,9 +99,14 @@ class AnalyisMethod: 
   def __init__(self,name):

      self.name = name



class Analysis:

   __slots__ = ('analysisCode','analyzedMaterial','comment','dataset','citation','dataResults')



class DataResult:

   __slots__ = ('unit', 'laboratory', 'variable', 'methodName', 'methodComment', 'value', 'methodCode')



def _url(path):

   print(urllib.parse.quote(path))

   #print(urllib.parse.quote(path))

   return "http://api.moondb.org" + urllib.parse.quote(path)



def _check_resp(resp):
@@ -115,12 +127,15 @@ def _get_resp(path): 
      sys.exit(0)

   _check_resp(resp)

   r = resp.json()

   #print(r)

   # To be checked with Peng

   count = r['count']

   if 'results' in r:

      return count,r['results']

   if 'result' in r:

      return count,r['result']

   if 'result' and 'count' in r:

      return r['count'],r['result']

   if 'results' and 'count' in r:

      return r['count'],r['results']

   else:

      return r







def _json_object_hook(d):
@@ -130,9 +145,158 @@ def json2obj(data): 
   return json.loads(data, object_hook=_json_object_hook)





def get_specimens(sc=None,mn=None,ln=None,sty=None,ste=None):

   '''

   Returns the specimen by specifying:

   

   Parameters

   ----------

   sc: list 

       list of specimen codes

   mn: list

       list of mission names

   ln: list

       list of landmark names

   sty: list

       list of specimen types

   ste: list

       list of sampling techniques

   '''

   sp_list = []

   if sc:

      for s in sc: 

         spec = _get_resp('/specimen/'+s) 

         sp_list.append(spec)

   if mn:

      for n in mn:

         spec = _get_resp('/specimenlist/mission/'+n) 

         sp_list.append(spec)



   if ln:

      for n in ln:

         spec = _get_resp('/specimenlist/mission/'+n) 

         sp_list.append(spec)



   if sty:

      for st in sty:

         spec = _get_resp('/specimenlist/mission/'+st) 

         sp_list.append(spec)



   if ste:

      for st in ste:

         spec = _get_resp('/specimenlist/mission/'+st) 

         sp_list.append(spec)



   sp_obj_list = []



   for s in sp_list:

      # dict unpack

      s_o = Specimen(**s)

      sp_obj_list.append(s_o)



   return sp_obj_list





def get_missions():

   missions = []

   resp = requests.get(_url('/authorities/missions/'))

   _check_resp(resp)

   for m_item in resp.json()['results']:

      missions.append( Mission(m_item['name'] ))

   return missions



## Controlled Vocabularies



def get_specimentypes():

   st_list = []

   count,st = _get_resp('/cv/specimentypes') 

   for s in st:

      stobj = SpecimenType(s['name'])

      st_list.append(stobj)

   return st_list



def get_samplingtechniques():

   st_list = []

   count,st = _get_resp('/cv/samplingtechniques') 

   for s in st:

      stobj = SamplingTechnique(s['name'])

      st_list.append(stobj)

   return st_list



def get_analyzedmaterials():

   st_list = []

   count,st = _get_resp('/cv/analyzedmaterials') 

   for s in st:

      stobj = AnalyzedMaterial(s['name'])

      st_list.append(stobj)

   return st_list



def get_analytes():

   analytes = []

   count,an = _get_resp('/cv/analyzedmaterials') 

   for a in an:

      analytes.append( Analyte(m_item['name'] ))

   return analytes



def get_analysismethods():

   am_list = []

   count,am = _get_resp('/cv/analysismethods') 

   for a in am:

      aobj = AnalysisMethod(s['name'])

      am_list.append(aobj)

   return am_list



def get_landmarks():

   lm_list = []

   count,lmlist = _get_resp('/authorities/landmarks') 

   for l in lmlist:

      lobj = Landmark(**l)

      lm_list.append(lobj)

   return lm_list



def get_landmark( landmark_name ):

   lms = get_landmarks()

   for l in lms:

      if l.name == landmark_name:

         return l

   return None



def get_samplingtechnique():

   pass







class AnalysisFilter:

   def __init__(self):

      self.mission = []

      self.landmark = []

      self.specimenType = []

      self.samplingTechnique = []

      self.analyzedMaterial = []

      self.analyte = []

      self.analysisMethod = []

   def _toJSON(self):

      return json.dumps(self, default=lambda o: o.__dict__,sort_keys=True,separators=(",", ":"))



   def get_results(self):

      resp = requests.get(_url('/data/'+self._toJSON() ))

      res_list = []

      

      for r in resp.json()['results']:

         rd = dict(r)

         analysis = namedtuple("Analysis", rd.keys())(*rd.values())

         data_res_list = []

         for r in analysis.dataResults:

            data_res = namedtuple("dataResult", r.keys())(*r.values())

            data_res_list.append(data_res)

         analysis = analysis._replace(dataResults = data_res_list )

         res_list.append(analysis) 

      return res_list





if __name__ == "__main__":

   m = get_missions()

   f = Filter()

   f = AnalysisFilter()

   f.specimenType = ["SOIL"]

   f.analyte = ["H2O","Ti"]

   f.mission = ["Apollo 11"]

moondb/helpers.py

+3 −119
Original line number Diff line number Diff line
@@ -4,125 +4,9 @@ 
# 

# MoonDB Python module



def get_specimen(sc=None,mn=None,ln=None,sty=None,ste=None):

   '''

   Get specimen by:

   - specimen code - sc

   - mission name - mn

   - landmark name - ln

   - speciment type - sty

   - sampling technique - ste

   '''

   sp_list = []

   if sc:

      for s in sc: 

         count,spec = _get_resp('/specimen/'+s) 

         sp_list.extend(spec)

   if mn:

      for n in mn:

         count,spec = _get_resp('/specimenlist/mission/'+n) 

         sp_list.extend(spec)



   if ln:

      for n in ln:

         count,spec = _get_resp('/specimenlist/mission/'+n) 

         sp_list.extend(spec)



   if sty:

      for st in sty:

         count,spec = _get_resp('/specimenlist/mission/'+st) 

         sp_list.extend(spec)



   if ste:

      for st in ste:

         count,spec = _get_resp('/specimenlist/mission/'+st) 

         sp_list.extend(spec)

from .core import *



from collections import namedtuple

import requests



   sp_obj_list = []



   for s in sp_list:

      s_obj = namedtuple("Specimen", s.keys())(*s.values())

      sp_obj_list.append(s_obj)



   return sp_obj_list





def get_missions():

   missions = []

   resp = requests.get(_url('/authorities/missions/'))

   _check_resp(resp)

   for m_item in resp.json()['results']:

      missions.append( Mission(m_item['name'] ))

   return missions



## Controlled Vocabularies



def get_specimentypes():

   st_list = []

   count,st = _get_resp('/cv/specimentypes') 

   for s in st:

      stobj = SpecimenType(s['name'])

      st_list.append(stobj)

   return st_list



def get_samplingtechniques():

   st_list = []

   count,st = _get_resp('/cv/samplingtechniques') 

   for s in st:

      stobj = SamplingTechnique(s['name'])

      st_list.append(stobj)

   return st_list



def get_analyzedmaterials():

   st_list = []

   count,st = _get_resp('/cv/analyzedmaterials') 

   for s in st:

      stobj = AnalyzedMaterial(s['name'])

      st_list.append(stobj)

   return st_list



def get_analytes():

   analytes = []

   count,an = _get_resp('/cv/analyzedmaterials') 

   for a in an:

      analytes.append( Analyte(m_item['name'] ))

   return analytes



def get_analysismethods():

   am_list = []

   count,am = _get_resp('/cv/analysismethods') 

   for a in am:

      aobj = AnalysisMethod(s['name'])

      am_list.append(aobj)

   return am_list



def get_landmark():

   pass



def get_samplingtechnique():

   pass







class Filter:

   def __init__(self):

      self.mission = []

      self.landmark = []

      self.specimenType = []

      self.samplingTechnique = []

      self.analyzedMaterial = []

      self.analyte = []

      self.analysisMethod = []

   def _toJSON(self):

      return json.dumps(self, default=lambda o: o.__dict__,sort_keys=True,separators=(",", ":"))



   def get_results(self):

      resp = requests.get(_url('/data/'+self._toJSON() ))

      res_list = []

      for r in resp.json()['results']:

         rd = dict(r)

         res_list.append(rd) 

      return res_list