From 7812e803c979e0549bfc99d56e612c9a3be1218d Mon Sep 17 00:00:00 2001 From: Ross Date: Fri, 22 Dec 2023 16:08:54 +0000 Subject: [PATCH] start repo --- .gitignore | 4 + anon.py | 34 ++ anon2/anon_gui.spec | 44 ++ anon_gui.py | 1130 +++++++++++++++++++++++++++++++++++++++++++ anon_gui.spec | 46 ++ client.py | 65 +++ reslice.py | 64 +++ setup.py | 16 + sortList.py | 43 ++ uploader.py | 193 ++++++++ wxDicomViewer.py | 384 +++++++++++++++ 11 files changed, 2023 insertions(+) create mode 100644 .gitignore create mode 100644 anon.py create mode 100644 anon2/anon_gui.spec create mode 100644 anon_gui.py create mode 100644 anon_gui.spec create mode 100644 client.py create mode 100644 reslice.py create mode 100644 setup.py create mode 100644 sortList.py create mode 100644 uploader.py create mode 100644 wxDicomViewer.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..15f1ea8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.env/* +.venv/* + +.pyc \ No newline at end of file diff --git a/anon.py b/anon.py new file mode 100644 index 0000000..89e0cf9 --- /dev/null +++ b/anon.py @@ -0,0 +1,34 @@ +import os + +from pathlib import Path + +import pydicom +import dicognito.anonymizer + +import typer + + +anonymizer = dicognito.anonymizer.Anonymizer() + +app = typer.Typer() + + +def anonymize_file(filepath: Path, new_filepath: Path): + with pydicom.dcmread(filepath) as dataset: + anonymizer.anonymize(dataset) + dataset.save_as(new_filepath) + return new_filepath + +@app.command() +def annonymise(filepath: Path, output_dir: Path): + + new_filepath : Path + # Get next available filename + i : int = 0 + while os.path.exists(new_filepath := Path(output_dir).joinpath(f"IMG_{i:03}.dcm")): + i += 1 + + anonymize_file(filepath, new_filepath) + +if __name__ == "__main__": + app() diff --git a/anon2/anon_gui.spec b/anon2/anon_gui.spec new file mode 100644 index 0000000..a95eb65 --- /dev/null +++ b/anon2/anon_gui.spec @@ -0,0 +1,44 @@ +# -*- mode: python ; coding: utf-8 -*- + + +block_cipher = None + + +a = Analysis( + ['anon_gui.py'], + pathex=[], + binaries=[], + datas=[], + hiddenimports=[], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + win_no_prefer_redirects=False, + win_private_assemblies=False, + cipher=block_cipher, + noarchive=False, +) +pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) + +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.zipfiles, + a.datas, + [], + name='anon_gui', + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + upx_exclude=[], + runtime_tmpdir=None, + console=False, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +) diff --git a/anon_gui.py b/anon_gui.py new file mode 100644 index 0000000..71def97 --- /dev/null +++ b/anon_gui.py @@ -0,0 +1,1130 @@ +from collections import defaultdict +import os +import threading +from typing import Iterable +from bs4 import BeautifulSoup +import requests + +# from blake3 import blake3 + +import zmq +from blake3 import blake3 + +# import time +# +# import sys +# +# from datauri import DataURI + +from pathlib import Path + +import pydicom +import pydicom.errors +import pydicom.encoders.gdcm +import pydicom.encoders.pylibjpeg +import dicognito.anonymizer +from six import create_bound_method + +import typer + +import time +import sys +import tempfile + +# from typing import List +import wx +import wx.adv + + +from wxDicomViewer import BasicDicomViewer, ViewerFrame + +# from watchdog.observers import Observer + +# from watchdog.events import LoggingEventHandler +# from watchdog.events import PatternMatchingEventHandler + +import hashlib + +# import pywintypes +# import win32api +# import win32com + + +try: + from agw import hyperlink as hl +except ImportError: # if it's not there locally, try the wxPython lib. + import wx.lib.agw.hyperlink as hl + + +import wx.lib.agw.thumbnailctrl as TC +import wx.lib.agw.ultimatelistctrl as ULC + + +from wx.lib.agw.scrolledthumbnail import ( + EVT_THUMBNAILS_SEL_CHANGED, + EVT_THUMBNAILS_POINTED, + EVT_THUMBNAILS_DCLICK, +) + +import wx.lib.mixins.listctrl as listmix + +import time +import datetime +import contextlib + +# import watchdog.observers + +anonymizer = dicognito.anonymizer.Anonymizer() + +# determine if application is a script file or frozen exe +if getattr(sys, "frozen", False): + application_path = os.path.dirname(sys.executable) +elif __file__: + application_path = os.path.dirname(__file__) + +from loguru import logger + +try: + logger.add( + sys.stderr, format="{time} {level} {message}", filter="my_module", level="DEBUG" + ) +except TypeError: + pass +try: + logger.add(os.path.join(application_path, "anon_gui_{time}.log")) +except TypeError: + pass + +logger.debug("Starting") + +# import logging +# logging.basicConfig(filename=os.path.join(sys.path[0], f'anon_gui_{datetime.datetime.now().strftime("%Y-%m-%d--%H-%M-%S")}.log'), encoding='utf-8', level=logging.DEBUG) + + +def get_image_hash(img) -> (str, bool): + is_dicom = False + # Try and read the file as a dicom + try: + # and generate a hash from the pixel data + # TODO: improve? + dataset = pydicom.dcmread(img) + # flatten = dataset.pixel_array.astype(str).flatten() + # print("flatteded") + # pre_join = ",".join(flatten) + # print(pre_join) + # hash = hashlib.md5(pre_join.encode()).hexdigest() + # ---- + md5 = hashlib.md5() + first = True + for i in dataset.pixel_array.astype(str).flatten(): + if first: + first = False + md5.update(f"{i}".encode()) + else: + md5.update(f",{i}".encode()) + + hash = md5.hexdigest() + is_dicom = True + # ---- + + except pydicom.errors.InvalidDicomError: + try: # This is horrible (but needed for current unit tests) + # (we use a temporary file that breaks here) + img.file.open() + hash = hashlib.md5(img.read()).hexdigest() + except AttributeError: + return "12345ABCD", False + + return hash, is_dicom + + +app = typer.Typer(add_completion=False) + + +@app.command() +@logger.catch() +def annonymise( + path: Path, output_dir: Path, run_gui: bool = True, file_list: bool = False +): + logger.debug(locals()) + if run_gui: # Let the GUI handle it all + launchGUI( + input_dir=path, + output_dir=output_dir, + silent_fail=True, + clear_files=False, + clear_files_on_close=True, + dev=dev, + ) + return + if path.is_dir(): + files = list(Path(path).glob("*.dcm")) + elif path.is_file(): + if file_list: + with open(path) as f: + files = f.read().splitlines() + # files = [path] + else: + files = [path] + else: + typer.echo("Invalid path") + return + + # progress_dialog = wx.ProgressDialog("Annonymise files", "Removing data...", maximum=len(files), style=wx.PD_APP_MODAL|wx.PD_ELAPSED_TIME|wx.PD_ESTIMATED_TIME) + for n, file in enumerate(files): + # progress_dialog.Update(n) + new_filepath: Path + # Get next available filename + i: int = 0 + while os.path.exists(new_filepath := output_dir.joinpath(f"IMG_{i:03}.dcm")): + i += 1 + # new_filepath = Path(output_dir).joinpath(f"IMG_{i:03}.dcm") + + anonymize_file(file, new_filepath) + + # progress_dialog.Destroy() + + return new_filepath + + +def anonymize_file(file: Path, new_filepath: Path): + logger.debug(f"Annonymise file, in={file}, out={new_filepath}") + # Ensure the output folder exists + new_filepath.parents[0].mkdir(parents=True, exist_ok=True) + # was this on windows + # new_filepath.mkdir(parents=True, exist_ok=True) + # with new_filepath.open() as f: + # f.write("", "wb") + # new_filepath.touch() + with pydicom.dcmread(file) as dataset: + anonymizer.anonymize(dataset) + dataset.save_as(new_filepath) + return new_filepath + + +def order_files_by_dicom_attribute( + paths: Iterable[Path], dicom_attribute: str = "SliceLocation" +) -> Iterable[Path]: + files = [] + + for path in paths: + dataset = pydicom.dcmread(path) + files.append((path, dataset)) + + try: + sorted_datasets = sorted( + files, key=lambda f: f[1][dicom_attribute].value, reverse=True + ) + except KeyError: # Happens if no slicelocation + return paths + return [d[0] for d in sorted_datasets] + + +def get_dicom_pixel_md5_hash(file: Path): + try: + with pydicom.dcmread(file) as dataset: + a = ",".join([str(i) for i in dataset.pixel_array.flatten()]) + + return hashlib.md5(a.encode()).hexdigest() + except pydicom.errors.InvalidDicomError: + logger.debug(f"Invalid dicom: {file}") + with open(file) as f: + return hashlib.md5(f.read()).hexdigest() + + +def get_file_hash(file: Path, hasher="blake3"): + match hasher: + case "blake3": + file_hash = blake3() + case "md5": + file_hash = hashlib.md5() + case _: + raise Exception(f"{_} is not a valid hasher") + + with open(file, "rb") as f: + while chunk := f.read(8192): + file_hash.update(chunk) + + return file_hash.hexdigest() + + with open(file) as f: + return hashlib.md5(f.read()).hexdigest() + + +# def get_dicom_blake_hash(file: Path)@ + + +class LoginDialog(wx.Dialog): + """ + Class to define login dialog + """ + + # ---------------------------------------------------------------------- + def __init__(self, parent): + """Constructor""" + wx.Dialog.__init__(self, None, title="Login") + + self.parent = parent + + # user info + user_sizer = wx.BoxSizer(wx.HORIZONTAL) + + user_lbl = wx.StaticText(self, label="Username:") + user_sizer.Add(user_lbl, 0, wx.ALL | wx.CENTER, 5) + self.user = wx.TextCtrl(self) + user_sizer.Add(self.user, 0, wx.ALL, 5) + + # pass info + p_sizer = wx.BoxSizer(wx.HORIZONTAL) + + p_lbl = wx.StaticText(self, label="Password:") + p_sizer.Add(p_lbl, 0, wx.ALL | wx.CENTER, 5) + self.password = wx.TextCtrl(self, style=wx.TE_PASSWORD | wx.TE_PROCESS_ENTER) + self.password.Bind(wx.EVT_TEXT_ENTER, self.on_login) + p_sizer.Add(self.password, 0, wx.ALL, 5) + + main_sizer = wx.BoxSizer(wx.VERTICAL) + main_sizer.Add(user_sizer, 0, wx.ALL, 5) + main_sizer.Add(p_sizer, 0, wx.ALL, 5) + + btn = wx.Button(self, label="Login") + btn.Bind(wx.EVT_BUTTON, self.on_login) + main_sizer.Add(btn, 0, wx.ALL | wx.CENTER, 5) + + self.main_sizer = main_sizer + + self.error_text = wx.StaticText(self, label="Error logging in.") + + self.error_text.SetForegroundColour(wx.RED) + self.error_text.Hide() + + self.main_sizer.Add(self.error_text) + + self.SetSizer(main_sizer) + + # ---------------------------------------------------------------------- + def on_login(self, event): + """ + Check credentials and login + """ + LOGIN_URL = f"{self.parent.base_site_url}accounts/login/" + + user = self.user.GetValue() + password = self.password.GetValue() + + rqst = requests.session() + rsp = rqst.get(LOGIN_URL) + + token = ( + BeautifulSoup(rsp.content, "html.parser").find("input").attrs["value"] + ) # , attr={"name": "csrfmiddlewaretoken"}).attrs("value") + + # token = rsp.cookies["csrftoken"] + # header = {"X-CSRFToken": token} + # cookies = {"csrftoken": token} + + data = { + "username": user, + "password": password, + "csrfmiddlewaretoken": token, + "next": "/", + } + + print(data) + rqst.headers["Referer"] = LOGIN_URL + + rsp = rqst.post( + LOGIN_URL, + data=data, + # headers=header, + # cookies=cookies + ) + + soup = BeautifulSoup(rsp.content, "html.parser") + + if soup.find("button") and soup.find("button").find(string="Login"): + print("login fail") + pass + else: + print("login success") + self.parent.login_success = True + # print(rqst.headers) + + rqst.headers["X-CSRFToken"] = token + self.parent.rqst = rqst + + print(rqst.headers) + print(rqst.cookies) + new = rqst.get(f"{self.parent.base_site_url}accounts/api/atlas/get_cases_user") + + print("CASESE", new) + print(new.content) + + if self.parent.login_success: + self.parent.upload_button.Show() + self.parent.login_button.Hide() + self.parent.panel.Layout() + self.Destroy() + + # self.parent.login_button.SetBackgroundColour(wx.BLUE) + # self.parent.login_button.SetForegroundColour(wx.WHITE) + else: + self.login_fail() + + def login_fail(self): + self.error_text.Show() + self.Layout() + pass + + +class SortableList(ULC.UltimateListCtrl, listmix.ColumnSorterMixin): + def __init__(self, parent, col_number, DATA={}): + ULC.UltimateListCtrl.__init__( + self, + parent, + wx.ID_ANY, + agwStyle=ULC.ULC_REPORT | ULC.ULC_VRULES | ULC.ULC_HRULES + # | ULC.ULC_SINGLE_SEL + | ULC.ULC_HAS_VARIABLE_ROW_HEIGHT, + ) + self.itemDataMap = DATA + listmix.ColumnSorterMixin.__init__(self, col_number) + self.Bind(wx.EVT_LIST_COL_CLICK, self.OnColumn) + + def OnColumn(self, e): + self.Refresh() + e.Skip() + + def GetListCtrl(self): + return self + + def UpdateData(self, DATA): + self.itemDataMap = DATA + + +class SeriesGrid(wx.Panel): + def __init__(self, parent): + wx.Panel.__init__(self, parent, size=(-1, 200), style=wx.SUNKEN_BORDER) + + self.parent = parent + + self.main_sizer = wx.BoxSizer(wx.VERTICAL) + + self.series_sizer = wx.GridSizer(wx.HORIZONTAL) + self.series_sizer.SetVGap(40) + + self.text = wx.StaticText(self, label="SERIES TO UPLOAD") + self.main_sizer.Add(self.text) + + self.main_sizer.Add(self.series_sizer) + + self.SetSizerAndFit(self.main_sizer) + + self.series = [] + + def display_series(self): + self.series_sizer.Clear(True) + + for name, number in self.series: + logger.debug(f"add series: {name}") + panel = SeriesPanel(self, name, number) + + self.series_sizer.Add(panel) + + self.Layout() + + +class SeriesPanel(wx.Panel): + def __init__(self, parent, name, number): + wx.Panel.__init__(self, parent, size=(-1, 200), style=wx.SUNKEN_BORDER) + + self.parent = parent + + self.SetBackgroundColour(wx.BLUE) + self.SetForegroundColour(wx.WHITE) + + sizer = wx.BoxSizer(wx.VERTICAL) + + name = wx.StaticText(self, label=name) + number = wx.StaticText( + self, label=f"[{number}]", style=wx.ALIGN_CENTER_HORIZONTAL + ) + sizer.Add(name) + sizer.Add(number) + + self.SetSizerAndFit(sizer) + + +class AnonGui(wx.App): + def __init__( + self, + redirect=False, + filename=None, + input_dir="", + output_dir="", + silent_fail=False, + clear_files=True, + clear_files_on_close=True, + dev=False, + ): + self.dir = output_dir + self.input_dir = input_dir + + logger.debug(f"Input dir: {input_dir}") + logger.debug(f"Output dir: {output_dir}") + + self.base_site_url = "https://www.penracourses.org.uk/" + + if dev: + self.base_site_url = "http://localhost:8000/" + + self.login_success = False + + self.files = {} + self.series_map = {} + self.series_map_details = {} + + self.active = () + + self.silent_fail = silent_fail + self.clear_files_on_start = clear_files + self.clear_files_on_close = clear_files_on_close + + self.rqst = None + + wx.App.__init__(self, redirect, filename) + + def my_listener(self, message, arg2=None): + """ + Listener function + """ + logger.debug(f"Received the following message: {message}") + if arg2: + logger.debug(f"Received another arguments: {arg2}") + + def OnInit(self): + self.name = "SingleApp-%s" % wx.GetUserId() + self.instance = wx.SingleInstanceChecker(self.name) + + if self.instance.IsAnotherRunning(): + logger.debug("Another instance detected") + context = zmq.Context() + socket = context.socket(zmq.PUSH) + socket.connect("tcp://127.0.0.1:5556") + socket.send_string("loaded") + logger.debug("Trigger refresh") + + time.sleep(5) + if not self.silent_fail: + wx.MessageBox("Another instance is running", "ERROR") + return False + + def zmq_listener(self): + context = zmq.Context() + self.socket = context.socket(zmq.PULL) + self.socket.bind("tcp://127.0.0.1:5556") + logger.debug("idle") + while True: + msg = self.socket.recv() + # if msg == 'zeromq': + logger.debug(f"{msg=}") + if msg == b"loaded": + logger.debug("loaded1") + wx.CallAfter(self.AnnonymiseFilesInDir, self.input_dir, self.dir) + + self.zmq_thread = threading.Thread( + target=zmq_listener, args=(self,), daemon=True + ) + self.zmq_thread.start() + + self.frame = wx.Frame( + None, + wx.ID_ANY, + title="Uploader", + style=wx.DEFAULT_FRAME_STYLE | wx.STAY_ON_TOP, + size=(300, 300), + ) + + self.panel = wx.Panel(self.frame, wx.ID_ANY) + + self.file_detail_map = {} + + self.list_ctrl = SortableList( + self.panel, + 6, + ) + self.series_grid = SeriesGrid(self.panel) + + if self.clear_files_on_start: + self.ClearFiles() + + self.panel.sizer = wx.BoxSizer(wx.VERTICAL) + + self.link_rapid_upload = hl.HyperLinkCtrl( + self.panel, + wx.ID_ANY, + "Rapid Uploader", + URL=f"{self.base_site_url}rapids/create/", + ) + # self.panel.sizer.Add(self.link_rapid_upload, 0, wx.ALL, 10) + + self.link_long_upload = hl.HyperLinkCtrl( + self.panel, + wx.ID_ANY, + "Long Series", + URL=f"{self.base_site_url}longs/create/series", + ) + # self.panel.sizer.Add(self.link_long_upload, 0, wx.ALL, 10) + hoz_links = wx.BoxSizer(wx.HORIZONTAL) + hoz_links.Add(self.link_rapid_upload, 1, wx.EXPAND) + hoz_links.Add(self.link_long_upload, 1, wx.EXPAND) + self.panel.sizer.Add(hoz_links, 1, wx.EXPAND) + + # self.dirctrl = wx.GenericDirCtrl(self.panel, wx.ID_ANY, + # dir=dir, + # style=wx.DIRCTRL_SHOW_FILTERS | + # wx.DIRCTRL_3D_INTERNAL | + # wx.DIRCTRL_MULTIPLE, + # filter="Dicom files (*.dcm)|*.dcm") + self.drag_button = wx.Button( + self.panel, id=wx.ID_ANY, label="Click here to drag files" + ) + self.upload_button = wx.Button( + self.panel, id=wx.ID_ANY, label="Click here to upload files" + ) + self.upload_button.Hide() + + # Use some sizers to see layout options + # self.panel.sizer.Add(self.dirctrl, 1, wx.EXPAND) + + # self.Bind(wx.EVT_BUTTON, self.OnDragInit, self.drag_button) + self.Bind(wx.EVT_LEFT_DOWN, self.OnDragInit, self.drag_button) + self.Bind(wx.EVT_LEFT_DOWN, self.OnFileUpload, self.upload_button) + self.Bind(wx.EVT_LIST_COL_CLICK, self.OnColumnClick) + + # self.list_ctrl = wx.ListCtrl(self.panel, + # style=wx.LC_REPORT + # |wx.BORDER_SUNKEN + # ) + + self.Bind(ULC.EVT_LIST_ITEM_RIGHT_CLICK, self.ShowPopupMenu, self.list_ctrl) + + self.list_ctrl.InsertColumn(0, "Filename") + self.list_ctrl.InsertColumn(1, "Examination") + self.list_ctrl.InsertColumn(2, "Instance Number") + self.list_ctrl.InsertColumn(3, "Name") + self.list_ctrl.InsertColumn(4, "Series Description") + self.list_ctrl.InsertColumn(5, "Slice location") + + self.AnnonymiseFilesInDir(self.input_dir, self.dir) + + # self.UpdateListCtrl() + + self.login_button = wx.Button(self.panel, id=wx.ID_ANY, label="Login") + self.login_button.SetForegroundColour(wx.BLUE) + self.clear_files_button = wx.Button( + self.panel, id=wx.ID_ANY, label="Clear files" + ) + self.refresh_files_button = wx.Button(self.panel, id=wx.ID_ANY, label="Refresh") + self.reorder_files_button = wx.Button(self.panel, id=wx.ID_ANY, label="Reorder") + self.reannonymise_files_button = wx.Button( + self.panel, id=wx.ID_ANY, label="Reannonymise" + ) + self.manual_load_button = wx.Button(self.panel, id=wx.ID_ANY, label="Load") + + # wx.Bind(self.clear_files_button, ) + self.Bind(wx.EVT_BUTTON, self.Login, self.login_button) + self.Bind(wx.EVT_BUTTON, self.ClearFiles, self.clear_files_button) + self.Bind(wx.EVT_BUTTON, lambda evt: self.LoadDir(), self.refresh_files_button) + self.Bind(wx.EVT_BUTTON, self.ReorderFiles, self.reorder_files_button) + self.Bind(wx.EVT_BUTTON, self.ReAnnonymiseFiles, self.reannonymise_files_button) + self.Bind( + wx.EVT_BUTTON, + lambda evt: self.AnnonymiseFilesInDir(self.input_dir, self.dir), + self.manual_load_button, + ) + + # Use some sizers to see layout options + # self.panel.sizer.Add(self.dirctrl, 1, wx.EXPAND) + + hoz = wx.BoxSizer(wx.HORIZONTAL) + hoz.Add(self.login_button, 1, wx.EXPAND) + hoz.Add(self.clear_files_button, 1, wx.EXPAND) + hoz.Add(self.refresh_files_button, 1, wx.EXPAND) + hoz.Add(self.reorder_files_button, 1, wx.EXPAND) + hoz.Add(self.reannonymise_files_button, 1, wx.EXPAND) + hoz.Add(self.manual_load_button, 1, wx.EXPAND) + + self.panel.sizer.Add(hoz, 1, wx.EXPAND) + + self.panel.sizer.Add(self.series_grid, 2, wx.EXPAND) + self.panel.sizer.Add(self.list_ctrl, 8, wx.EXPAND) + self.panel.sizer.Add(self.drag_button, 2, wx.EXPAND) + self.panel.sizer.Add(self.upload_button, 2, wx.EXPAND) + + + # Layout sizers + self.panel.SetSizer(self.panel.sizer) + self.panel.SetAutoLayout(1) + self.panel.sizer.Fit(self.panel) + self.frame.Show() + # self.spinner.Hide() + + self.Bind(wx.EVT_KEY_DOWN, self.onKeyPress) + + return super().OnInit() + + def populate_series_grid(self): + self.series_grid.series = [] + + for series in self.series_map: + logger.debug(f"Create series: {series}") + self.series_grid.series.append( + ( + self.series_map_details[series]["series_description"], + len(self.series_map[series]), + ) + ) + + self.series_grid.display_series() + + def AnnonymiseFilesInDir(self, input_dir, output_dir): + logger.debug("AnnonymiseFilesInDir") + logger.debug(f"input: {input_dir} / output {output_dir}") + if input_dir.is_dir(): + logger.debug("is dir") + files = list(Path(input_dir).rglob("*.dcm")) + else: + return + + logger.debug(files) + + files = order_files_by_dicom_attribute(files) + + progress_dialog = wx.ProgressDialog( + "Annonymise files", + "Removing data...", + maximum=len(files), + style=wx.PD_APP_MODAL | wx.PD_ELAPSED_TIME | wx.PD_ESTIMATED_TIME, + ) + + for n, file in enumerate(files): + progress_dialog.Update(n) + new_filepath: Path + # Get next available filename + i: int = 0 + while os.path.exists( + new_filepath := output_dir.joinpath(f"IMG_{i:03}.dcm") + ): + i += 1 + # new_filepath = Path(output_dir).joinpath(f"IMG_{i:03}.dcm") + + anonymize_file(file, new_filepath) + progress_dialog.Destroy() + + self.LoadDir(self.dir) + + def onKeyPress(self, event): + keycode = event.GetKeyCode() + logger.debug(keycode) + if keycode == wx.WXK_F1: + logger.debug("you pressed the F1!") + self.OnAboutBox() + return + event.Skip() + + def OnAboutBox(self, event=None): + description = f""" + An upload/export tool + + Settings + -------- + Watch dir: {self.dir} +""" + + licence = """GPL""" + + info = wx.adv.AboutDialogInfo() + + # info.SetIcon(wx.Icon('hunter.png', wx.BITMAP_TYPE_PNG)) + info.SetName("RadUploader") + info.SetVersion("0.1") + info.SetDescription(description) + + info.SetLicence(licence) + + wx.adv.AboutBox(info) + + def OnExit(self): + if self.clear_files_on_close: + self.ClearFiles() + + # self.zmq_thread.join() + + return super().OnExit() + + def LoadDir(self, dir=None, refresh=True, clear_file_detail_map=False): + if clear_file_detail_map: + self.file_detail_map = {} + if dir is None: + dir = self.dir + if refresh: + self.files = {} + self.series_map = {} + self.series_map_details = {} + for path in Path(dir).glob("*.dcm"): + self.AddFile(path) + + self.UpdateListCtrl() + + def AddFile(self, path): + hash = get_file_hash(path) + logger.debug("Hash", hash) + + if hash not in self.files.values(): + self.files[path] = hash + # Remove duplicate files + else: + os.remove(path) + + def RemoveFile(self, path): + del self.files[path] + + def OnColumnClick(self, event): + self.list_ctrl.Refresh() + event.Skip() + + def ShowPopupMenu(self, event): + if self.list_ctrl.GetSelectedItemCount(): + pass + else: + ind = event.GetIndex() + if ind > -1: + self.list_ctrl.Select(ind) + else: + return + menu = wx.Menu() + menu.Append(1, "Delete selected items") + menu.Bind(wx.EVT_MENU, self.DeleteItems, id=1) + menu.Append(2, "View Dicom") + menu.Bind(wx.EVT_MENU, self.ViewDicom, id=2) + self.frame.PopupMenu(menu) + + def ViewDicom(self, event): + selectedItems = [] + for i in range(self.list_ctrl.GetItemCount()): + if self.list_ctrl.IsSelected(i): + #path = self.list_ctrl.GetItemPyData(i["file"]) + path = self.list_ctrl.GetItemPyData(i)["file"] + print(path) + + viewer = ViewerFrame(None, "Viewer", path) + # viewer.MainLoop() + break + + def ReAnnonymiseFiles(self, event): + global anonymizer + anonymizer = dicognito.anonymizer.Anonymizer() + paths = [] + + # self.file_observer.pause() + + tempdir = Path(tempfile.gettempdir()) + to_move = [] + for i in range(self.list_ctrl.GetItemCount()): + path = self.list_ctrl.GetItemPyData(i["file"]) + logger.debug("--", path) + # paths.append(path) + new_path = annonymise(path, tempdir, run_gui=False) + os.remove(path) + to_move.append((new_path, Path(self.dir, new_path.name))) + + for path, new_path in to_move: + logger.debug("**", path, new_path) + os.rename(path, new_path) + + self.LoadDir(self.dir, clear_file_detail_map=True) + + def ReorderFiles(self, event): + paths = [] + + # self.file_observer.pause() + + tempdir = Path(tempfile.gettempdir()) + to_move = [] + for i in range(self.list_ctrl.GetItemCount()): + path = self.list_ctrl.GetItemPyData(i["file"]) + logger.debug("--", path) + # paths.append(path) + + os.replace(path, new_path := tempdir.joinpath(f"IMG_{i:03}.dcm")) + # os.remove(path) + to_move.append((new_path, Path(self.dir, new_path.name))) + i += 1 + + for path, new_path in to_move: + logger.debug("**", path, new_path) + os.rename(path, new_path) + + self.LoadDir(self.dir) + # self.file_observer.resume() + + def DeleteItems(self, event): + selectedItems = [] + for i in range(self.list_ctrl.GetItemCount()): + if self.list_ctrl.IsSelected(i): + selectedItems.append(self.list_ctrl.GetItemPyData(i["file"])) + + logger.debug(selectedItems) + for f in selectedItems: + self.RemoveFile(f) + os.remove(f) + + wx.CallAfter(self.UpdateListCtrl) + + def Login(self, event=None): + dlg = LoginDialog(self) + + dlg.ShowModal() + + def ClearFiles(self, event=None): + files = Path(self.dir).glob("*") + for f in files: + os.remove(f) + self.LoadDir(clear_file_detail_map=True) + + def UpdateListCtrl(self, event=None): + logger.debug("update list ctrl") + self.list_ctrl.DeleteAllItems() + n = 0 + DATA = {} + + progress_dialog = wx.ProgressDialog( + "Loading files", + "Extracting data from files", + maximum=len(self.files), + style=wx.PD_APP_MODAL | wx.PD_ELAPSED_TIME | wx.PD_ESTIMATED_TIME, + ) + + self.series_map = defaultdict(list) + self.series_map_details = defaultdict(dict) + + for n, f in enumerate(self.files): + progress_dialog.Update(n) + # dataset = pydicom.read_file(f) + + if f in self.file_detail_map: + d, hash, d2 = self.file_detail_map[f] + else: + try: + dataset = pydicom.dcmread(f) + except FileNotFoundError: + return + + try: + study_desc = dataset.StudyDescription + except: + study_desc = "..." + try: + series_desc = dataset.SeriesDescription + except: + series_desc = "..." + try: + instance_number = f"{dataset.InstanceNumber:03}" + except: + instance_number = "..." + try: + patient_name = str(dataset.PatientName) + except: + patient_name = "..." + try: + slice_location = str(dataset.SliceLocation) + except: + slice_location = "..." + + d = ( + f.name, + study_desc, + instance_number, + patient_name, + series_desc, + slice_location, + ) + + series_instances_uid = str(dataset.SeriesInstanceUID) + d2 = { + "series_instances_uid": series_instances_uid, + "series_description": series_desc, + "study_description": study_desc, + "name": patient_name, + } + + #md5 = hashlib.md5() + #first = True + #for i in dataset.pixel_array.astype(str).flatten(): + # if first: + # first = False + # md5.update(f"{i}".encode()) + # else: + # md5.update(f",{i}".encode()) + + #hash = md5.hexdigest() + + hash = None # If we don't use the hash... + self.file_detail_map[f] = d, hash, d2 + + series_instances_uid = d2["series_instances_uid"] + try: + self.series_map[series_instances_uid].append(f) + self.series_map_details[series_instances_uid][ + "series_description" + ] = d2["series_description"] + self.series_map_details[series_instances_uid]["study_description"] = d2[ + "study_description" + ] + self.series_map_details[series_instances_uid]["name"] = d2["name"] + except: + logger.debug(f"Unable to get Series Instance UID: {f.name}") + + logger.debug(d) + DATA[n] = d + self.list_ctrl.Append(d) + self.list_ctrl.SetItemPyData(n, {"file": f, "hash": hash}) + self.list_ctrl.SetItemData(n, n) + n = n + 1 + self.list_ctrl.UpdateData(DATA) + self.populate_series_grid() + + progress_dialog.Destroy() + + def OnColRightClick(self, event): + item = self.list_ctrl.GetColumn(event.GetColumn()) + logger.debug(item) + + def OnFileUpload(self, event): + # self.spinner.Show() + wx.CallAfter(self.upload) + + def upload(self): + files_to_upload = [] + logger.debug("Upload files") + for f in self.files: + files_to_upload.append(("files", open(str(f), "rb"))) + + # chunck files + n = 10 + chunked_files = [files_to_upload[i:i + n] for i in range(0, len(files_to_upload), n)] + + progress_dialog = wx.ProgressDialog( + "Uploading files", + "Uploading files. This is done in batches...", + maximum=len(chunked_files), + style=wx.PD_ELAPSED_TIME | wx.PD_ESTIMATED_TIME, + ) + + upload_file_list = [] + duplicate_file_list = [] + + for n, files in enumerate(chunked_files): + + def upload_files(files): + self.rqst.headers["X-CSRFToken"] = self.rqst.cookies["csrftoken"] + #print(self.rqst.headers) + #print(self.rqst.cookies) + resp = self.rqst.post( + f"{self.base_site_url}api/atlas/upload_dicom", + # data=data, + files=files, + ) + + logger.debug(f"Resp: {resp}") + logger.debug(f"{resp.content}") + return resp + + + + progress_dialog.Update(n, f"Uploading batch {n}/{len(chunked_files)}") + + logger.debug(f"n: {n}") + # try to upload the files + resp = upload_files(files) + if resp.status_code == 200: + upload_file_list.extend(resp.json()["uploaded"]) + duplicate_file_list.extend(resp.json()["duplicates"]) + + + continue + + logger.debug(f"n: {n} fail") + # try again if we fail + resp = upload_files(files) + if resp.status_code == 200: + upload_file_list.extend(resp.json()["uploaded"]) + duplicate_file_list.extend(resp.json()["duplicates"]) + continue + + progress_dialog.Destroy() + + print(upload_file_list) + print(duplicate_file_list) + + wx.CallAfter(self.upload_complete_message, upload_file_list, duplicate_file_list) + + + def upload_complete_message(self, upload_file_list, duplicate_file_list): + wx.MessageBox(f"Files uploaded: {len(upload_file_list)}\nDuplicate files (not uploaded): {len(duplicate_file_list)}") + + def OnDragInit(self, event): + my_data = wx.FileDataObject() + # files = pathlib.Path(self.dir).glob("*.dcm") + for f in self.files: + my_data.AddFile(str(f)) + + dragSource = wx.DropSource(self.panel) + dragSource.SetData(my_data) + + result = dragSource.DoDragDrop(True) + + +# --------------------------------------------------------------------------- + + +@app.command() +@app.callback(invoke_without_command=True) +def runGUI( + ctx: typer.Context, + input_dir: Path = "C:\\Temp\\", + # input_dir: Path = "/home/ross/temp/in/", + output_dir: Path = "C:\\temp_upload\\", + # output_dir: Path = "/home/ross/temp/out/", + silent_fail: bool = False, + clear_files: bool = True, + clear_files_on_close: bool = True, + dev: bool = False, +): + if ctx.invoked_subcommand is None: + try: + typer.echo("run gui") + except AttributeError: # No console + pass + launchGUI( + input_dir, output_dir, silent_fail, clear_files, clear_files_on_close, dev + ) + + +def launchGUI( + input_dir, output_dir, silent_fail, clear_files, clear_files_on_close, dev +): + app = AnonGui( + input_dir=input_dir, + output_dir=output_dir, + silent_fail=silent_fail, + clear_files=clear_files, + clear_files_on_close=clear_files_on_close, + dev=dev, + ) + app.MainLoop() + + +if __name__ == "__main__": + app() diff --git a/anon_gui.spec b/anon_gui.spec new file mode 100644 index 0000000..4891402 --- /dev/null +++ b/anon_gui.spec @@ -0,0 +1,46 @@ +# -*- mode: python ; coding: utf-8 -*- + + +block_cipher = None + + +a = Analysis( + ['anon_gui.py'], + pathex=[], + #binaries=[(r"C:\Users\krugerro\Desktop\rad_tools\.venv\Lib\site-packages\pywin32_system32\pythoncom311.dll", "."), + #(r"C:\Users\krugerro\Desktop\rad_tools\.venv\Lib\site-packages\pywin32_system32\pywintypes311.dll", "."), + #], + datas=[], + hiddenimports=[], + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[], + win_no_prefer_redirects=False, + win_private_assemblies=False, + cipher=block_cipher, + noarchive=False, +) +pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) + +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.zipfiles, + a.datas, + [], + name='anon_gui', + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, + upx_exclude=[], + runtime_tmpdir=None, + console=True, + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, +) diff --git a/client.py b/client.py new file mode 100644 index 0000000..feb385d --- /dev/null +++ b/client.py @@ -0,0 +1,65 @@ +import typer +import requests + +app = typer.Typer() + +from bs4 import BeautifulSoup + +# client.py +LOGIN_URL = "https://www.penracourses.org.uk/accounts/login" +# ADD_URL = 'http://localhost/add' + + +@app.command() +def login(user: str, password: str): + rqst = requests.session() + rsp = rqst.get(LOGIN_URL) + + token = BeautifulSoup(rsp.content, "html.parser").find("input").attrs["value"]#, attr={"name": "csrfmiddlewaretoken"}).attrs("value") + + #token = rsp.cookies["csrftoken"] + #header = {"X-CSRFToken": token} + #cookies = {"csrftoken": token} + + data = { + "username": user, + "password": password, + "csrfmiddlewaretoken": token, + "next": "/", + } + + print(data) + rqst.headers["Referer"] = LOGIN_URL + + rsp = rqst.post( + LOGIN_URL, + data=data, + #headers=header, + #cookies=cookies + ) + + soup = BeautifulSoup(rsp.content, "html.parser") + print(token) + print(rqst.headers) + print(rqst.cookies) + + login_success = False + if soup.find("button") and soup.find("button").find(string="Login"): + print("login fail") + pass + else: + print("login success") + login_success = True + #print(rqst.headers) + #print(rsp.content) + + #new = rqst.get( + # "https://www.penracourses.org.uk/api/atlas/get_cases_user" + #) + + #print(new) + #print(new.content) + + +if __name__ == "__main__": + app() diff --git a/reslice.py b/reslice.py new file mode 100644 index 0000000..b70b9bf --- /dev/null +++ b/reslice.py @@ -0,0 +1,64 @@ +import pydicom +import numpy as np +import matplotlib.pyplot as plt +import sys +import glob + +# load the DICOM files +files = [] +print('glob: {}'.format(sys.argv[1])) +for fname in glob.glob(sys.argv[1], recursive=False): + print("loading: {}".format(fname)) + files.append(pydicom.dcmread(fname)) + +print("file count: {}".format(len(files))) + +# skip files with no SliceLocation (eg scout views) +slices = [] +skipcount = 0 +for f in files: + if hasattr(f, 'SliceLocation'): + slices.append(f) + else: + skipcount = skipcount + 1 + +print("skipped, no SliceLocation: {}".format(skipcount)) + +# ensure they are in the correct order +slices = sorted(slices, key=lambda s: s.SliceLocation) + +# pixel aspects, assuming all slices are the same +ps = slices[0].PixelSpacing +ss = slices[0].SliceThickness +ax_aspect = ps[1]/ps[0] +sag_aspect = ps[1]/ss +cor_aspect = ss/ps[0] + +# create 3D array +img_shape = list(slices[0].pixel_array.shape) +img_shape.append(len(slices)) +img3d = np.zeros(img_shape) + +# fill 3D array with the images from the files +for i, s in enumerate(slices): + img2d = s.pixel_array + img3d[:, :, i] = img2d + +# plot 3 orthogonal slices +a1 = plt.subplot(2, 2, 1) +plt.imshow(img3d[:, :, img_shape[2]//2]) +a1.set_aspect(ax_aspect) + +a2 = plt.subplot(2, 2, 2) +plt.imshow(img3d[:, img_shape[1]//2, :]) +a2.set_aspect(sag_aspect) + +a3 = plt.subplot(2, 2, 3) +plt.imshow(img3d[img_shape[0]//2, :, :].T) +a3.set_aspect(cor_aspect) + +a4 = plt.subplot(2, 2, 4) +plt.imshow(img3d[img_shape[0]//2, :, :].T) +a4.set_aspect(cor_aspect) + +plt.show() \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..e4e594f --- /dev/null +++ b/setup.py @@ -0,0 +1,16 @@ +from distutils.core import setup # Need this to handle modules +import py2exe +import math # We have to import all modules used in our program + +import zmq + +setup(console=['anon_gui.py'], py_modules=['anon', 'anon_gui', 'reslice', 'sortList', 'uploader', 'wxDicomViewer'], options={ + 'py2exe': { + 'includes': ['zmq.backend.cython'], + 'excludes': ['zmq.libzmq'], + 'dll_excludes': ['libzmq.pyd'], + } + }, + data_files=[ + ('lib', (zmq.libzmq.__file__,)) + ]) diff --git a/sortList.py b/sortList.py new file mode 100644 index 0000000..edb1af5 --- /dev/null +++ b/sortList.py @@ -0,0 +1,43 @@ +DATA = { +0 : ("3", "3", "1"), +1 : ("2", "1", "2"), +2 : ("1", "2", "3") +} + +import wx +import wx.lib.mixins.listctrl as listmix +from wx.lib.agw import ultimatelistctrl as ULC + +class MyList(ULC.UltimateListCtrl, listmix.ColumnSorterMixin): + def __init__(self, parent, columns): + ULC.UltimateListCtrl.__init__(self, parent, agwStyle=ULC.ULC_REPORT | ULC.ULC_HAS_VARIABLE_ROW_HEIGHT) + self.itemDataMap = DATA + listmix.ColumnSorterMixin.__init__(self, columns) + self.Bind(wx.EVT_LIST_COL_CLICK, self.OnColumn) + + def OnColumn(self, e): + self.Refresh() + e.Skip() + + def GetListCtrl(self): + return self + +class MainWindow(wx.Frame): + def __init__(self, *args, **kwargs): + wx.Frame.__init__(self, *args, **kwargs) + + self.list = MyList(self, 3) + self.list.InsertColumn(0, "A") + self.list.InsertColumn(1, "B") + self.list.InsertColumn(2, "C") + + items = DATA.items() + for key, data in items: + index = self.list.Append(data) + self.list.SetItemData(index, key) + + self.Show() + +app = wx.App(False) +win = MainWindow(None) +app.MainLoop() \ No newline at end of file diff --git a/uploader.py b/uploader.py new file mode 100644 index 0000000..6f6624b --- /dev/null +++ b/uploader.py @@ -0,0 +1,193 @@ +import webview +import time + +import sys + +from datauri import DataURI + +from pathlib import Path + +import pydicom +import dicognito.anonymizer + +import typer +from typing import List + +from watchdog.observers import Observer + +# from watchdog.events import LoggingEventHandler +from watchdog.events import PatternMatchingEventHandler + + +anonymizer = dicognito.anonymizer.Anonymizer() + +app = typer.Typer() + + +def anonymize_file(filepath): + try: + with pydicom.dcmread(filepath) as dataset: + anonymizer.anonymize(dataset) + p = Path(filepath) + new_filename = Path(p.parent, f"{p.stem}_anon{p.suffix}") + dataset.save_as(new_filename) + return new_filename + except: + # Cannot anonymise so just return the same path + return filepath + + +class PythonJavascriptApi: + def __init__(self): + self.cancel_heavy_stuff_flag = False + + def init(self): + response = {"message": "Hello from Python {0}".format(sys.version)} + return response + + def sayHelloTo(self, name): + response = {"message": "Hello {0}!".format(name)} + return response + + def error(self): + raise Exception("This is a Python exception") + + +class Watcher: + def __init__(self, window, anonymise: bool) -> None: + self.window = window + self.anonymise = anonymise + + def on_created(self, event): + print(f"hey, {event.src_path} has been created!") + path = event.src_path + if self.anonymise: + path = anonymize_file(path) + + try: + file_uri = DataURI.from_file(path) + except FileNotFoundError: + print("File deleted...") + return + + file_name = Path(path).name + self.injectFile(file_uri, file_name) + + def on_deleted(self, event): + print(f"what the f**k! Someone deleted {event.src_path}!") + + def on_modified(self, event): + print(f"hey buddy, {event.src_path} has been modified") + + def on_moved(self, event): + print(f"ok ok ok, someone moved {event.src_path} to {event.dest_path}") + + def injectFile(self, file_uri: DataURI, file_name: str): + current_url = webview.windows[0].get_current_url() + if current_url == "https://www.penracourses.org.uk/rapids/create/": + print(f"inject {file_name}") + self.window.evaluate_js( + f""" + function urltoFile(url, filename, mimeType){{ + return (fetch(url) + .then(function(res){{ return res.arrayBuffer(); }}) + .then(function(buf){{return new File([buf], filename,{{type:mimeType}});}}) + ); + }} + + + $(document).ready(() => {{ + //Usage example: + urltoFile('{file_uri}', '{file_name}', '{file_uri.mimetype}') + .then(function(file){{ + + console.log(file); + + //inputs = $("#rapid-form input[type=file][id^=id_images-]"); + inputs = extendInputs(1); + console.log(inputs); + el = inputs.get(0); + console.log(el); + + let dT = new DataTransfer(); + dT.clearData(); + dT.items.add(file); + + el.files = dT.files; + + // Manually fire the change event (qt doesn't seem to...) + $(el).change(); + + + }}); + + }}); + """ + ) + + +@app.command() +def anatomy(filepath: str): + raise NotImplementedError + + +@app.command() +def rapid(watch_path: Path = "/home/ross/watch", anonymise: bool = True): + + api = PythonJavascriptApi() + window = webview.create_window( + "Upload Rapid", "https://www.penracourses.org.uk/rapids/create/", js_api=api + ) + + # def on_loaded(): + # print('DOM is ready') + # current_url = webview.windows[0].get_current_url() + + # if current_url == "https://www.penracourses.org.uk/rapids/create/": + # #window.evaluate_js("alert('tent')") + + # for file_uri, file_name in uri_name_list: + # #time.sleep(2) + + # ## unsubscribe event listener + # #webview.windows[0].loaded -= on_loaded + # #webview.windows[0].load_url('https://google.com') + + # window.loaded += on_loaded + # window = upload_rapid(["/home/ross/Downloads/DICOM CT/DICOM CXR/STD1/SER1/CXR.dcm"]) + + patterns = ["*"] + ignore_patterns = ["*anon*"] + ignore_directories = False + case_sensitive = True + my_event_handler = PatternMatchingEventHandler( + patterns, ignore_patterns, ignore_directories, case_sensitive + ) + + watcher = Watcher(window, anonymise) + my_event_handler.on_created = watcher.on_created + my_event_handler.on_deleted = watcher.on_deleted + my_event_handler.on_modified = watcher.on_modified + my_event_handler.on_moved = watcher.on_moved + my_observer = Observer() + + Path(watch_path).mkdir(parents=True, exist_ok=True) + my_observer.schedule(my_event_handler, watch_path, recursive=True) + + my_observer.start() + # try: + # while True: + # time.sleep(1) + # except KeyboardInterrupt: + # my_observer.stop() + # my_observer.join() + webview.start(None, window, debug=True, gui="qt") + + + +# f = "/media/ross/f6200b79-c343-4882-a245-28fc06d43006/Downloads/temp/breasts.jpg" +f = "/home/ross/Downloads/DICOM CT/DICOM CXR/STD1/SER1/CXR.dcm" + + +if __name__ == "__main__": + app() diff --git a/wxDicomViewer.py b/wxDicomViewer.py new file mode 100644 index 0000000..1adaefd --- /dev/null +++ b/wxDicomViewer.py @@ -0,0 +1,384 @@ +# =================================================================== +# imViewer-Simple.py +# +# An example program that opens uncompressed DICOM images and +# converts them via numPy and PIL to be viewed in wxWidgets GUI +# apps. The conversion is currently: +# +# pydicom->NumPy->PIL->wxPython.Image->wxPython.Bitmap +# +# Gruesome but it mostly works. Surely there is at least one +# of these steps that could be eliminated (probably PIL) but +# haven't tried that yet and I may want some of the PIL manipulation +# functions. +# +# This won't handle RLE, embedded JPEG-Lossy, JPEG-lossless, +# JPEG2000, old ACR/NEMA files, or anything wierd. Also doesn't +# handle some RGB images that I tried. +# +# Have added Adit Panchal's LUT code. It helps a lot, but needs +# to be further generalized. Added test for window and/or level +# as 'list' type - crude, but it worked for a bunch of old MR and +# CT slices I have. +# +# Testing: moderate +# Tested on Windows 7 and MacOS 10.13 using numpy 1.15.2, +# Pillow 5.2.0, Python 2.7.13 / 3.6.5, and wxPython 4.0.3 +# +# Originally written by Dave Witten: Nov. 11, 2009 +# Updated by Aditya Panchal: Oct. 9, 2018 +# =================================================================== + +from io import BytesIO +import pathlib +from PIL import Image +import dicom2jpg +import pydicom +import wx +import cv2 + +have_PIL = True +try: + import PIL.Image +except ImportError: + have_PIL = False + +have_numpy = True +try: + import numpy as np +except ImportError: + have_numpy = False + +# ---------------------------------------------------------------- +# Initialize image capabilities. +# ---------------------------------------------------------------- + +wx.InitAllImageHandlers() + + +def MsgDlg(window, string, caption='OFAImage', style=wx.YES_NO | wx.CANCEL): + """Common MessageDialog.""" + dlg = wx.MessageDialog(window, string, caption, style) + result = dlg.ShowModal() + dlg.Destroy() + return result + + +class ViewerFrame(wx.Frame): + """Class for main window.""" + + def __init__(self, parent, title, initial_file=None): + """Create the pydicom image example's main frame window.""" + + style = wx.DEFAULT_FRAME_STYLE | wx.SUNKEN_BORDER | wx.CLIP_CHILDREN + + wx.Frame.__init__( + self, + parent, + id=-1, + title="", + pos=wx.DefaultPosition, + size=wx.Size(1024, 768), + style=style) + + # -------------------------------------------------------- + # Set up the menubar. + # -------------------------------------------------------- + self.mainmenu = wx.MenuBar() + + # Make the 'File' menu. + menu = wx.Menu() + item = menu.Append(wx.ID_ANY, '&Open...\tCtrl+O', 'Open file for editing') + self.Bind(wx.EVT_MENU, self.OnFileOpen, item) + item = menu.Append(wx.ID_ANY, 'E&xit', 'Exit Program') + self.Bind(wx.EVT_MENU, self.OnFileExit, item) + self.mainmenu.Append(menu, '&File') + + # Attach the menu bar to the window. + self.SetMenuBar(self.mainmenu) + + # -------------------------------------------------------- + # Set up the main splitter window. + # -------------------------------------------------------- + self.mainSplitter = wx.SplitterWindow(self) + self.mainSplitter.SetMinimumPaneSize(1) + + # ------------------------------------------------------------- + # Create the folderTreeView on the left. + # ------------------------------------------------------------- + dsstyle = wx.TR_LINES_AT_ROOT | wx.TR_HAS_BUTTONS + self.dsTreeView = wx.TreeCtrl(self.mainSplitter, style=dsstyle) + + # -------------------------------------------------------- + # Create the ImageView on the right pane. + # -------------------------------------------------------- + imstyle = wx.VSCROLL | wx.HSCROLL | wx.CLIP_CHILDREN + self.imView = wx.Panel(self.mainSplitter, style=imstyle) + + self.imView.Bind(wx.EVT_PAINT, self.OnPaint) + self.imView.Bind(wx.EVT_ERASE_BACKGROUND, self.OnEraseBackground) + + self.imView.Bind(wx.EVT_SIZE, self.OnSize) + + # -------------------------------------------------------- + # Install the splitter panes. + # -------------------------------------------------------- + self.mainSplitter.SplitVertically(self.dsTreeView, self.imView) + self.mainSplitter.SetSashPosition(300, True) + + # -------------------------------------------------------- + # Initialize some values + # -------------------------------------------------------- + self.dcmdsRoot = False + self.foldersRoot = False + self.loadCentered = True + self.bitmap = None + self.Show(True) + + print(initial_file) + + if initial_file is not None: + path = pathlib.Path(initial_file) + self.show_file2(path.name, str(path)) + + def OnFileExit(self, event): + """Exits the program.""" + self.Destroy() + event.Skip() + + def OnSize(self, event): + """Window 'size' event.""" + self.Refresh() + + def OnEraseBackground(self, event): + """Window 'erase background' event.""" + pass + + def populateTree(self, ds): + """ Populate the tree in the left window with the [desired] + dataset values""" + if not self.dcmdsRoot: + self.dcmdsRoot = self.dsTreeView.AddRoot(text="DICOM Objects") + else: + self.dsTreeView.DeleteChildren(self.dcmdsRoot) + self.recurse_tree(ds, self.dcmdsRoot) + self.dsTreeView.ExpandAll() + + def recurse_tree(self, ds, parent, hide=False): + """ order the dicom tags """ + for data_element in ds: + if isinstance(data_element.value, str): + text = str(data_element) + ip = self.dsTreeView.AppendItem(parent, text=text) + else: + ip = self.dsTreeView.AppendItem(parent, text=str(data_element)) + + if data_element.VR == "SQ": + for i, ds in enumerate(data_element.value): + item_describe = data_element.name.replace(" Sequence", "") + item_text = "%s %d" % (item_describe, i + 1) + rjust = item_text.rjust(128) + parentNodeID = self.dsTreeView.AppendItem(ip, text=rjust) + self.recurse_tree(ds, parentNodeID) + + # --- Most of what is important happens below this line --------------------- + + def OnFileOpen(self, event): + """Opens a selected file.""" + msg = 'Choose a file to add.' + dlg = wx.FileDialog(self, msg, '', '', '*.*', wx.FD_OPEN) + if dlg.ShowModal() == wx.ID_OK: + fullPath = dlg.GetPath() + imageFile = dlg.GetFilename() + # checkDICMHeader() + self.show_file2(imageFile, fullPath) + + def OnPaint(self, event): + """Window 'paint' event.""" + dc = wx.PaintDC(self.imView) + dc = wx.BufferedDC(dc) + + # paint a background just so it isn't *so* boring. + dc.SetBackground(wx.Brush("WHITE")) + dc.Clear() + dc.SetBrush(wx.Brush("GREY", wx.CROSSDIAG_HATCH)) + windowsize = self.imView.GetSize() + dc.DrawRectangle(0, 0, windowsize[0], windowsize[1]) + bmpX0 = 0 + bmpY0 = 0 + if self.bitmap is not None: + if self.loadCentered: + bmpX0 = (windowsize[0] - self.bitmap.Width) / 2 + bmpY0 = (windowsize[1] - self.bitmap.Height) / 2 + dc.DrawBitmap(self.bitmap, bmpX0, bmpY0, False) + + # ------------------------------------------------------------ + # ViewerFrame.ConvertWXToPIL() + # Expropriated from Andrea Gavana's + # ShapedButton.py in the wxPython dist + # ------------------------------------------------------------ + def ConvertWXToPIL(self, bmp): + """ Convert wx.Image Into PIL Image. """ + width = bmp.GetWidth() + height = bmp.GetHeight() + im = wx.EmptyImage(width, height) + im.fromarray("RGBA", (width, height), bmp.GetData()) + return im + + # ------------------------------------------------------------ + # ViewerFrame.ConvertPILToWX() + # Expropriated from Andrea Gavana's + # ShapedButton.py in the wxPython dist + # ------------------------------------------------------------ + def ConvertPILToWX(self, pil, alpha=True): + """ Convert PIL Image Into wx.Image. """ + if alpha: + image = wx.Image(pil.size[0], pil.size[1], clear=True) + image.SetData(pil.convert("RGB").tobytes()) + image.SetAlpha(pil.convert("RGBA").tobytes()[3::4]) + else: + image = wx.Image(pil.size[0], pil.size[1], clear=True) + new_image = pil.convert('RGB') + data = new_image.tobytes() + image.SetData(data) + return image + + def get_LUT_value(self, data, window, level): + """Apply the RGB Look-Up Table for the given + data and window/level value.""" + if not have_numpy: + raise ImportError("Numpy is not available. " + "See http://numpy.scipy.org/ " + "to download and install") + + win = window + lvl = level + + e = [ + 0, 255, + lambda data: ((data - (lvl - 0.5)) / (win - 1) + 0.5) * (255 - 0) + ] + return np.piecewise(data, [ + data <= (lvl - 0.5 - (win - 1) / 2), + data > (lvl - 0.5 + (win - 1) / 2) + ], e) + + # ----------------------------------------------------------- + # ViewerFrame.loadPIL_LUT(dataset) + # Display an image using the Python Imaging Library (PIL) + # ----------------------------------------------------------- + def loadPIL_LUT(self, dataset): + if not have_PIL: + raise ImportError("Python Imaging Library is not available." + " See http://www.pythonware.com/products/pil/" + " to download and install") + if 'PixelData' not in dataset: + raise TypeError("Cannot show image -- " + "DICOM dataset does not have pixel data") + + # can only apply LUT if these values exist + if ('WindowWidth' not in dataset) or ('WindowCenter' not in dataset): + bits = dataset.BitsAllocated + samples = dataset.SamplesPerPixel + if bits == 8 and samples == 1: + mode = "L" + elif bits == 8 and samples == 3: + mode = "RGB" + # not sure about this -- PIL source says is + # 'experimental' and no documentation. + elif bits == 16: + # Also, should bytes swap depending + # on endian of file and system?? + mode = "I;16" + else: + msg = "Don't know PIL mode for %d BitsAllocated" % (bits) + msg += " and %d SamplesPerPixel" % (samples) + raise TypeError(msg) + size = (dataset.Columns, dataset.Rows) + + # Recommended to specify all details by + # http://www.pythonware.com/library/pil/handbook/image.htm + im = PIL.Image.frombuffer(mode, size, dataset.PixelData, "raw", + mode, 0, 1) + else: + ew = dataset['WindowWidth'] + ec = dataset['WindowCenter'] + ww = int(ew.value[0] if ew.VM > 1 else ew.value) + wc = int(ec.value[0] if ec.VM > 1 else ec.value) + image = self.get_LUT_value(dataset.pixel_array, ww, wc) + + # Convert mode to L since LUT has only 256 values: + # http://www.pythonware.com/library/pil/handbook/image.htm + im = PIL.Image.fromarray(image).convert('L') + return im + + def show_file(self, imageFile, fullPath): + """ Load the DICOM file, make sure it contains at least one + image, and set it up for display by OnPaint(). ** be + careful not to pass a unicode string to read_file or it will + give you 'fp object does not have a defer_size attribute, + or some such.""" + ds = pydicom.dcmread(str(fullPath)) + + # change strings to unicode + ds.decode() + self.populateTree(ds) + if 'PixelData' in ds: + self.dImage = self.loadPIL_LUT(ds) + if self.dImage is not None: + tmpImage = self.ConvertPILToWX(self.dImage, False) + self.bitmap = wx.Bitmap(tmpImage) + self.Refresh() + + + def show_file2(self, imageFile, fullPath): + """ Load the DICOM file, make sure it contains at least one + image, and set it up for display by OnPaint(). ** be + careful not to pass a unicode string to read_file or it will + give you 'fp object does not have a defer_size attribute, + or some such.""" + ds = pydicom.dcmread(str(fullPath)) + + # change strings to unicode + ds.decode() + self.populateTree(ds) + + with open(fullPath, "rb") as f: + img_data = dicom2jpg.io2img(BytesIO(f.read())) + img = Image.fromarray(img_data) + #img.show() + + tmpImage = self.ConvertPILToWX(img, False) + self.bitmap = wx.Bitmap(tmpImage) + self.Refresh() + + +# ------ This is just the initialization of the App ---- + +# ======================================================= +# The main App Class. +# ======================================================= + + +class BasicDicomViewer(wx.App): + """Image Application.""" + def __init__(self, initial_file=None, redirect=False, filename=None, useBestVisual=False, clearSigInt=True): + self.initial_file = initial_file + super().__init__(redirect=redirect, filename=filename, useBestVisual=useBestVisual, clearSigInt=clearSigInt) + + + def OnInit(self): + """Create the Image Application.""" + print(1) + ViewerFrame(None, 'Simple Dicom Viewer', initial_file=self.initial_file) + return True + + +# ------------------------------------------------------ +# If this file is running as main or a +# standalone test, begin execution here. +# ------------------------------------------------------ + +if __name__ == '__main__': + app = BasicDicomViewer() + app.MainLoop() \ No newline at end of file