Commit 6b6889cb authored by Gabriel Couture's avatar Gabriel Couture

Adding client, with tests

parent 2d5e86b4
Pipeline #629 failed with stages
......@@ -3,6 +3,7 @@
!.gitignore
!requirements.txt
!README.md
!setup.py
!brachy_dose_calculation_client/
!brachy_dose_calculation_client/**
......
from .client import BrachyDoseCalculationClient
__all__ = [
'BrachyDoseCalculationClient'
]
from typing import Tuple, Optional, Union
import pydicom
import requests
from pydicom import FileDataset
from pydicom.errors import InvalidDicomError
from pydicom.filebase import DicomBytesIO
from requests import Response
from .util import make_file_payload
class BrachyDoseCalculationClient:
def __init__(self, url: str) -> None:
self.url = url
def ping(self) -> Tuple[bool, Response]:
response = requests.get(self.url + '/ping')
return response.status_code == 200, response
def calculate_dose(
self,
rtstruct: Union[FileDataset, str],
rtplan: Union[FileDataset, str]) -> Tuple[Optional[FileDataset], Response]:
files = {'rtstruct': make_file_payload(rtstruct), 'rtplan': make_file_payload(rtplan)}
response = requests.post('http://localhost:8000/dose', files=files)
try:
dicom_data = DicomBytesIO(response.content)
rtdose = pydicom.dcmread(dicom_data)
return rtdose, response
except InvalidDicomError:
return None, response
import io
from typing import Dict, Any, Union
import pydicom
from pydicom.filebase import DicomFileLike
def make_file_payload(dataset_or_file_path: Union[pydicom.FileDataset, str]) -> Dict[str, Any]:
if type(dataset_or_file_path) == str:
file_path = dataset_or_file_path
return open(file_path, 'rb')
if type(dataset_or_file_path) == pydicom.FileDataset:
dataset = dataset_or_file_path
return get_dataset_raw_data(dataset)
raise TypeError('File path or pydicom.FileDataset should be given')
def get_dataset_raw_data(file_dataset: pydicom.FileDataset) -> bytes:
with io.BytesIO() as buffer:
"""Code from https://pydicom.github.io/pydicom/stable/auto_examples/memory_dataset.html"""
memory_dataset = DicomFileLike(buffer)
pydicom.dcmwrite(memory_dataset, file_dataset)
memory_dataset.seek(0)
return memory_dataset.read()
MICROSERVICE_URL = 'http://localhost:8000'
import http
import unittest
import pydicom
from brachy_dose_calculation_client import BrachyDoseCalculationClient
from tests.config import MICROSERVICE_URL
class TestCalculateDose(unittest.TestCase):
def setUp(self) -> None:
self.client = BrachyDoseCalculationClient(MICROSERVICE_URL)
def tearDown(self) -> None:
self.client = None
def test_givenRTStructAndRTPlanDatasets_whenCalculatingDose_thenResultIsRTDoseDatasetAndResponse(self):
rtstruct, rtplan = pydicom.dcmread('./tests/data/rtstruct.dcm'), pydicom.dcmread('./tests/data/rtplan.dcm')
result, response = self.client.calculate_dose(rtstruct, rtplan)
self.assertEqual(result.Modality, 'RTDOSE')
self.assertEqual(response.status_code, http.HTTPStatus.OK)
def test_givenRTStructAndRTPlanFilePath_whenCalculatingDose_thenResultIsRTDoseDatasetAndResponse(self):
rtstruct_filepath, rtplan_filepath = './tests/data/rtstruct.dcm', './tests/data/rtplan.dcm'
result, response = self.client.calculate_dose(rtstruct_filepath, rtplan_filepath)
self.assertEqual(result.Modality, 'RTDOSE')
self.assertEqual(response.status_code, http.HTTPStatus.OK)
def test_givenNonExistantFilePath_whenCalculatingDose_thenRaiseFileNotFoundError(self):
self.assertRaises(
FileNotFoundError,
lambda: self.client.calculate_dose('./non-existance/path.dcm', './non-existance/path.dcm')
)
def test_givenWrongDatasets_whenCalculatingDose_thenResultIsNoneAndNotAcceptableResponse(self):
rtstruct_1, rtstruct_2 = pydicom.dcmread('./tests/data/rtstruct.dcm'), pydicom.dcmread('./tests/data/rtstruct.dcm')
result, response = self.client.calculate_dose(rtstruct_1, rtstruct_2)
self.assertIsNone(result)
self.assertEqual(response.status_code, http.HTTPStatus.NOT_ACCEPTABLE)
import unittest
from brachy_dose_calculation_client import BrachyDoseCalculationClient
from .config import MICROSERVICE_URL
class TestPing(unittest.TestCase):
def setUp(self) -> None:
self.client = BrachyDoseCalculationClient(MICROSERVICE_URL)
def tearDown(self) -> None:
self.client = None
def test_whenPing_thenResultIsTrueAndResponse(self):
result, response = self.client.ping()
self.assertTrue(result)
self.assertEqual(response.text, '')
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment