orthancInterface/orthancDatabaseBrowser.py

import orthancInterface
import json
import chardet
import io
import sys

def extractJSON(data):
   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:
         pass

      if sys.version_info[1]<6:
         print('Failed to decode with [{}]: {}'.format(x,data))
      else:
         print(f'Failed to decode with [{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):
        url=self.net.getCoreURL()
        url+='/instances/'+orthancInstanceId+'/content/'+dicomField
        response=self.net.get(url)
        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,'json',jsonData)
        encoding=chardet.detect(response.data)["encoding"]
        return json.loads(response.data.decode(encoding))


    def upload(self, f) :
        url=self.net.getCoreURL()
        url+='/instances'
        fobj=open (f, 'rb')
        response=self.net.post(url,'octet-stream',fobj.read())
        try:
            encoding=chardet.detect(response.data)["encoding"]
        except AttributeError:
            print(response)
            return json.loads('{"status":"BAD"}')
        return json.loads(response.data.decode(encoding))

    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}'
      response=self.net.get(url)
      return extractJSON(response.data)