import json import requests from PIL import Image import io #def fetch_and_print_articles(api_url): # response = requests.get(api_url) # # if response.status_code == 200: # articles = response.json().get('articles', []) # # for index, article in enumerate(articles[:3], start=1): # print(f"Article {index}:\n{json.dumps(article, sort_keys=True, indent=4)}\n") # else: # print(f"Error: {response.status_code}") #API_KEY = '3805f6bbabcb42b3a0c08a489baf603d' #api_endpoint = f"https://newsapi.org/v2/top-headlines?country=us&category=business&apiKey={API_KEY}" API_URL = "https://cloud.cimar.co.uk/api/v3" STORAGE_API_URL = 'https://storelpu.cimar.co.uk' + "/api/v3/storage" class CimarAPI: def __init__(self, sid=None, username=None, password=None): self.sid = sid self.context = False self.username = username self.password = password # Default namespace to save studies to self.namespace = "eff0ca95-6ce6-443c-80dd-1136c8a58e4d" def get_storage(self, endpoint ): pass def post(self, endpoint, data=None, add_sid=True): if data is None: data = {} if add_sid: if self.sid is None: raise NoSession("No session ID") data["sid"] = self.sid print(f"{data=}") r = requests.post(API_URL + endpoint, data=data) if r.status_code == 412: match r.json(): case {"status": "ERROR", "error_type": "NOT_FOUND"}: raise NotFoundError("Study not found") case {"status": "ERROR", "error_type": error}: raise APIError(error) if r.status_code == 401: raise InvalidCredentials("Invalid credentials") print(r.status_code) print(r.content) return r.json() def login(self, username=None, password=None): if username is None: username = self.username if password is None: password = self.password data={"login": username, "password": password} r = requests.post(API_URL + "/session/login", data=data) if r.status_code == 412: raise InvalidCredentials() self.sid = r.json()["sid"] return self.sid def check_login_status(self): try: data = self.post("/session/user") except InvalidCredentials: return False return data["status"] == "OK" def logout(self): data = self.post("/session/logout") def __enter__(self): if self.sid is None: self.login() self.context = True return self def __exit__(self, exc_type, exc_val, exc_tb): if self.context: self.logout() self.sid = None def list_studies(self): return self.post("/study/list") def get_study_image_thumbnail(self, study_json): schema = self.get_study_schema(study_json) image = schema["series"][0]["images"][0] r = requests.get(STORAGE_API_URL + f"/study/{study_json["storage_namespace"]}/{study_json["study_uid"]}/image/{image["id"]}/version/{image["version"]}/frame/0/thumbnail", params={"sid": self.sid}) print(r.status_code) img = Image.open(io.BytesIO(r.content)) print(r.json()) # def get_series_image_thumbnail_link(self, study_json, series): # # middle_point = len(series["images"]) // 2 # # image = series["images"][middle_point] # # return requests.Request("GET",STORAGE_API_URL + f"/study/{study_json["storage_namespace"]}/{study_json["study_uid"]}/image/{image["id"]}/version/{image["version"]}/frame/0/thumbnail").prepare() # def get_diagnostic_image(self, study_json, image_id, version, frame): r = requests.get(STORAGE_API_URL + f"/study/{study_json["storage_namespace"]}/{study_json["study_uid"]}/image/{image_id}/version/{version}/frame/{frame}/diagnostic", params={"sid": self.sid}) print(r.status_code) img = Image.open(io.BytesIO(r.content)) print(r.json()) def get_study_by_uuid(self, uuid): return self.post("/study/get", data={"uuid": uuid}) def get_study_viewer_url_by_uuid(self, uuid): return self.get_study_by_uuid(uuid)["viewer_link"] def get_study_schema_by_uuid(self, uuid): study_json = self.get_study_by_uuid(uuid) return self.get_study_schema(study_json) def get_study_by_study_uid(self, study_uid): return self.post("/study/get", data={"study_uid": study_uid, "storage_namespace": self.namespace, "phi_namespace": self.namespace}) def get_study_schema(self, study_json): r = requests.get(STORAGE_API_URL + f"/study/{study_json["storage_namespace"]}/{study_json["study_uid"]}/schema", params={"sid": self.sid}) return r.json() def get_series_data(self, cimar_study_json): series_json = [] for series in cimar_study_json["series"]: instances = [] for image in series["images"]: p= requests.Request("GET", STORAGE_API_URL + f"/study/{cimar_study_json["namespace"]}/{cimar_study_json["study_uid"]}/image/{image["id"]}/version/{image["version"]}/frame/0/diagnostic", params={"sid": self.sid}).prepare() instances.append({"metadata": {}, "url": f"{p.url}"}) new_json = { "SeriesInstanceUID": series["series_uid"], "SeriesDescription": series["description"], "Modality": series["modality"], "instances": instances } series_json.append(new_json) return series_json def get_study_dicom_json(self, cimar_study_json): series_json = [] for series in cimar_study_json["series"]: instances = [] for image in series["images"]: p= requests.Request("GET", STORAGE_API_URL + f"/study/{cimar_study_json["namespace"]}/{cimar_study_json["study_uid"]}/image/{image["id"]}/version/{image["version"]}/frame/0/diagnostic", params={"sid": self.sid}).prepare() instances.append({"metadata": {}, "url": f"dicomweb:{p.url}"}) new_json = { "SeriesInstanceUID": series["series_uid"], "SeriesDescription": series["description"], "Modality": series["modality"], "instances": instances } series_json.append(new_json) studies_json = { "studies": [ { "StudyInstanceUID": "1.3.6.1.4.1.14519.5.2.1.6279.6001.298806137288633453246975630178", #"StudyDate": "20000101", "StudyTime": "", "PatientName": "Patient^Name", "PatientID": "LIDC-IDRI-0001", "AccessionNumber": "", "PatientAge": "", "PatientSex": "", "series": series_json, } ] } return studies_json def get_session_user(self): return self.post("/session/user") def upload_dicom(self, file, study_uid=""): if study_uid: study_uid = f"&study_uid={study_uid}" #data = {} #data["sid"] = self.sid #data["study_uid"] = "1.2.840.113773.2.10070.7173.20210208144916.37" #data["image_uid"]= "1.2.840.113773.4.10070.7173.20210208144916.39" files = open(file, "rb") #headers = {'content-type': 'application/dicom'} r = requests.post(STORAGE_API_URL + f"/namespace/{self.namespace}/image?sid={self.sid}{study_uid}", data=files)#, headers=headers) print(r) return r.json() #CimarAdminApi = CimarAPI().login("") if __name__ == "__main__": with CimarAPI(username="ross.kruger@nhs.net", password="") as api: print("***") #data = api.get_session_user() #data = api.upload_dicom() data = api.get_study_by_study_uid("2.1159655940009962407176401939062805636618") print(data) pass #print(api.sid) ##studies = api.list_studies() ##test_study = studies["studies"][0] ##schema = api.get_study_schema(test_study) #schema = api.get_study_schema_by_uuid("c9417b53-87aa-4c40-aca0-3fa34c225536") #api.get_study_dicom_json(schema) #print(schema) ##img = api.get_study_image_thumbnail(test_study) class NoSession(Exception): pass class InvalidCredentials(Exception): pass class NotFoundError(Exception): pass class APIError(Exception): pass