start integrating cimar
This commit is contained in:
Ross
@@ -0,0 +1,284 @@
|
||||
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
|
||||
Reference in New Issue
Block a user