orthancInterface/orthancDatabaseBrowser.py

import orthancInterface
import json
import chardet
import io
import sys

def extractJSON(data):
   if len(data)==0:
      return {}
   encoding=chardet.detect(data)["encoding"]
   #try with a set of encodings to maximize probability of success
   encodings=[encoding,'utf_8']
   for x in encodings:
      try:
         return json.loads(data.decode(x))
      except (UnicodeDecodeError,TypeError):
         pass

      print('Failed to decode with [{}]: {}'.format(x,data))

   return None


class orthancDB:
   def __init__(self,net):
      self.net=net

   def getList(self,category):
      url=self.net.getCoreURL()
      url+='/'+category
      response=self.net.get(url)
      return extractJSON(response.data)
    
   def getPatients(self):
      return self.getList('patients')

   def getStudies(self):
      return self.getList('studies')
    
   def getPatientData(self, orthancId):
      return self.getData('patients',orthancId)

   def getStudyData(self, studyOrthancId):
      return self.getData('studies',studyOrthancId)

   def getSeriesData(self, seriesOrthancId):
      return self.getData('series',seriesOrthancId)

   def getNumpy(self,level,orthancId):
      url=self.net.getCoreURL()
      url+='/'+level+'/'+orthancId+'/numpy'
      response=self.net.get(url,binary=True)
      return io.BytesIO(response.data)

   def getData(self, level, orthancId):
      url=self.net.getCoreURL()
      url+='/'+level+'/'+orthancId
      response=self.net.get(url)
      return extractJSON(response.data)


   def getDicomField(self,orthancInstanceId,dicomField,raw=False):
      url=self.net.getCoreURL()
      url+='/instances/'+orthancInstanceId+'/content/'+dicomField
      response=self.net.get(url)
      if raw:
         return response.data
      try:
         encoding=chardet.detect(response.data)["encoding"]
         return response.data.decode(encoding)
      except TypeError:
         return ''

   def selectRows(self, qfilter):
      obj={}
      obj["Level"]="Instance"
      obj["Query"]=qfilter
      jsonData=json.dumps(obj)

      url=self.net.getCoreURL()
      url+='/tools/find'
      response=self.net.post(url,jsonData,'json')
      return extractJSON(response.data)

   def find(self,par):
      v={"Level":par.get("Level","Instance"),
         "Query":par["Query"]}
      apiURL='/'.join([self.net.getCoreURL(),'tools','find'])
      response=self.net.post(apiURL,json.dumps(v),'json')
      return extractJSON(response.data)



   def upload(self, f) :
      url=self.net.getCoreURL()
      url+='/instances'
      fobj=open (f, 'rb')
      response=self.net.post(url,fobj.read(),'octet-stream')
      return extractJSON(response.data)

   def remove(self,level,orthancId):
      url=self.net.getCoreURL()
      url+='/'+level+'/'+orthancId
      response=self.net.delete(url)
      print(response.data)

   def changes(self,limit=10,since=0):
      url=self.net.getCoreURL()
      #url+=f'/changes?limit={limit}&since={since}'
      url+='/changes?limit={}&since={}'.format(limit,since)
      response=self.net.get(url)
      return extractJSON(response.data)