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 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, "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, ): 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 = [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") anonymize_file(file, new_filepath) # progress_dialog.Destroy() return new_filepath 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.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.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.remove_files_on_annonymisation = remove_files_on_annonymisation 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") 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="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.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() 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.menu_bar.Append(file_menu, "&File") # Create Edit menu edit_menu = wx.Menu() edit_menu.Append(wx.ID_UNDO, "&Undo") edit_menu.Append(wx.ID_REDO, "&Redo") edit_menu.AppendSeparator() edit_menu.Append(wx.ID_CUT, "Cu&t") edit_menu.Append(wx.ID_COPY, "&Copy") edit_menu.Append(wx.ID_PASTE, "&Paste") self.menu_bar.Append(edit_menu, "&Edit") # 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") # 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") upload_drag_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() upload_drag_sizer.Add(self.upload_button, 4, wx.EXPAND) upload_drag_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") self.clear_files_button = wx.Button( self.panel, id=wx.ID_ANY, label="Clear files" ) self.clear_files_button.SetToolTip("Click to clear all 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.reorder_files_button.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.") self.reannonymise_files_button = wx.Button( self.panel, id=wx.ID_ANY, label="Reannonymise" ) self.reannonymise_files_button.SetToolTip("Click to reannonymise the currently loaded files.") self.manual_load_button = wx.Button(self.panel, id=wx.ID_ANY, label="Load") self.manual_load_button.SetToolTip("Click to manually load dicom from a directory.") # 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, #) self.Bind(wx.EVT_BUTTON, self.import_from_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) self.panel.sizer.Add(upload_drag_sizer, 2, 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", 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) 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 if not files: logger.debug("No files to annonymise") wx.adv.NotificationMessage("Annonymise files", "No files to annonymise").Show() 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, # ) with self.progress(len(files), "Annonymise files") as p: # p.SetRange(len(files)) for n, file in enumerate(files): # progress_dialog.Update(n) p.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") dataset, filepath = anonymize_file(file, new_filepath, remove_original=self.remove_files_on_annonymisation) self.add_datails_to_file_detail_map(dataset, filepath) # 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.Layout() self.panel.sizer.Fit(self.panel) 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}") 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.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(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 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"] ): menu.Append(3, "View on site") 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 ReAnnonymiseFiles(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 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, load_dir: bool = True): files = Path(self.dir).glob("*") for f in files: os.remove(f) 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): # 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("in map") else: 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() # 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 = [] 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, # ) with self.progress(len(chunked_files), "Uploading files (in batches)") as p: # p.SetRange(len(chunked_files)) 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 p.update(n) # 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)}" ) 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()