diff --git a/anon_gui.py b/anon_gui.py deleted file mode 100644 index cb70922..0000000 --- a/anon_gui.py +++ /dev/null @@ -1,1596 +0,0 @@ -# This is a legacy app written in wxPython -# Whilst it works (or did prior to any API changes) -# it was not particularly well designed -from collections import defaultdict -import json -import os -import threading -from typing import Iterable -from bs4 import BeautifulSoup -import numpy as np -import requests -import webbrowser - -# 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 - from agw import pygauge as PG -except ImportError: # if it's not there locally, try the wxPython lib. - import wx.lib.agw.hyperlink as hl - import wx.lib.agw.pygauge as PG - - -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 threading - -# import watchdog.observers - -__VERSION__ = "0.1" -__APP_NAME__ = "Uploader" - -# Define notification event for thread completion -EVT_RESULT_ID = wx.NewId() - -def EVT_RESULT(win, func): - """Define Result Event.""" - win.Connect(-1, -1, EVT_RESULT_ID, func) - -class ResultEvent(wx.PyEvent): - """Simple event to carry arbitrary result data.""" - def __init__(self, data): - """Init Result Event.""" - wx.PyEvent.__init__(self) - self.SetEventType(EVT_RESULT_ID) - self.data = data - -# Thread class that executes processing -class WorkerThread(threading.Thread): - """Worker Thread Class.""" - def __init__(self, notify_window): - """Init Worker Thread Class.""" - threading.Thread.__init__(self) - self._notify_window = notify_window - self._want_abort = 0 - # This starts the thread running on creation, but you could - # also make the GUI thread responsible for calling this - self.start() - - def run(self): - """Run Worker Thread.""" - # This is the code executing in the new thread. Simulation of - # a long process (well, 10s here) as a simple loop - you will - # need to structure your processing so that you periodically - # peek at the abort variable - for i in range(10): - time.sleep(1) - if self._want_abort: - # Use a result of None to acknowledge the abort (of - # course you can use whatever you'd like or even - # a separate event type) - wx.PostEvent(self._notify_window, ResultEvent(None)) - return - # Here's where the result would be returned (this is an - # example fixed result of the number 10, but it could be - # any Python object) - wx.PostEvent(self._notify_window, ResultEvent(10)) - - def abort(self): - """abort worker thread.""" - # Method for use by main thread to signal an abort - self._want_abort = 1 - -def resource_path(relative_path): - """ Get absolute path to resource, works for dev and for PyInstaller """ - try: - # PyInstaller creates a temp folder and stores path in _MEIPASS - base_path = sys._MEIPASS - except Exception: - base_path = os.path.abspath(".") - - return os.path.join(base_path, relative_path) - -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, "logs", "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, - dev: bool = False, - remove_files_on_annonymisation: bool = True, -): - 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, - remove_files_on_annonymisation=remove_files_on_annonymisation, - ) - return - if path.is_dir(): - files = list(Path(path).glob("*.dcm")) - elif path.is_file(): - if file_list: - with open(path) as f: - files = [Path(p) for p in 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") - - logger.debug(f"anon: {n}") - anonymize_file(file, new_filepath) - - # progress_dialog.Destroy() - - return new_filepath - - -@logger.catch() -def anonymize_file(file: Path, new_filepath: Path, remove_original: bool = False): - 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) - - if remove_original: - os.remove(file) - return dataset, 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, id=wx.ID_OK, 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.Bind(wx.EVT_CLOSE, self.on_close) - - self.SetSizer(main_sizer) - - def on_close(self, event): - self.Destroy() - - # ---------------------------------------------------------------------- - 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.Enable() - 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, - remove_files_on_annonymisation=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.clear_files_on_upload = True - self.remove_files_on_annonymisation = remove_files_on_annonymisation - - self.rqst = None - - self.worker = 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") - wx.adv.NotificationMessage("Another instance detected", "Trigger refresh").Show() - 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=__APP_NAME__, - style=wx.DEFAULT_FRAME_STYLE | wx.STAY_ON_TOP, - size=(300, 400), - ) - - self.frame.SetIcon(wx.Icon(resource_path("icon\icon1.ico"))) - - self.panel = wx.Panel(self.frame, wx.ID_ANY) - - self.file_detail_map = {} - self.file_duplicate_check = {} - - self.list_ctrl = SortableList( - self.panel, - 6, - ) - self.series_grid = SeriesGrid(self.panel) - - if self.clear_files_on_start: - self.ClearFiles(load_dir=False) - - self.panel.sizer = wx.BoxSizer(wx.VERTICAL) - - #self.link_rapid_upload = hl.HyperLinkCtrl( - # self.panel, - # wx.ID_ANY, - # "Rapid", - # 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.link_atlas_case = hl.HyperLinkCtrl( - # self.panel, - # wx.ID_ANY, - # "Atlas Case", - # URL=f"{self.base_site_url}/atlas/case/create/", - #) - # self.panel.sizer.Add(self.link_long_upload, 0, wx.ALL, 10) - - # Create menu bar - self.menu_bar = wx.MenuBar() - - - # Create File menu - file_menu = wx.Menu() - - - # Create Load menu item - load_item = wx.MenuItem(file_menu, wx.ID_ANY, "&Import Directory") - #load_item.SetToolTip("Click to manually load dicom from a directory.") - file_menu.Append(load_item) - - self.Bind(wx.EVT_MENU, self.import_from_dir, load_item) - #file_menu.Append(wx.ID_OPEN, "&Open") - #file_menu.Append(wx.ID_SAVE, "&Save") - #file_menu.AppendSeparator() - file_menu.Append(wx.ID_EXIT, "&Exit") - self.Bind(wx.EVT_MENU, lambda evt: wx.Exit(), id=wx.ID_EXIT) - - self.menu_bar.Append(file_menu, "&File") - - - # Create Images menu - images_menu = wx.Menu() - - # Create Clear Files menu item - clear_files_item = wx.MenuItem(images_menu, wx.ID_ANY, "&Clear files") - #clear_files_item.SetToolTip("Click to clear all files.") - images_menu.Append(clear_files_item) - - self.menu_bar.Append(images_menu, "&Images") - - # Create Reorder Files menu item - reorder_files_item = wx.MenuItem(images_menu, wx.ID_ANY, "&Reorder") - #reorder_files_item.SetToolTip("Click to reorder files as they are shown below. This will sequentially rename the files. Please note this is usually better done once the files have been uploaded.") - images_menu.Append(reorder_files_item) - - self.Bind(wx.EVT_MENU, self.on_reorder_files, reorder_files_item) - - # Bind Clear Files menu item event - self.Bind(wx.EVT_MENU, self.ClearFiles, clear_files_item) - - - # Create Reannonymise Files menu item - reannonymise_files_item = wx.MenuItem(images_menu, wx.ID_ANY, "&Reannonymise") - #reannonymise_files_item.SetToolTip("Click to reannonymise the currently loaded files.") - images_menu.Append(reannonymise_files_item) - - self.Bind(wx.EVT_MENU, self.on_reannonymise_files, reannonymise_files_item) - - - # Create Refresh Files menu item - refresh_files_item = wx.MenuItem(images_menu, wx.ID_ANY, "&Refresh") - images_menu.Append(refresh_files_item) - - self.Bind(wx.EVT_MENU, lambda evt: self.LoadDir(), refresh_files_item) - - - # Create Item creation menu - create_menu = wx.Menu() - - create_menu_items = ( - ("&Rapid", f"/rapids/create/"), - ("&Long Series", f"/longs/create/series"), - ("&Atlas Case", f"/atlas/case/create/"), - ) - - for label, url in create_menu_items: - item = wx.MenuItem(create_menu, wx.ID_ANY, label) - create_menu.Append(item) - self.Bind(wx.EVT_MENU, lambda evt, url=url: webbrowser.open(f"{self.base_site_url}{url}"), item) - - self.menu_bar.Append(create_menu, "&Create") - - # Create Help menu - help_menu = wx.Menu() - help_menu.Append(wx.ID_HELP, "&Help") - help_menu.Append(wx.ID_ABOUT, "&About") - self.menu_bar.Append(help_menu, "&Help") - - self.Bind(wx.EVT_MENU, self.on_about_box, id=wx.ID_ABOUT) - self.Bind(wx.EVT_MENU, self.on_help, id=wx.ID_HELP) - - # Set menu bar - self.frame.SetMenuBar(self.menu_bar) - - #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) - #hoz_links.Add(self.link_atlas_case, 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") - bottom_button_sizer = wx.BoxSizer(wx.HORIZONTAL) - self.drag_button = wx.Button( - self.panel, id=wx.ID_ANY, label="Drag" - ) - self.drag_button.SetToolTip("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.SetToolTip("Upload files to the server. You must be logged in.") - self.upload_button.Disable() - bottom_button_sizer.Add(self.upload_button, 4, wx.EXPAND) - bottom_button_sizer.Add(self.drag_button, 1, wx.EXPAND) - - # 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.list_ctrl.InsertColumn(6, "Hash") - self.list_ctrl.InsertColumn(7, "Series Instance UID") - - # self.UpdateListCtrl() - - self.login_button = wx.Button(self.panel, id=wx.ID_ANY, label="Login") - self.login_button.SetForegroundColour(wx.BLUE) - self.login_button.SetBackgroundColour(wx.WHITE) - self.login_button.SetToolTip("Click to login to the server") - - - # 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) - # 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) - - 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) - self.panel.sizer.Add(bottom_button_sizer, 1, wx.EXPAND) - - self.progress_sizer = wx.BoxSizer(wx.VERTICAL) - self.progress_bar = wx.Gauge( - self.panel, style=wx.GA_HORIZONTAL | wx.GA_PROGRESS - ) - # self.progress_bar = PG.PyGauge(self.panel, style=wx.GA_HORIZONTAL|wx.GA_PROGRESS) - # self.progress_bar.SetDrawValue(draw=True, drawPercent=True, font=None, colour=wx.RED, formatString=None) - self.progress_text = wx.StaticText(self.panel, label="Text") - self.progress_sizer.Add(self.progress_bar, 2, wx.EXPAND) - self.progress_sizer.Add(self.progress_text, 2, wx.EXPAND) - self.panel.sizer.Add(self.progress_sizer, 2, wx.EXPAND) - self.progress_sizer.ShowItems(show=False) - - # Layout sizers - self.panel.SetSizer(self.panel.sizer) - self.panel.SetAutoLayout(1) - self.panel.sizer.Fit(self.panel) - self.frame.Show() - # self.spinner.Hide() - - wx.CallAfter(self.post_init) - - return super().OnInit() - - def post_init(self): - self.AnnonymiseFilesInDir(self.input_dir, self.dir) - - self.Bind(wx.EVT_KEY_DOWN, self.onKeyPress) - - self.image_hash_check_in_progress = False - self.duplicate_check_timer = wx.Timer() - self.Bind(wx.EVT_TIMER, self.check_image_hashes, self.duplicate_check_timer) - self.duplicate_check_timer.Start(2000) - - 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 import_from_dir(self, event): - default_path = str(self.input_dir.resolve()) - with wx.DirDialog( - self.frame, "Choose a directory to import dicom files from.", style=wx.DD_DIR_MUST_EXIST, defaultPath=default_path - ) as dlg: - if dlg.ShowModal() == wx.ID_OK: - load_path = Path(dlg.GetPath()) - self.AnnonymiseFilesInDir(load_path, self.dir, remove_original=False) - - def AnnonymiseFilesInDir(self, input_dir, output_dir, remove_original: bool | None=None): - 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")) - - if not files: - # TODO load from DICOMDIR - to_exclude = ("DICOMDIR") - files = [i for i in Path(input_dir).rglob("*") if i.is_file() and i.name not in to_exclude] - - - - - - else: - return - - if not files: - logger.debug("No files to annonymise") - wx.adv.NotificationMessage("Annonymise files", "No files to annonymise").Show() - return - - logger.debug(files) - - - if remove_original is None: - remove_original = self.remove_files_on_annonymisation - - # 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, - # ) - with self.progress(len(files), "Annonymise files") as p: - # p.SetRange(len(files)) - - i: int = 0 - for n, file in enumerate(files): - # progress_dialog.Update(n) - p.update(n) - new_filepath: Path - # Get next available filename - 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") - - logger.debug(f"anon: {n}") - try: - dataset, filepath = anonymize_file(file, new_filepath, remove_original=remove_original) - except pydicom.errors.InvalidDicomError: - continue - self.add_datails_to_file_detail_map(dataset, filepath) - wx.GetApp().Yield() - # progress_dialog.Destroy() - - self.LoadDir(self.dir) - - @contextlib.contextmanager - def progress(self, total: int, text: str = "Working"): - # self.progress_sizer.Show() - self.progress_text.SetLabelText(text) - self.progress_bar.SetRange(total) - self.progress_sizer.ShowItems(show=True) - self.panel.sizer.Fit(self.panel) - self.frame.Layout() - try: - - class P: - def __init__(self, gui, total, text): - self.gui = gui - self.text = text - self.total = total - - def update(self, n: int): - self.gui.progress_bar.SetValue(n) - self.gui.progress_text.SetLabelText(f"{self.text} : {n}") - self.gui.frame.Refresh() - - yield P(self, total, text) - finally: - # self.progress_sizer.Hide() - self.progress_sizer.ShowItems(show=False) - self.panel.Layout() - # self.panel.SetAutoLayout(1) - # self.panel.sizer.Fit(self.panel) - - def onKeyPress(self, event): - keycode = event.GetKeyCode() - logger.debug(keycode) - if keycode == wx.WXK_F1: - logger.debug("you pressed the F1!") - self.on_help() - return - event.Skip() - - def on_help(self, event=None): - webbrowser.open(f"{self.base_site_url}/atlas/help") - - def on_about_box(self, event=None): - description = f""" -A dicom export/upload tool - -Settings --------- -Watch/import dir: {self.input_dir} -Annonymised files dir: {self.dir} - -clear files on start: {self.clear_files_on_start} -clear files on close: {self.clear_files_on_close} -remove files on annonymisations: {self.remove_files_on_annonymisation} - -base site url: {self.base_site_url} -""" - - licence = """GPL""" - - info = wx.adv.AboutDialogInfo() - - # info.SetIcon(wx.Icon('hunter.png', wx.BITMAP_TYPE_PNG)) - info.SetName(__APP_NAME__) - info.SetVersion(__VERSION__) - info.SetDescription(description) - icon_path = resource_path("icon\icon1.png") - desired_width = 200 - desired_height = 200 - - # Load the icon image - image = wx.Image(icon_path) - - # Resize the image - resized_image = image.Rescale(desired_width, desired_height) - - # Create a new icon with the resized image - resized_icon = wx.Icon(resized_image.ConvertToBitmap()) - - info.SetIcon(resized_icon) - - info.SetLicence(licence) - - wx.adv.AboutBox(info) - - def OnExit(self, event=None): - print("on exit") - if self.clear_files_on_close: - self.ClearFiles(load_dir=False) - - # self.zmq_thread.join() - - return super().OnExit() - - def LoadDir(self, dir=None, refresh=True, clear_file_detail_map=False): - """Loads data from the output directory (annonymised files)""" - if clear_file_detail_map: - self.file_detail_map = {} - self.file_duplicate_check = {} - if dir is None: - dir = self.dir - - logger.debug(f"LoadDir : {dir} / refresh: {refresh} / clear_file_detail_map: {clear_file_detail_map}") - - 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) - - for i in range(self.list_ctrl.GetItemCount()): - if self.list_ctrl.IsSelected(i): - path = self.list_ctrl.GetItemPyData(i)["file"] - if ( - path in self.file_duplicate_check - and self.file_duplicate_check[path]["id"] - ): - if self.file_duplicate_check[path]["type"] == "series": - menu.Append(3, "View on site (series)") - menu.Bind( - wx.EVT_MENU, - lambda evt: webbrowser.open( - f"{self.base_site_url}{self.file_duplicate_check[path]['url']}" - ), - id=3, - ) - else: - menu.Append(3, "View on site (uncategorised)") - menu.Bind( - wx.EVT_MENU, - lambda evt: webbrowser.open( - f"{self.base_site_url}{self.file_duplicate_check[path]['url']}" - ), - id=3, - ) - break - - 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 on_reannonymise_files(self, event): - global anonymizer - anonymizer = dicognito.anonymizer.Anonymizer() - paths = [] - - # self.file_observer.pause() - - tempdir = Path(tempfile.gettempdir()) - to_move = [] - with self.progress(self.list_ctrl.GetItemCount(), "Reannonymise files") as p: - for i in range(self.list_ctrl.GetItemCount()): - p.update(i) - 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 on_reorder_files(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, load_dir: bool = True, files:None| list[(str, str)] = None): - if files is None: - files = Path(self.dir).glob("*") - else: - files = [Path(self.dir, f) for f, file_hash in files] - - for f in files: - try: - os.remove(f) - except OSError: - logger.debug(f"File to delete not found: {f}") - pass - - - if load_dir: - self.LoadDir(clear_file_detail_map=True) - - def add_datails_to_file_detail_map(self, dataset, file): - 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 = "..." - - 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()) - # md5.update(dataset.PixelData) - # md5.update("".join(dataset.pixel_array.astype(str).flatten()).encode()) - # md5.update(np.char.join("", dataset.pixel_array.astype(str).flatten()).encode()) - # hash = md5.hexdigest() - # print(hash) - - # md5 = hashlib.md5() - ##hash = md5.hexdigest() - ##md5 = hashlib.md5() - ##first = True - ##s = ",".join([f"{i}" for i in dataset.pixel_array.astype(str).flatten()]).encode() - ##print(s) - # print(dataset.PixelData[:20]) - # print(len(dataset.PixelData)) - # md5.update(dataset.PixelData) - - # hash = md5.hexdigest() - - hasher = blake3() - hasher.update(dataset.PixelData) - hash = hasher.hexdigest() - # dataset.PixelData - - d = ( - file.name, - study_desc, - instance_number, - patient_name, - series_desc, - slice_location, - hash, - series_instances_uid, - ) - - # hasher = blake3() - # first = True - # for i in dataset.pixel_array.astype(str).flatten(): - # if first: - # first = False - # hasher.update(f"{i}".encode()) - # else: - # hasher.update(f",{i}".encode()) - - # hash = hasher.digest() - - # hash = get_file_hash(file, hasher="md5") - - # hash = None # If we don't use the hash... - self.file_detail_map[file] = d, hash, d2 - - return d, hash, d2 - - 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, - # ) - with self.progress((len(self.files)), "Loading files") as p: - # p.SetRange(len(self.files)) - - self.series_map = defaultdict(list) - self.series_map_details = defaultdict(dict) - - hashes = [] - - for n, f in enumerate(self.files): - logger.debug(f"update {n}") - # progress_dialog.Update(n) - #p.update(n) - # dataset = pydicom.read_file(f) - - if f in self.file_detail_map: - d, hash, d2 = self.file_detail_map[f] - #print(f"{n}: in map") - else: - logger.debug("read: {f}") - try: - dataset = pydicom.dcmread(f) - except FileNotFoundError: - return - - #print("recalc") - d, hash, d2 = self.add_datails_to_file_detail_map(dataset, f) - - hashes.append(hash) - - 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) - - if f in self.file_duplicate_check: - if self.file_duplicate_check[f]["id"]: - self.list_ctrl.SetItemBackgroundColour(n, wx.RED) - - n = n + 1 - #self.list_ctrl.UpdateData(DATA) - self.populate_series_grid() - - self.frame.Refresh() - self.frame.Layout() - - # progress_dialog.Destroy() - - # if hashes: - # self.check_image_hashes(hashes) - - def check_image_hashes(self, evt): - if self.login_success and not self.image_hash_check_in_progress: - self.image_hash_check_in_progress = True - - to_check = self.file_detail_map.keys() - self.file_duplicate_check.keys() - - if to_check: - hash_file_map = {} - for file in to_check: - hash_file_map[self.file_detail_map[file][1]] = file - - hashes = list(hash_file_map.keys()) - - 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/check_image_hashes/", - data=json.dumps(hashes), - # files=files, - ) - - for hash, dup in resp.json().items(): - self.file_duplicate_check[hash_file_map[hash]] = dup - - if resp.json().items(): - wx.CallAfter(self.UpdateListCtrl) - self.image_hash_check_in_progress = False - - 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 = [] - duplicate_files_not_uploaded = [] - logger.debug("Upload files") - for f in self.files: - if f in self.file_duplicate_check and self.file_duplicate_check[f]["id"]: - # duplicate the api return format so ClearFiles works - duplicate_files_not_uploaded.append((f.name, "...")) - else: - 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, - # ) - with self.progress(len(chunked_files), "Uploading files (in batches)") as p: - # p.SetRange(len(chunked_files)) - - upload_file_list = [] - duplicate_file_list = [] - failed = [] - - 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 - - p.update(n) - # progress_dialog.Update(n, f"Uploading batch {n}/{len(chunked_files)}") - - logger.debug(f"n: {n}") - # try to upload the files - - for i in range(3): - resp = upload_files(files) - if resp.status_code == 200: - upload_file_list.extend(resp.json()["uploaded"]) - duplicate_file_list.extend(resp.json()["duplicates"]) - failed.extend(resp.json()["failed"]) - - continue - - logger.debug(f"n: {n} fail (attempt {i})") - - # progress_dialog.Destroy() - - print(upload_file_list) - print("dup", duplicate_file_list) - print("failed", failed) - - - - wx.CallAfter( - self.upload_complete_message, upload_file_list, duplicate_file_list, failed, duplicate_files_not_uploaded - ) - - if self.clear_files_on_upload: - all_files = upload_file_list + duplicate_file_list + duplicate_files_not_uploaded - print(all_files) - wx.CallAfter(self.ClearFiles,event=None, load_dir=True, files=all_files) - - def upload_complete_message(self, upload_file_list, duplicate_file_list, failed, duplicate_files_not_uploaded): - wx.MessageBox( - f"""Files uploaded: {len(upload_file_list)} -Duplicate files (not uploaded): {len(duplicate_file_list)} -Failed to upload: {len(failed)} -Duplicate files (not uploaded): {len(duplicate_files_not_uploaded)}""" - ) - self.file_duplicate_check = {} - - 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\\", - output_dir: Path = "C:\\temp_upload\\", - silent_fail: bool = False, - clear_files: bool = True, - clear_files_on_close: bool = True, - remove_files_on_annonymisation: bool = True, - dev: bool = False, -): - if "linux" in sys.platform: - input_dir = Path("/home/ross/temp/in/") - output_dir = Path("/home/ross/temp/out/") - - 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, - remove_files_on_annonymisation, - ) - - -def launchGUI( - input_dir, - output_dir, - silent_fail, - clear_files, - clear_files_on_close, - dev, - remove_files_on_annonymisation, -): - 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, - remove_files_on_annonymisation=remove_files_on_annonymisation, - ) - app.MainLoop() - - -if __name__ == "__main__": - app() - diff --git a/anon_gui.spec b/anon_gui.spec deleted file mode 100644 index 2156776..0000000 --- a/anon_gui.spec +++ /dev/null @@ -1,41 +0,0 @@ -# -*- mode: python ; coding: utf-8 -*- - - -a = Analysis( - ['anon_gui.py'], - pathex=[], - binaries=[], - datas=[ - ('icon\\icon1.ico', 'icon'), - ('icon\\icon1.png', 'icon'), - ], - hiddenimports=[], - hookspath=[], - hooksconfig={}, - runtime_hooks=[], - excludes=[], - noarchive=False, -) -pyz = PYZ(a.pure) - -exe = EXE( - pyz, - a.scripts, - a.binaries, - 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, - icon='icon\\icon1.ico' -) diff --git a/requirements.txt b/requirements.txt index cab43c7..42a30f7 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,12 +1,12 @@ typer -wxpython pydicom dicognito python-datauri bs4 requests -zmq +pynng blake3 loguru dicom2jpg -pyinstaller \ No newline at end of file +pyinstaller +nicegui diff --git a/wxDicomViewer.py b/wxDicomViewer.py deleted file mode 100644 index 4787f1b..0000000 --- a/wxDicomViewer.py +++ /dev/null @@ -1,384 +0,0 @@ -# =================================================================== -# 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, int(bmpX0), int(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