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)


    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)

    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)