Compare commits
33
Commits
66b4fffb7b
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
134bbf85b6 | ||
|
|
b7242d19de | ||
|
|
854fd315fa | ||
|
|
ffc4c79def | ||
|
|
8780a409de | ||
|
|
2973e86d45 | ||
|
|
bb0d3d8af1 | ||
|
|
9c7777323d | ||
|
|
796027a971 | ||
|
|
3680716d71 | ||
|
|
7528202978 | ||
|
|
377be258de | ||
|
|
84d18f25e1 | ||
|
|
a0ee79e318 | ||
|
|
27f4aeec5e | ||
|
|
f1b8b81c76 | ||
|
|
305e47ba8f | ||
|
|
6a49f47a5b | ||
|
|
b40e802627 | ||
|
|
f0f50381ec | ||
|
|
5f2bcb81a4 | ||
|
|
f42c1a1839 | ||
|
|
094d66b137 | ||
|
|
6366557d20 | ||
|
|
fa469a79fc | ||
|
|
e93e7a8861 | ||
|
|
6db2a8fcf3 | ||
|
|
21328b53dd | ||
|
|
eb061992a3 | ||
|
|
a8c48c12f9 | ||
|
|
c735b45a25 | ||
|
|
05c80a8fd7 | ||
|
|
ae4d127e89 |
+3
-1
@@ -13,4 +13,6 @@ test/*
|
|||||||
|
|
||||||
*.build
|
*.build
|
||||||
|
|
||||||
C:/* # Windows paths on linux
|
C:/* # Windows paths on linux
|
||||||
|
|
||||||
|
uploader.log
|
||||||
-1596
File diff suppressed because it is too large
Load Diff
@@ -1,46 +1,204 @@
|
|||||||
import shutil
|
import shutil
|
||||||
|
import importlib.util
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
import PyInstaller.__main__
|
import PyInstaller.__main__
|
||||||
import typer
|
import typer
|
||||||
from rich import print
|
from rich import print
|
||||||
from rich.progress import Progress
|
import time
|
||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
|
||||||
app = typer.Typer()
|
app = typer.Typer()
|
||||||
|
|
||||||
|
|
||||||
@app.command()
|
@app.command()
|
||||||
def build(build: bool = True, copy: bool = True, test: bool = False):
|
def build(build: bool = True, copy: bool = True, test: bool = False, dest: str | None = None):
|
||||||
|
"""Windows-focused build script.
|
||||||
|
|
||||||
|
- `--test` creates a one-file test binary in `dist\test`.
|
||||||
|
- `--copy` (default) copies the artifact to the shared network path used previously.
|
||||||
|
"""
|
||||||
|
|
||||||
|
project_root = Path(__file__).resolve().parent
|
||||||
|
|
||||||
if build:
|
if build:
|
||||||
|
if test:
|
||||||
|
print("Building Windows test one-file bundle with PyInstaller")
|
||||||
|
# On Windows use ';' as add-data separator (src;dest)
|
||||||
|
# Use absolute path for the icon folder so PyInstaller can find it
|
||||||
|
icon_src = str(project_root / "icon")
|
||||||
|
add_data_args = ["--add-data", f"{icon_src};icon"]
|
||||||
|
|
||||||
print("Building with PyInstaller")
|
# Try to include dicognito package data (release_notes.md) which
|
||||||
PyInstaller.__main__.run([
|
# PyInstaller may miss. This prevents runtime FileNotFoundError for
|
||||||
"anon_gui.spec"
|
# dicognito/release_notes.md when the package expects the file at runtime.
|
||||||
])
|
try:
|
||||||
|
spec = importlib.util.find_spec("dicognito")
|
||||||
|
if spec and spec.submodule_search_locations:
|
||||||
|
dicognito_path = Path(spec.submodule_search_locations[0])
|
||||||
|
release_notes = dicognito_path / "release_notes.md"
|
||||||
|
if release_notes.exists():
|
||||||
|
# PyInstaller on Windows expects 'src;dest'
|
||||||
|
add_data_args += ["--add-data", f"{str(release_notes)};dicognito"]
|
||||||
|
else:
|
||||||
|
# include whole package folder if the single file wasn't found
|
||||||
|
add_data_args += ["--add-data", f"{str(dicognito_path)};dicognito"]
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Warning: couldn't locate dicognito package: {e}")
|
||||||
|
|
||||||
|
# Try to include pythonnet runtime DLLs to avoid missing
|
||||||
|
# Python.Runtime.dll at runtime (used by pythonnet / webview).
|
||||||
|
add_binary_args = []
|
||||||
|
try:
|
||||||
|
spec_py = importlib.util.find_spec("pythonnet")
|
||||||
|
if spec_py and spec_py.submodule_search_locations:
|
||||||
|
pn_path = Path(spec_py.submodule_search_locations[0])
|
||||||
|
runtime_dir = pn_path / "runtime"
|
||||||
|
if runtime_dir.exists():
|
||||||
|
for f in runtime_dir.rglob("*.dll"):
|
||||||
|
# PyInstaller on Windows expects 'src;dest'
|
||||||
|
add_binary_args += ["--add-binary", f"{str(f)};pythonnet\\runtime"]
|
||||||
|
# also include the full pythonnet package folder as data so
|
||||||
|
# package-relative lookups succeed at runtime
|
||||||
|
add_data_args += ["--add-data", f"{str(pn_path)};pythonnet"]
|
||||||
|
# try to include clr_loader package as well
|
||||||
|
try:
|
||||||
|
spec_clr = importlib.util.find_spec("clr_loader")
|
||||||
|
if spec_clr and spec_clr.submodule_search_locations:
|
||||||
|
clr_path = Path(spec_clr.submodule_search_locations[0])
|
||||||
|
add_data_args += ["--add-data", f"{str(clr_path)};clr_loader"]
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Warning: couldn't locate pythonnet runtime: {e}")
|
||||||
|
|
||||||
|
# Ensure dist/work/spec directories exist
|
||||||
|
dist_dir = project_root / "dist" / "test"
|
||||||
|
work_dir = project_root / "build" / "test"
|
||||||
|
spec_dir = project_root / "build" / "specs"
|
||||||
|
dist_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
work_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
spec_dir.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
# Decide build name and try to remove any locked previous artifact.
|
||||||
|
base_name = "Uploader_test"
|
||||||
|
output_exe = dist_dir / f"{base_name}.exe"
|
||||||
|
|
||||||
|
def try_remove_with_retries(path: Path, retries: int = 5, delay: float = 0.5) -> bool:
|
||||||
|
"""Try to remove a file with retries and backoff. Returns True if removed or not present."""
|
||||||
|
if not path.exists():
|
||||||
|
return True
|
||||||
|
for attempt in range(retries):
|
||||||
|
try:
|
||||||
|
path.unlink()
|
||||||
|
print(f"Removed existing file: {path}")
|
||||||
|
return True
|
||||||
|
except PermissionError:
|
||||||
|
# Try to move/rename it out of the way
|
||||||
|
try:
|
||||||
|
backup = path.with_suffix(f".old.{int(time.time())}")
|
||||||
|
path.rename(backup)
|
||||||
|
print(f"Renamed locked file to: {backup}")
|
||||||
|
return True
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
print(f"File locked, retrying in {delay} seconds ({attempt+1}/{retries})")
|
||||||
|
time.sleep(delay)
|
||||||
|
delay *= 1.5
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Failed to remove existing file {path}: {e}")
|
||||||
|
return False
|
||||||
|
return False
|
||||||
|
|
||||||
|
name_used = base_name
|
||||||
|
if not try_remove_with_retries(output_exe):
|
||||||
|
# If the previous artifact is locked, fall back to a unique name to avoid build failure
|
||||||
|
ts = datetime.now().strftime("%Y%m%dT%H%M%S")
|
||||||
|
name_used = f"{base_name}_{ts}"
|
||||||
|
print(f"Previous artifact appears locked; building as {name_used} to avoid PermissionError")
|
||||||
|
|
||||||
|
# Ensure cert folder is included for test builds as well (if present)
|
||||||
|
cert_dir = project_root / "cert"
|
||||||
|
if cert_dir.exists() and cert_dir.is_dir():
|
||||||
|
arg = f"{str(cert_dir)};cert"
|
||||||
|
# Avoid adding duplicate entries
|
||||||
|
if arg not in add_data_args:
|
||||||
|
add_data_args += ["--add-data", arg]
|
||||||
|
|
||||||
|
PyInstaller.__main__.run([
|
||||||
|
str(project_root / "nice.py"),
|
||||||
|
"--onefile",
|
||||||
|
"--name",
|
||||||
|
name_used,
|
||||||
|
"--distpath",
|
||||||
|
str(dist_dir),
|
||||||
|
"--workpath",
|
||||||
|
str(work_dir),
|
||||||
|
"--specpath",
|
||||||
|
str(spec_dir),
|
||||||
|
] + add_data_args + add_binary_args + [
|
||||||
|
# add some common hidden imports that PyInstaller sometimes misses
|
||||||
|
"--hidden-import",
|
||||||
|
"bs4",
|
||||||
|
"--hidden-import",
|
||||||
|
"requests",
|
||||||
|
"--hidden-import",
|
||||||
|
"nicegui",
|
||||||
|
])
|
||||||
|
else:
|
||||||
|
print("Building using spec: nice_uploader.spec")
|
||||||
|
PyInstaller.__main__.run([str(project_root / "nice_uploader.spec")])
|
||||||
|
|
||||||
if copy:
|
if copy:
|
||||||
print("Copying file")
|
# Windows-only default destination used previously in the repo
|
||||||
source = Path(r"C:\Users\krugerro\Desktop\uploader\dist\anon_gui.exe" )
|
default_dest = Path(r"\\ict\Go\RCH\Shared\Dragon-Xray\rad-tools\uploader\Uploader")
|
||||||
dest =Path(r"T:\rad-tools\uploader" )
|
|
||||||
|
|
||||||
|
artifact_dir = project_root / "dist" / ("test" if test else "")
|
||||||
|
if not artifact_dir.exists():
|
||||||
|
artifact_dir = project_root / "dist"
|
||||||
|
|
||||||
|
# Look for a matching artifact. Prefer exe files named Uploader*.exe,
|
||||||
|
# then any .exe, then directories named Uploader* (onedir builds).
|
||||||
|
# Exclude obvious non-artifacts like log files.
|
||||||
|
candidates = []
|
||||||
|
# Uploader*.exe (preferred)
|
||||||
|
candidates.extend(sorted(artifact_dir.glob("Uploader*.exe")))
|
||||||
|
# any .exe
|
||||||
|
candidates.extend(sorted(artifact_dir.glob("*.exe")))
|
||||||
|
# onedir: directories starting with Uploader
|
||||||
|
candidates.extend([p for p in sorted(artifact_dir.glob("Uploader*")) if p.is_dir()])
|
||||||
|
|
||||||
n = 0
|
# Filter out files that look like logs or non-executables
|
||||||
while True:
|
def is_valid_artifact(p: Path) -> bool:
|
||||||
if n > 5:
|
if not p.exists():
|
||||||
break
|
return False
|
||||||
|
if p.is_file():
|
||||||
|
# require .exe for files (Windows builds)
|
||||||
|
return p.suffix.lower() == ".exe"
|
||||||
|
if p.is_dir():
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
s = n if n > 0 else ""
|
valid = [p for p in candidates if is_valid_artifact(p)]
|
||||||
filename = Path(f"cris_tools{s}.exe")
|
if not valid:
|
||||||
|
print(f"No build artifact (.exe or Uploader directory) found in {artifact_dir}, skipping copy")
|
||||||
|
return
|
||||||
|
|
||||||
print(f"Destination: {filename}")
|
artifact = valid[0]
|
||||||
try:
|
|
||||||
shutil.copy(source, dest / filename)
|
|
||||||
break
|
|
||||||
|
|
||||||
except PermissionError:
|
dest_path = Path(dest) if dest else default_dest
|
||||||
n = n + 1
|
dest_path.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
if artifact.is_file():
|
||||||
|
target = dest_path / artifact.name
|
||||||
|
print(f"Copying {artifact} -> {target}")
|
||||||
|
shutil.copy2(artifact, target)
|
||||||
|
else:
|
||||||
|
target = dest_path / artifact.name
|
||||||
|
print(f"Copying directory {artifact} -> {target}")
|
||||||
|
if target.exists():
|
||||||
|
shutil.rmtree(target)
|
||||||
|
shutil.copytree(artifact, target)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
@@ -0,0 +1,21 @@
|
|||||||
|
-----BEGIN CERTIFICATE-----
|
||||||
|
MIIDhzCCAm+gAwIBAgIQHLnU4uDoKaJAaqa5Yi/NoTANBgkqhkiG9w0BAQUFADBW
|
||||||
|
MQswCQYDVQQGEwJHQjERMA8GA1UECBMIQ29ybndhbGwxFTATBgNVBAoTDENvcm53
|
||||||
|
YWxsIE5IUzEdMBsGA1UEAxMUQ29ybndhbGwgSUNUIFJvb3QgQ0EwHhcNMTMwNjI2
|
||||||
|
MDkwNjExWhcNMzMwNjI2MDkxNjA1WjBWMQswCQYDVQQGEwJHQjERMA8GA1UECBMI
|
||||||
|
Q29ybndhbGwxFTATBgNVBAoTDENvcm53YWxsIE5IUzEdMBsGA1UEAxMUQ29ybndh
|
||||||
|
bGwgSUNUIFJvb3QgQ0EwggEiMA0GCSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQCm
|
||||||
|
m8jF64E3wkxhf2BLfD3lT3v9RebGhJ684jtv0A6qL2BTe/Co4SJ/obYiDw82K4Vw
|
||||||
|
1O+jwIg77hVrH/YaCsjPiefxxwL/T0bSURlXq5Q8pCQaHjd1iH1JNY1WfW0Ywkni
|
||||||
|
a7xbm6/2+zwsyTHfhuSU+zmSQxejOX+Ffz6MWZSb+JjppOH6OBKTHGw69JIXcAUu
|
||||||
|
wT3GSAdYyYKgP6i6jfsEeJkq+s/hd6cVxFWPrWJJaGtfIvLZT38vZeitYDYT9Ad0
|
||||||
|
gIQPDyLH35tuNszFWkU5DA/U6xXvWkAV7Mq/EaGSVHckGeQiGAQNmV74KugqpcC/
|
||||||
|
9reOFKEbc6qvDdGaASjZAgMBAAGjUTBPMAsGA1UdDwQEAwIBhjAPBgNVHRMBAf8E
|
||||||
|
BTADAQH/MB0GA1UdDgQWBBShR/WOiN4mvXi2tYLGtbeAaCXE7DAQBgkrBgEEAYI3
|
||||||
|
FQEEAwIBADANBgkqhkiG9w0BAQUFAAOCAQEAEMca1BLnKJIXRZhJUTfLrSWtQDgV
|
||||||
|
T1jGnHFKngPxWUB1u0fVwpnqeuHSWQ2qkNvEZoBQFbJhEny83r8thLamJB1UHEyj
|
||||||
|
yhWrtjif7dne3LwzLtmiMZIKPjeyCa+zzl5YW+lUenQErl48PC7sJc5gq4lZajOo
|
||||||
|
fsEtw5zZLWFqXE/KJNt4vrGhQ0SplphbOzau7VrFIFciiKHHHw3rnV5byjqYDy/Z
|
||||||
|
IJytye488Pi75lBKq8pWW6YbUFmKly5bqF6Q8oicd9y0/9GGmTZGrzp1WbspYF9Y
|
||||||
|
lc2p6bkhfQ1f7Q+8/sF0D/sGfjfo099OlRhc881Mwt5p8mEZtJTxk3SZlQ==
|
||||||
|
-----END CERTIFICATE-----
|
||||||
@@ -0,0 +1,19 @@
|
|||||||
|
-----BEGIN CERTIFICATE-----
|
||||||
|
MIIDJjCCAg6gAwIBAgIIUW6l3kYeVMEwDQYJKoZIhvcNAQELBQAwMTEOMAwGA1UE
|
||||||
|
ChMFQ2lzY28xHzAdBgNVBAMTFkNpc2NvIFVtYnJlbGxhIFJvb3QgQ0EwHhcNMTYw
|
||||||
|
NjI4MTUzNzUzWhcNMzYwNjI4MTUzNzUzWjAxMQ4wDAYDVQQKEwVDaXNjbzEfMB0G
|
||||||
|
A1UEAxMWQ2lzY28gVW1icmVsbGEgUm9vdCBDQTCCASIwDQYJKoZIhvcNAQEBBQAD
|
||||||
|
ggEPADCCAQoCggEBAO7ZjfBSCaz5EMYSiWYoXjHPP/w7xFT4bXa82lOZ9CJJXDQw
|
||||||
|
bZpBdmuqX9UWo769LIAaSUvkYEeZqcTsjrx/7juPKoOErhJY0cPK12LU9PbHXqEd
|
||||||
|
XESIqBjdOC5oiIFHhTAKuuKRlL7rhPYkYhZtgdll4h0FLIG+xNsMVfzJb7z69X8Y
|
||||||
|
vF9r1drLkd7oR2xHuRkXgzeblFVpF+DRF7WXNhLy0By38ZxtClxYUSitdz53W0ic
|
||||||
|
maelG7EyCVNVxARxn5waaphRvki1hkuqqrm3JdlV165zAOdSz3JKzRISQinCTQuT
|
||||||
|
+RK/w0qLsDTyOVO/mEIVWLXu/Z1NtuXgj/jhegcCAwEAAaNCMEAwDgYDVR0PAQH/
|
||||||
|
BAQDAgEGMA8GA1UdEwEB/wQFMAMBAf8wHQYDVR0OBBYEFENzAN4kukAaQFQsfXzV
|
||||||
|
AEiJDHCkMA0GCSqGSIb3DQEBCwUAA4IBAQBIEoceSPZLmo5sLmgDfQA+Fq5BKztL
|
||||||
|
qg8aAvZdrbdMEKEBr1RDB0OAhuPcaaVxZi6Hjyql1N999Zmp8qIw/lLTt3VSTmEa
|
||||||
|
29uPgjdMGLl9KyfZjARiA/PPvPdHTwg7TMJOet+w7P5nWabLNW55+Wc/JzCSFE30
|
||||||
|
+0Kdz/jojxlA/8t0xYLCdS2UK7zC4kuAbojHLJDbIQO3HeEWwVmg4FO89AHVvC4R
|
||||||
|
Y+V0t7SaEradv6tPG9DHX7PLwjQ/Xs95NGDIJTeFwCRqYUlBu9iZjIvKba0e0tST
|
||||||
|
Vuyw2+P2HuWazjBPawGrbfyw+uO3KO4WnNGjMutJJ920o8B5M8gW1+Ye
|
||||||
|
-----END CERTIFICATE-----
|
||||||
@@ -0,0 +1,12 @@
|
|||||||
|
[package]
|
||||||
|
name = "launcher"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition = "2021"
|
||||||
|
|
||||||
|
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
nng = "1.0.1"
|
||||||
|
#rfd = "0.14.1"
|
||||||
|
#toml = "0.8.13"
|
||||||
|
#serde = { version = "1.0", features = ["derive"] }
|
||||||
@@ -0,0 +1,81 @@
|
|||||||
|
use nng::{Protocol, Socket, Error};
|
||||||
|
use std::process::Command;
|
||||||
|
//use rfd::MessageDialog;
|
||||||
|
//use serde::Deserialize;
|
||||||
|
//use std::fs;
|
||||||
|
//use std::path::PathBuf;
|
||||||
|
|
||||||
|
//#[derive(Debug, Deserialize)]
|
||||||
|
//struct Settings {
|
||||||
|
// cris_tools_path: PathBuf,
|
||||||
|
// port: toml::Value,
|
||||||
|
//}
|
||||||
|
|
||||||
|
//fn load_settings() -> Result<Settings, toml::de::Error> {
|
||||||
|
// // Read the entire contents of the file
|
||||||
|
// let contents = fs::read_to_string("uploader.toml")
|
||||||
|
// .expect("Failed to read settings file");
|
||||||
|
//
|
||||||
|
// // Parse the file contents and deserialize into the Settings struct
|
||||||
|
// toml::from_str(&contents)
|
||||||
|
//}
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
//let settings = match load_settings() {
|
||||||
|
// Ok(settings) => settings,
|
||||||
|
// Err(e) => {
|
||||||
|
// eprintln!("Failed to load settings: {:?}", e);
|
||||||
|
// std::process::exit(1);
|
||||||
|
// }
|
||||||
|
//};
|
||||||
|
|
||||||
|
// Create a socket of type REQ (request)
|
||||||
|
let socket = Socket::new(Protocol::Pair0).expect("Failed to create socket");
|
||||||
|
|
||||||
|
// Connect to the NNG server
|
||||||
|
//match socket.dial(format!("tcp://localhost:{}", settings.port).as_str()) {
|
||||||
|
match socket.dial(format!("tcp://localhost:9976").as_str()) {
|
||||||
|
Ok(_) => {
|
||||||
|
println!("Connected to server");
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
println!("Failed to connect to server: {:?}", e);
|
||||||
|
println!("Launching uploader tool");
|
||||||
|
|
||||||
|
let output = Command::new("Uploader.exe")
|
||||||
|
.spawn();
|
||||||
|
|
||||||
|
std::process::exit(1);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Get the command line argument
|
||||||
|
//let arg = std::env::args().nth(1).expect("Missing argument");
|
||||||
|
//let arg = match std::env::args().nth(1) {
|
||||||
|
// Some(arg) => arg,
|
||||||
|
// None => {
|
||||||
|
// eprintln!("Missing argument");
|
||||||
|
// std::process::exit(1);
|
||||||
|
// }
|
||||||
|
//};
|
||||||
|
|
||||||
|
// Send the argument to the server
|
||||||
|
//let message = "run/".to_string() + &arg;
|
||||||
|
let message = "loaded".to_string();
|
||||||
|
println!("Send message: {}", message);
|
||||||
|
socket.send(message.as_bytes()).expect("Failed to send message");
|
||||||
|
|
||||||
|
// Receive the response from the server
|
||||||
|
match socket.recv() {
|
||||||
|
Ok(response) => {
|
||||||
|
let response = String::from_utf8(response.to_vec()).expect("Failed to receive response");
|
||||||
|
println!("Response: {}", response);
|
||||||
|
}
|
||||||
|
Err(Error::TimedOut) => {
|
||||||
|
println!("Failed to receive response: Timeout");
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
println!("Failed to receive response: {:?}", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -14,6 +14,7 @@ from datetime import datetime
|
|||||||
|
|
||||||
from loguru import logger
|
from loguru import logger
|
||||||
import sys
|
import sys
|
||||||
|
import os
|
||||||
|
|
||||||
from anonymiser import Anonymizer
|
from anonymiser import Anonymizer
|
||||||
import pydicom
|
import pydicom
|
||||||
@@ -29,11 +30,29 @@ from local_file_picker import local_file_picker
|
|||||||
from local_folder_picker import local_folder_picker
|
from local_folder_picker import local_folder_picker
|
||||||
|
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
import tempfile
|
||||||
|
import glob
|
||||||
|
|
||||||
from fastapi import Request
|
from fastapi import Request
|
||||||
from fastapi.responses import RedirectResponse
|
from fastapi.responses import RedirectResponse
|
||||||
from starlette.middleware.base import BaseHTTPMiddleware
|
from starlette.middleware.base import BaseHTTPMiddleware
|
||||||
import typer
|
import typer
|
||||||
|
import logging
|
||||||
|
import subprocess
|
||||||
|
import shutil
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
import socket
|
||||||
|
import platform
|
||||||
|
import importlib.util
|
||||||
|
|
||||||
|
# traceback was previously imported for debugging but is unused
|
||||||
|
|
||||||
|
# SSL verification setting. Can be a boolean (True/False) or a path to
|
||||||
|
# a CA bundle (PEM file) used by requests' 'verify' parameter.
|
||||||
|
# Default is True. It can be overridden via CLI options (--ca-bundle,
|
||||||
|
# --insecure) or environment variables: NICE_UPLOADER_CA_BUNDLE,
|
||||||
|
# REQUESTS_CA_BUNDLE or SSL_CERT_FILE.
|
||||||
|
VERIFY_SSL = True
|
||||||
|
|
||||||
SOCKET_PATH = "tcp://localhost:9976"
|
SOCKET_PATH = "tcp://localhost:9976"
|
||||||
|
|
||||||
@@ -57,12 +76,34 @@ LOGGED_IN_USER = "None"
|
|||||||
|
|
||||||
|
|
||||||
REFRESH_TRIGGERED = True
|
REFRESH_TRIGGERED = True
|
||||||
|
RELOAD_ANON_FILES = False
|
||||||
SHUTDOWN = False
|
SHUTDOWN = False
|
||||||
|
|
||||||
DELETE_FILES_ON_UPLOAD = True
|
DELETE_FILES_ON_UPLOAD = True
|
||||||
|
|
||||||
DEBUG = False
|
DEBUG = False
|
||||||
|
|
||||||
|
logging.getLogger('niceGUI').setLevel(logging.INFO)
|
||||||
|
|
||||||
|
# Choose a log file path that works both during development and when the
|
||||||
|
# application is frozen into a single executable (PyInstaller --onefile).
|
||||||
|
try:
|
||||||
|
if getattr(sys, "frozen", False):
|
||||||
|
# When frozen by PyInstaller, data files are extracted to a temporary
|
||||||
|
# folder available as sys._MEIPASS. Fall back to the executable
|
||||||
|
# directory when _MEIPASS is not present.
|
||||||
|
meipass = getattr(sys, "_MEIPASS", None)
|
||||||
|
if meipass:
|
||||||
|
base_path = Path(meipass)
|
||||||
|
else:
|
||||||
|
base_path = Path(sys.executable).resolve().parent
|
||||||
|
else:
|
||||||
|
base_path = Path(__file__).resolve().parent
|
||||||
|
except Exception:
|
||||||
|
base_path = Path.cwd()
|
||||||
|
|
||||||
|
log_path = base_path / "uploader.log"
|
||||||
|
logger.add(str(log_path), rotation="10 MB")
|
||||||
|
|
||||||
if DEBUG:
|
if DEBUG:
|
||||||
BASE_SITE_URL = "http://localhost:8000"
|
BASE_SITE_URL = "http://localhost:8000"
|
||||||
@@ -98,6 +139,7 @@ async def read_nng_messages():
|
|||||||
ui.notify("Export done", color="positive")
|
ui.notify("Export done", color="positive")
|
||||||
global REFRESH_TRIGGERED
|
global REFRESH_TRIGGERED
|
||||||
REFRESH_TRIGGERED = True
|
REFRESH_TRIGGERED = True
|
||||||
|
await sub.asend(b"done")
|
||||||
logger.debug("loaded")
|
logger.debug("loaded")
|
||||||
case _:
|
case _:
|
||||||
logger.error("invalid message")
|
logger.error("invalid message")
|
||||||
@@ -110,6 +152,7 @@ async def read_nng_messages():
|
|||||||
# This shouldn't be necessary
|
# This shouldn't be necessary
|
||||||
global SHUTDOWN
|
global SHUTDOWN
|
||||||
SHUTDOWN = True
|
SHUTDOWN = True
|
||||||
|
logger.debug("Triggering shutdown due to socket in use")
|
||||||
return
|
return
|
||||||
|
|
||||||
# asyncio.create_task(read_nng_messages())
|
# asyncio.create_task(read_nng_messages())
|
||||||
@@ -188,7 +231,11 @@ async def reload_anonymized_start(progress_bar, queue):
|
|||||||
)
|
)
|
||||||
progress_bar.visible = False
|
progress_bar.visible = False
|
||||||
|
|
||||||
ui.notify(f"Files reloaded: {len(loaded_files)}, [Series: {len(loaded_series)}] ", color="positive")
|
for client in Client.instances.values():
|
||||||
|
if not client.has_socket_connection:
|
||||||
|
continue
|
||||||
|
with client:
|
||||||
|
ui.notify(f"Files reloaded: {len(loaded_files)}, [Series: {len(loaded_series)}] ", color="positive")
|
||||||
# load_series_view(loaded_series, loaded_series_data)
|
# load_series_view(loaded_series, loaded_series_data)
|
||||||
loaded_series_ui.refresh()
|
loaded_series_ui.refresh()
|
||||||
loaded_files_ui.refresh()
|
loaded_files_ui.refresh()
|
||||||
@@ -273,17 +320,21 @@ def load_files(q: Queue, src_path: Path | None = None):
|
|||||||
dataset, output_file = anonymizer.anonymize_file(file, remove_original=True)
|
dataset, output_file = anonymizer.anonymize_file(file, remove_original=True)
|
||||||
# with processed_files:
|
# with processed_files:
|
||||||
# ui.item(f"{datetime.now().strftime('%H:%M:%S')} - {file.name} -> {output_file.name}")
|
# ui.item(f"{datetime.now().strftime('%H:%M:%S')} - {file.name} -> {output_file.name}")
|
||||||
|
try:
|
||||||
|
series_description = dataset.SeriesDescription
|
||||||
|
except AttributeError:
|
||||||
|
series_description = ""
|
||||||
|
|
||||||
loaded_files[output_file] = (
|
loaded_files[output_file] = (
|
||||||
dataset.StudyDescription,
|
dataset.StudyDescription,
|
||||||
dataset.SeriesDescription,
|
series_description,
|
||||||
str(dataset.SeriesInstanceUID),
|
str(dataset.SeriesInstanceUID),
|
||||||
)
|
)
|
||||||
|
|
||||||
loaded_series[dataset.SeriesInstanceUID].append(output_file)
|
loaded_series[dataset.SeriesInstanceUID].append(output_file)
|
||||||
loaded_series_data[dataset.SeriesInstanceUID] = (
|
loaded_series_data[dataset.SeriesInstanceUID] = (
|
||||||
dataset.StudyDescription,
|
dataset.StudyDescription,
|
||||||
dataset.SeriesDescription,
|
series_description
|
||||||
)
|
)
|
||||||
|
|
||||||
q.put_nowait((n + 1) / to_process_len)
|
q.put_nowait((n + 1) / to_process_len)
|
||||||
@@ -333,15 +384,147 @@ def reload_anonymized(q: Queue):
|
|||||||
|
|
||||||
|
|
||||||
def clear_anonymized_files():
|
def clear_anonymized_files():
|
||||||
|
global RELOAD_ANON_FILES
|
||||||
for file in ANON_PATH.iterdir():
|
for file in ANON_PATH.iterdir():
|
||||||
if file.is_file():
|
if file.is_file():
|
||||||
file.unlink()
|
file.unlink()
|
||||||
|
|
||||||
logger.debug("Deleted all files in ANON_PATH")
|
logger.debug("Deleted all files in ANON_PATH")
|
||||||
|
|
||||||
|
RELOAD_ANON_FILES = True
|
||||||
|
|
||||||
return
|
return
|
||||||
|
|
||||||
|
|
||||||
|
def ssl_check_cmd(url: str) -> str:
|
||||||
|
"""Run openssl s_client -showcerts against the given URL host and return output.
|
||||||
|
|
||||||
|
Falls back to a minimal Python-based peer certificate fetch if openssl is not available.
|
||||||
|
"""
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
|
try:
|
||||||
|
parsed = urlparse(url)
|
||||||
|
host = parsed.hostname or url
|
||||||
|
port = parsed.port or (443 if parsed.scheme in ("https", "") else 80)
|
||||||
|
except Exception:
|
||||||
|
host = url
|
||||||
|
port = 443
|
||||||
|
|
||||||
|
logger.debug(f"ssl_check_cmd: host={host} port={port}")
|
||||||
|
openssl_path = shutil.which("openssl")
|
||||||
|
if openssl_path:
|
||||||
|
cmd = [openssl_path, "s_client", "-showcerts", "-connect", f"{host}:{port}", "-servername", host]
|
||||||
|
logger.debug(f"Running openssl: {' '.join(cmd)}")
|
||||||
|
try:
|
||||||
|
# Provide a small stdin so s_client can complete instead of
|
||||||
|
# waiting for interactive input. Capture both stdout and stderr.
|
||||||
|
proc = subprocess.run(cmd, input='\n', capture_output=True, text=True, timeout=60)
|
||||||
|
out = proc.stdout + "\n" + proc.stderr
|
||||||
|
logger.debug(f"OpenSSL finished, {len(out)} bytes returned, rc={proc.returncode}")
|
||||||
|
# Some older OpenSSL builds (or default s_client options) may
|
||||||
|
# negotiate an incompatible protocol version with modern servers
|
||||||
|
# and return a short error like 'tlsv1 alert protocol version'. If
|
||||||
|
# that appears, try again forcing TLS1.2 which most servers still
|
||||||
|
# support. This helps when the openssl on PATH is old (eg bundled
|
||||||
|
# with other software) and defaults to legacy negotiation.
|
||||||
|
if "tlsv1 alert protocol version" in out or "SSL23_GET_SERVER_HELLO" in out:
|
||||||
|
logger.debug("OpenSSL reported protocol version issue; retrying with -tls1_2")
|
||||||
|
try:
|
||||||
|
cmd2 = cmd + ["-tls1_2"]
|
||||||
|
proc2 = subprocess.run(cmd2, input='\n', capture_output=True, text=True, timeout=30)
|
||||||
|
out2 = proc2.stdout + "\n" + proc2.stderr
|
||||||
|
logger.debug(f"OpenSSL retry finished, {len(out2)} bytes returned, rc={proc2.returncode}")
|
||||||
|
return out + "\n--- retry with -tls1_2 ---\n" + out2
|
||||||
|
except Exception:
|
||||||
|
logger.exception("OpenSSL tls1_2 retry failed")
|
||||||
|
return out + "\nNote: OpenSSL appears to be too old to negotiate modern TLS.\nConsider installing a newer OpenSSL (eg Git for Windows or a Win64 OpenSSL build).\n"
|
||||||
|
|
||||||
|
return out
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
logger.debug("OpenSSL timed out")
|
||||||
|
return (
|
||||||
|
"OpenSSL check timed out after 60s\n"
|
||||||
|
"This can happen if the connection is blocked by a proxy or firewall. "
|
||||||
|
"Try running the openssl command manually or check your network/proxy settings.\n"
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception("OpenSSL check failed")
|
||||||
|
return f"OpenSSL check failed: {e}\n"
|
||||||
|
else:
|
||||||
|
# Fallback: use Python ssl to fetch the peer certificate (leaf only).
|
||||||
|
# If a proxy is configured, perform an HTTP CONNECT to the proxy first
|
||||||
|
# and then do TLS over the established tunnel so the result matches the
|
||||||
|
# path used by the running Python process.
|
||||||
|
import ssl
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
import base64
|
||||||
|
|
||||||
|
def fetch_via_proxy(proxy_url: str) -> str:
|
||||||
|
parsed = urlparse(proxy_url)
|
||||||
|
proxy_host = parsed.hostname
|
||||||
|
proxy_port = parsed.port or (443 if parsed.scheme == "https" else 80)
|
||||||
|
|
||||||
|
auth_header = None
|
||||||
|
if parsed.username:
|
||||||
|
user = parsed.username
|
||||||
|
pwd = parsed.password or ""
|
||||||
|
auth_header = base64.b64encode(f"{user}:{pwd}".encode("utf-8")).decode("ascii")
|
||||||
|
|
||||||
|
s = socket.create_connection((proxy_host, proxy_port), timeout=15)
|
||||||
|
try:
|
||||||
|
req = f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n"
|
||||||
|
if auth_header:
|
||||||
|
req += f"Proxy-Authorization: Basic {auth_header}\r\n"
|
||||||
|
req += "\r\n"
|
||||||
|
s.sendall(req.encode("ascii"))
|
||||||
|
|
||||||
|
# read headers
|
||||||
|
resp = b""
|
||||||
|
while b"\r\n\r\n" not in resp:
|
||||||
|
chunk = s.recv(4096)
|
||||||
|
if not chunk:
|
||||||
|
break
|
||||||
|
resp += chunk
|
||||||
|
if len(resp) > 65536:
|
||||||
|
break
|
||||||
|
|
||||||
|
first_line = resp.split(b"\r\n", 1)[0].decode("ascii", errors="ignore")
|
||||||
|
if not first_line.startswith("HTTP/") or ("200" not in first_line):
|
||||||
|
raise RuntimeError(f"Proxy CONNECT failed: {first_line}")
|
||||||
|
|
||||||
|
ctx = ssl.create_default_context()
|
||||||
|
with ctx.wrap_socket(s, server_hostname=host) as ss:
|
||||||
|
ss.settimeout(15)
|
||||||
|
der = ss.getpeercert(True)
|
||||||
|
return ssl.DER_cert_to_PEM_cert(der)
|
||||||
|
except Exception:
|
||||||
|
s.close()
|
||||||
|
raise
|
||||||
|
|
||||||
|
logger.debug("OpenSSL not found; using Python ssl fallback (proxy-aware)")
|
||||||
|
# Prefer HTTPS_PROXY then HTTP_PROXY
|
||||||
|
proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") or os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy")
|
||||||
|
try:
|
||||||
|
if proxy:
|
||||||
|
logger.debug(f"Using proxy for ssl check: {proxy}")
|
||||||
|
pem = fetch_via_proxy(proxy)
|
||||||
|
else:
|
||||||
|
ctx = ssl.create_default_context()
|
||||||
|
s = socket.socket(socket.AF_INET)
|
||||||
|
with ctx.wrap_socket(s, server_hostname=host) as ss:
|
||||||
|
ss.settimeout(15)
|
||||||
|
ss.connect((host, port))
|
||||||
|
der = ss.getpeercert(True)
|
||||||
|
pem = ssl.DER_cert_to_PEM_cert(der)
|
||||||
|
|
||||||
|
logger.debug("Python ssl fallback succeeded")
|
||||||
|
return pem
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception("Python SSL fallback failed")
|
||||||
|
return f"Python SSL fallback failed: {e}\n"
|
||||||
|
|
||||||
|
|
||||||
def upload_files(q, rqst):
|
def upload_files(q, rqst):
|
||||||
# global rqst#, uploaded_files
|
# global rqst#, uploaded_files
|
||||||
|
|
||||||
@@ -380,6 +563,7 @@ def upload_files(q, rqst):
|
|||||||
f"{BASE_SITE_URL}/api/atlas/upload_dicom",
|
f"{BASE_SITE_URL}/api/atlas/upload_dicom",
|
||||||
# data=data,
|
# data=data,
|
||||||
files=files,
|
files=files,
|
||||||
|
verify=VERIFY_SSL
|
||||||
)
|
)
|
||||||
|
|
||||||
logger.debug(f"Resp: {resp}")
|
logger.debug(f"Resp: {resp}")
|
||||||
@@ -428,6 +612,49 @@ def upload_files(q, rqst):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def perform_login(user: str, pw: str) -> dict:
|
||||||
|
"""Module-level blocking login function suitable for run.cpu_bound.
|
||||||
|
|
||||||
|
Returns a dict with keys:
|
||||||
|
- error: str on error
|
||||||
|
- session: requests.Session on success
|
||||||
|
- token: csrf token if found
|
||||||
|
- success: bool
|
||||||
|
"""
|
||||||
|
s = requests.session()
|
||||||
|
try:
|
||||||
|
r = s.get(LOGIN_URL, verify=VERIFY_SSL, timeout=10)
|
||||||
|
except Exception as e:
|
||||||
|
logger.debug(e)
|
||||||
|
return {"error": str(e)}
|
||||||
|
|
||||||
|
try:
|
||||||
|
token = BeautifulSoup(r.content, "html.parser").find("input").attrs.get("value")
|
||||||
|
except Exception:
|
||||||
|
token = None
|
||||||
|
|
||||||
|
data = {
|
||||||
|
"username": user,
|
||||||
|
"password": pw,
|
||||||
|
"csrfmiddlewaretoken": token,
|
||||||
|
"next": "/",
|
||||||
|
}
|
||||||
|
|
||||||
|
s.headers["Referer"] = LOGIN_URL
|
||||||
|
|
||||||
|
try:
|
||||||
|
r2 = s.post(LOGIN_URL, data=data, verify=VERIFY_SSL, timeout=10)
|
||||||
|
except Exception as e:
|
||||||
|
logger.debug(e)
|
||||||
|
return {"error": str(e)}
|
||||||
|
|
||||||
|
soup = BeautifulSoup(r2.content, "html.parser")
|
||||||
|
if soup.find("button") and soup.find("button").find(string="Login"):
|
||||||
|
return {"session": s, "success": False}
|
||||||
|
else:
|
||||||
|
return {"session": s, "success": True, "token": token}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@@ -436,65 +663,30 @@ def login() -> Optional[RedirectResponse]:
|
|||||||
dark = ui.dark_mode()
|
dark = ui.dark_mode()
|
||||||
dark.enable()
|
dark.enable()
|
||||||
|
|
||||||
def try_login() -> (
|
async def try_login() -> None:
|
||||||
None
|
# Run blocking network calls off the UI thread to keep the page responsive.
|
||||||
): # local function to avoid passing username and password as arguments
|
|
||||||
# Because I haven't gotten around to implementing a proper login system yet
|
|
||||||
# we just log in via requests and store teh session in the global rqst variable
|
|
||||||
user = username.value
|
user = username.value
|
||||||
pw = password.value
|
pw = password.value
|
||||||
|
|
||||||
global rqst, LOGIN_SUCCESS, LOGGED_IN_USER
|
global rqst, LOGIN_SUCCESS, LOGGED_IN_USER
|
||||||
rqst = requests.session()
|
|
||||||
try:
|
# Use the module-level perform_login function (picklable) to avoid
|
||||||
rsp = rqst.get(LOGIN_URL)
|
# multiprocessing pickling errors when using run.cpu_bound.
|
||||||
except requests.exceptions.ConnectionError:
|
result = await run.cpu_bound(perform_login, user, pw)
|
||||||
|
|
||||||
|
if result.get("error"):
|
||||||
ui.notify("Connection error!", color="negative")
|
ui.notify("Connection error!", color="negative")
|
||||||
|
logger.debug(result.get("error"))
|
||||||
return
|
return
|
||||||
|
|
||||||
token = (
|
if result.get("success"):
|
||||||
BeautifulSoup(rsp.content, "html.parser").find("input").attrs["value"]
|
rqst = result.get("session")
|
||||||
) # , attr={"name": "csrfmiddlewaretoken"}).attrs("value")
|
token = result.get("token")
|
||||||
|
if token:
|
||||||
# token = rsp.cookies["csrftoken"]
|
rqst.headers["X-CSRFToken"] = token
|
||||||
# header = {"X-CSRFToken": token}
|
|
||||||
# cookies = {"csrftoken": token}
|
|
||||||
|
|
||||||
data = {
|
|
||||||
"username": user,
|
|
||||||
"password": pw,
|
|
||||||
"csrfmiddlewaretoken": token,
|
|
||||||
"next": "/",
|
|
||||||
}
|
|
||||||
|
|
||||||
rqst.headers["Referer"] = LOGIN_URL
|
|
||||||
|
|
||||||
rsp = rqst.post(
|
|
||||||
LOGIN_URL,
|
|
||||||
data=data,
|
|
||||||
# headers=header,
|
|
||||||
# cookies=cookies
|
|
||||||
)
|
|
||||||
|
|
||||||
print(rsp.content)
|
|
||||||
|
|
||||||
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")
|
|
||||||
LOGIN_SUCCESS = True
|
LOGIN_SUCCESS = True
|
||||||
LOGGED_IN_USER = user
|
LOGGED_IN_USER = user
|
||||||
user_info_ui.refresh()
|
user_info_ui.refresh()
|
||||||
# user_label.clear()
|
|
||||||
# user_label.text = f"User: {user}"
|
|
||||||
# print(rqst.headers)
|
|
||||||
|
|
||||||
rqst.headers["X-CSRFToken"] = token
|
|
||||||
|
|
||||||
if LOGIN_SUCCESS:
|
|
||||||
ui.notify("Logged in!", color="positive")
|
ui.notify("Logged in!", color="positive")
|
||||||
ui.navigate.to("/")
|
ui.navigate.to("/")
|
||||||
else:
|
else:
|
||||||
@@ -513,6 +705,7 @@ def login() -> Optional[RedirectResponse]:
|
|||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
@logger.catch
|
||||||
@ui.page("/")
|
@ui.page("/")
|
||||||
def main_page():
|
def main_page():
|
||||||
async def watch_for_shutdown():
|
async def watch_for_shutdown():
|
||||||
@@ -537,13 +730,15 @@ def main_page():
|
|||||||
globals(), "LOGIN_SUCCESS", backward=lambda v: not v
|
globals(), "LOGIN_SUCCESS", backward=lambda v: not v
|
||||||
)
|
)
|
||||||
|
|
||||||
updates = ui.list()
|
# placeholder for update list (previously assigned but unused)
|
||||||
|
|
||||||
loaded_series_ui()
|
loaded_series_ui()
|
||||||
|
|
||||||
# typer_app()
|
# typer_app()
|
||||||
# Create a queue to communicate with the heavy computation process
|
# Create a queue to communicate with the heavy computation process
|
||||||
|
logger.debug("PRE QUEUE")
|
||||||
queue = Manager().Queue()
|
queue = Manager().Queue()
|
||||||
|
logger.debug("POST QUEUE")
|
||||||
# Update the progress bar on the main process
|
# Update the progress bar on the main process
|
||||||
ui.timer(
|
ui.timer(
|
||||||
0.1,
|
0.1,
|
||||||
@@ -551,6 +746,7 @@ def main_page():
|
|||||||
queue.get() if not queue.empty() else progressbar.value
|
queue.get() if not queue.empty() else progressbar.value
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
logger.debug("POST TImER")
|
||||||
|
|
||||||
# Create a queue to communicate with the heavy computation process
|
# Create a queue to communicate with the heavy computation process
|
||||||
upload_queue = Manager().Queue()
|
upload_queue = Manager().Queue()
|
||||||
@@ -561,6 +757,7 @@ def main_page():
|
|||||||
upload_queue.get() if not upload_queue.empty() else upload_progressbar.value
|
upload_queue.get() if not upload_queue.empty() else upload_progressbar.value
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
logger.debug("POST TImER")
|
||||||
|
|
||||||
upload_progress = ui.row().classes("w-full place-content-center bg-blue-900")
|
upload_progress = ui.row().classes("w-full place-content-center bg-blue-900")
|
||||||
|
|
||||||
@@ -616,17 +813,46 @@ def main_page():
|
|||||||
ui.button("Clear anon path", on_click=clear_anonymized_files).tooltip(
|
ui.button("Clear anon path", on_click=clear_anonymized_files).tooltip(
|
||||||
"Delete all files in ANON_PATH"
|
"Delete all files in ANON_PATH"
|
||||||
)
|
)
|
||||||
|
async def ssl_check_ui() -> None:
|
||||||
|
# Show a running dialog immediately so user sees progress
|
||||||
|
with ui.dialog() as running_dlg:
|
||||||
|
with ui.card().classes("p-4"):
|
||||||
|
ui.label("SSL check running...").classes("text-h6")
|
||||||
|
running_dlg.open()
|
||||||
|
|
||||||
|
try:
|
||||||
|
logger.debug("Starting SSL check via UI")
|
||||||
|
output = await run.cpu_bound(ssl_check_cmd, BASE_SITE_URL)
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception("SSL check failed")
|
||||||
|
running_dlg.close()
|
||||||
|
ui.notify(f"SSL check failed: {e}", color="negative")
|
||||||
|
return
|
||||||
|
|
||||||
|
# close the running dialog and show the results in a new dialog
|
||||||
|
running_dlg.close()
|
||||||
|
with ui.dialog() as d:
|
||||||
|
with ui.card().classes("p-4"):
|
||||||
|
ui.label("SSL Check output").classes("text-h6")
|
||||||
|
ui.code(output).props("white-space: pre-wrap")
|
||||||
|
ui.button("Close", on_click=d.close)
|
||||||
|
d.open()
|
||||||
|
|
||||||
|
ui.button("SSL check", on_click=ssl_check_ui).tooltip("Run openssl s_client -showcerts against the configured site and show output")
|
||||||
|
|
||||||
def shutdown_app():
|
def shutdown_app():
|
||||||
app.shutdown()
|
app.shutdown()
|
||||||
ui.run_javascript("window.close()")
|
ui.run_javascript("window.close()")
|
||||||
|
|
||||||
async def watch_for_refresh():
|
async def watch_for_refresh():
|
||||||
global REFRESH_TRIGGERED
|
global REFRESH_TRIGGERED, RELOAD_ANON_FILES
|
||||||
while not app.is_stopped:
|
while not app.is_stopped:
|
||||||
if REFRESH_TRIGGERED:
|
if REFRESH_TRIGGERED:
|
||||||
REFRESH_TRIGGERED = False
|
REFRESH_TRIGGERED = False
|
||||||
await load_files_start(anon_progress, queue)
|
await load_files_start(anon_progress, queue)
|
||||||
|
if RELOAD_ANON_FILES:
|
||||||
|
RELOAD_ANON_FILES = False
|
||||||
|
await reload_anonymized_start(anon_progress, queue)
|
||||||
await asyncio.sleep(1)
|
await asyncio.sleep(1)
|
||||||
|
|
||||||
asyncio.create_task(watch_for_refresh())
|
asyncio.create_task(watch_for_refresh())
|
||||||
@@ -647,36 +873,291 @@ def main_page():
|
|||||||
ui.button("About", on_click=about_dialog.open).props("outline color=white")
|
ui.button("About", on_click=about_dialog.open).props("outline color=white")
|
||||||
|
|
||||||
|
|
||||||
|
@logger.catch
|
||||||
def launch_app(
|
def launch_app(
|
||||||
work_dir: Path = typer.Option(WORK_DIR, help="Working directory"),
|
work_dir: Path = typer.Option(WORK_DIR, help="Working directory"),
|
||||||
nng: bool = typer.Option(True, help="Use nng"),
|
nng: bool = typer.Option(True, help="Use nng"),
|
||||||
native_mode: bool = typer.Option(False, help="Use native mode"),
|
native_mode: bool = typer.Option(False, help="Use native mode"),
|
||||||
|
debug: bool = typer.Option(False, help="Debug mode"),
|
||||||
|
ca_bundle: Optional[Path] = typer.Option(
|
||||||
|
None, help="Path to a CA bundle (PEM file) to verify HTTPS connections"
|
||||||
|
),
|
||||||
|
insecure: bool = typer.Option(False, help="Disable SSL verification (insecure)"),
|
||||||
):
|
):
|
||||||
global WORK_DIR, EXPORT_PATH, PROCESS_PATH, ANON_PATH
|
global WORK_DIR, EXPORT_PATH, PROCESS_PATH, ANON_PATH, DEBUG
|
||||||
|
|
||||||
WORK_DIR = work_dir
|
WORK_DIR = work_dir
|
||||||
EXPORT_PATH = WORK_DIR / Path("export/")
|
EXPORT_PATH = WORK_DIR / Path("export/")
|
||||||
PROCESS_PATH = WORK_DIR / Path("to_process/")
|
PROCESS_PATH = WORK_DIR / Path("to_process/")
|
||||||
ANON_PATH = WORK_DIR / Path("anon/")
|
ANON_PATH = WORK_DIR / Path("anon/")
|
||||||
|
DEBUG = debug
|
||||||
|
|
||||||
logger.debug(f"Work dir: {WORK_DIR}")
|
logger.debug(f"Work dir: {WORK_DIR}")
|
||||||
|
|
||||||
|
# Configure SSL verification behaviour. VERIFY_SSL can be:
|
||||||
|
# - True (default): use system certs
|
||||||
|
# - False: disable verification (insecure)
|
||||||
|
# - path (str): path to CA bundle PEM file used by requests
|
||||||
|
global VERIFY_SSL
|
||||||
|
|
||||||
|
env_ca = (
|
||||||
|
os.environ.get("NICE_UPLOADER_CA_BUNDLE")
|
||||||
|
or os.environ.get("REQUESTS_CA_BUNDLE")
|
||||||
|
or os.environ.get("SSL_CERT_FILE")
|
||||||
|
)
|
||||||
|
|
||||||
|
if insecure:
|
||||||
|
VERIFY_SSL = False
|
||||||
|
logger.warning(
|
||||||
|
"SSL verification disabled (insecure). Only use for troubleshooting in trusted networks."
|
||||||
|
)
|
||||||
|
elif ca_bundle is not None:
|
||||||
|
VERIFY_SSL = str(ca_bundle)
|
||||||
|
logger.info(f"Using CA bundle from CLI: {VERIFY_SSL}")
|
||||||
|
elif env_ca:
|
||||||
|
VERIFY_SSL = env_ca
|
||||||
|
logger.info(f"Using CA bundle from environment: {VERIFY_SSL}")
|
||||||
|
else:
|
||||||
|
VERIFY_SSL = True
|
||||||
|
|
||||||
|
# If verification is disabled, suppress urllib3 insecure warnings to keep logs clean
|
||||||
|
if VERIFY_SSL is False:
|
||||||
|
try:
|
||||||
|
import urllib3
|
||||||
|
|
||||||
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# If a bundled cert.pem was included in the frozen app, prefer that
|
||||||
|
# when the user hasn't explicitly provided a CA bundle or disabled
|
||||||
|
# verification. This lets builders ship a corporate CA with the app.
|
||||||
|
try:
|
||||||
|
# Prefer an entire bundled cert/ folder (so users can include many
|
||||||
|
# certificate files). If present, concatenate .pem/.cer/.crt files
|
||||||
|
# into a temporary combined bundle and use that. Otherwise fall back
|
||||||
|
# to a single cert.pem file if present.
|
||||||
|
bundled_dir = base_path / "cert"
|
||||||
|
if bundled_dir.exists() and bundled_dir.is_dir() and VERIFY_SSL is True:
|
||||||
|
# Collect candidate cert files
|
||||||
|
patterns = ["*.pem", "*.cer", "*.crt"]
|
||||||
|
files = []
|
||||||
|
for p in patterns:
|
||||||
|
files.extend(sorted(bundled_dir.glob(p)))
|
||||||
|
|
||||||
|
if files:
|
||||||
|
# Create a temporary combined CA bundle file (persisted)
|
||||||
|
tmp = tempfile.NamedTemporaryFile(delete=False, prefix="nice_uploader_ca_", suffix=".pem")
|
||||||
|
included = []
|
||||||
|
try:
|
||||||
|
import base64
|
||||||
|
|
||||||
|
for f in files:
|
||||||
|
try:
|
||||||
|
with open(f, "rb") as fh:
|
||||||
|
data = fh.read()
|
||||||
|
# If the file already contains a PEM block, write as-is
|
||||||
|
if b"-----BEGIN CERTIFICATE-----" in data:
|
||||||
|
tmp.write(data)
|
||||||
|
tmp.write(b"\n")
|
||||||
|
included.append(str(f))
|
||||||
|
else:
|
||||||
|
# Assume DER/binary and convert to PEM by base64-encoding
|
||||||
|
b64 = base64.encodebytes(data)
|
||||||
|
tmp.write(b"-----BEGIN CERTIFICATE-----\n")
|
||||||
|
tmp.write(b64)
|
||||||
|
tmp.write(b"-----END CERTIFICATE-----\n")
|
||||||
|
included.append(str(f) + " (converted from DER)")
|
||||||
|
except Exception:
|
||||||
|
# skip unreadable files
|
||||||
|
logger.debug(f"Skipping unreadable cert file: {f}")
|
||||||
|
tmp.flush()
|
||||||
|
tmp.close()
|
||||||
|
VERIFY_SSL = str(Path(tmp.name))
|
||||||
|
logger.info(f"Using bundled CA bundle dir combined to {VERIFY_SSL}")
|
||||||
|
logger.debug(f"Bundled cert files: {included}")
|
||||||
|
except Exception as e:
|
||||||
|
try:
|
||||||
|
tmp.close()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
logger.debug(f"Failed to build combined CA bundle: {e}")
|
||||||
|
else:
|
||||||
|
bundled = base_path / "cert.pem"
|
||||||
|
if bundled.exists() and VERIFY_SSL is True:
|
||||||
|
VERIFY_SSL = str(bundled)
|
||||||
|
logger.info(f"Using bundled CA bundle at {VERIFY_SSL}")
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# When running as a frozen executable, the native/webview backend
|
||||||
|
# (which depends on pythonnet) can fail to load inside the bundled
|
||||||
|
# runtime. Disable native mode automatically for frozen builds so the
|
||||||
|
# app falls back to launching in the browser and remains usable.
|
||||||
|
if getattr(sys, "frozen", False):
|
||||||
|
if native_mode:
|
||||||
|
logger.debug("Running frozen executable: disabling native mode to avoid pythonnet loader issues")
|
||||||
|
native_mode = False
|
||||||
|
|
||||||
if nng:
|
if nng:
|
||||||
|
socket_error = False
|
||||||
try:
|
try:
|
||||||
with pynng.Pair0(listen=SOCKET_PATH) as sub:
|
with pynng.Pair0(listen=SOCKET_PATH) as sub:
|
||||||
pass
|
pass
|
||||||
except pynng.exceptions.AddressInUse:
|
except pynng.exceptions.AddressInUse:
|
||||||
logger.debug("Address in use")
|
logger.debug("Address in use")
|
||||||
with pynng.Pair0(dial=SOCKET_PATH) as sub:
|
# Try to detect which process is holding the TCP port (helpful on Windows)
|
||||||
sub.send(b"loaded")
|
try:
|
||||||
sys.exit(0)
|
port_str = SOCKET_PATH.split(':')[-1]
|
||||||
|
port = int(port_str)
|
||||||
|
except Exception:
|
||||||
|
port = None
|
||||||
|
|
||||||
app.on_startup(read_nng_messages)
|
if port:
|
||||||
|
def find_process_using_port(port: int):
|
||||||
|
"""Return a list of (pid, name) tuples for processes listening on TCP port.
|
||||||
|
|
||||||
|
Uses psutil if available; otherwise falls back to parsing `netstat -ano` and
|
||||||
|
resolving PIDs with `tasklist` on Windows. Returns empty list if none found.
|
||||||
|
"""
|
||||||
|
results = []
|
||||||
|
# Prefer psutil (fast, cross-platform) when available
|
||||||
|
try:
|
||||||
|
import psutil
|
||||||
|
|
||||||
|
for conn in psutil.net_connections(kind='tcp'):
|
||||||
|
laddr = conn.laddr
|
||||||
|
if not laddr:
|
||||||
|
continue
|
||||||
|
# laddr can be an address tuple (ip, port)
|
||||||
|
try:
|
||||||
|
conn_port = laddr.port
|
||||||
|
except Exception:
|
||||||
|
# on some platforms laddr may be a string
|
||||||
|
conn_port = int(str(laddr).split(':')[-1])
|
||||||
|
if conn_port == port and conn.status in ('LISTEN', 'LISTENING'):
|
||||||
|
pid = conn.pid
|
||||||
|
name = None
|
||||||
|
try:
|
||||||
|
if pid:
|
||||||
|
p = psutil.Process(pid)
|
||||||
|
name = p.name()
|
||||||
|
except Exception:
|
||||||
|
name = None
|
||||||
|
results.append((pid, name))
|
||||||
|
return results
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
# Fallback: use netstat (Windows) and tasklist to resolve PIDs
|
||||||
|
try:
|
||||||
|
if platform.system().lower().startswith('win'):
|
||||||
|
cmd = ['netstat', '-ano', '-p', 'tcp']
|
||||||
|
proc = subprocess.run(cmd, capture_output=True, text=True)
|
||||||
|
out = proc.stdout.splitlines()
|
||||||
|
pids = set()
|
||||||
|
for line in out:
|
||||||
|
parts = line.split()
|
||||||
|
# Expected format: Proto Local Address Foreign Address State PID
|
||||||
|
if len(parts) >= 5:
|
||||||
|
local = parts[1]
|
||||||
|
state = parts[3]
|
||||||
|
pid = parts[4]
|
||||||
|
# local can be like '0.0.0.0:9976' or '[::]:9976'
|
||||||
|
if ':' in local and state.upper() in ('LISTEN', 'LISTENING'):
|
||||||
|
try:
|
||||||
|
local_port = int(local.rsplit(':', 1)[1])
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
if local_port == port:
|
||||||
|
try:
|
||||||
|
pid_i = int(pid)
|
||||||
|
pids.add(pid_i)
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
for pid in pids:
|
||||||
|
# Resolve process name via tasklist
|
||||||
|
try:
|
||||||
|
tl = subprocess.run(['tasklist', '/FI', f'PID eq {pid}', '/FO', 'CSV', '/NH'], capture_output=True, text=True)
|
||||||
|
line = tl.stdout.strip()
|
||||||
|
if line and '"' in line:
|
||||||
|
# CSV like "process.exe","1234","..."
|
||||||
|
cols = [c.strip('"') for c in line.split(',')]
|
||||||
|
name = cols[0] if cols else None
|
||||||
|
else:
|
||||||
|
name = None
|
||||||
|
except Exception:
|
||||||
|
name = None
|
||||||
|
results.append((pid, name))
|
||||||
|
return results
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
|
||||||
|
return results
|
||||||
|
|
||||||
|
holders = find_process_using_port(port)
|
||||||
|
if holders:
|
||||||
|
for pid, name in holders:
|
||||||
|
logger.error(f"Port {port} appears in use by PID={pid} name={name}")
|
||||||
|
else:
|
||||||
|
logger.debug(f"No process detected listening on port {port} via psutil/netstat")
|
||||||
|
try:
|
||||||
|
with pynng.Pair0(dial=SOCKET_PATH, send_timeout=4000, recv_timeout=4000) as sub:
|
||||||
|
sub.send(b"loaded")
|
||||||
|
msg = sub.recv()
|
||||||
|
print(msg)
|
||||||
|
sys.exit(0)
|
||||||
|
except pynng.exceptions.Timeout:
|
||||||
|
logger.error(f"Socket timeout: {SOCKET_PATH}")
|
||||||
|
# Happens when the socket is locked
|
||||||
|
socket_error = True
|
||||||
|
pass
|
||||||
|
|
||||||
|
if not socket_error:
|
||||||
|
app.on_startup(read_nng_messages)
|
||||||
|
|
||||||
app.on_exception(logger.debug)
|
app.on_exception(logger.debug)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
ui.run(reload=False, show=True, port=native.find_open_port(), native=native_mode, favicon="icon/icon1.ico")
|
port = native.find_open_port()
|
||||||
|
#port = 8001
|
||||||
|
logger.debug(f"Run on port {port}")
|
||||||
|
|
||||||
|
# If native mode was requested, ensure pywebview has a usable GUI
|
||||||
|
# backend (Qt or GTK). Use importlib.util.find_spec to check for
|
||||||
|
# available backends without importing them (avoids side-effects).
|
||||||
|
if native_mode:
|
||||||
|
try:
|
||||||
|
if importlib.util.find_spec('webview') is None:
|
||||||
|
logger.debug('pywebview package not found; disabling native mode')
|
||||||
|
native_mode = False
|
||||||
|
else:
|
||||||
|
backend_available = False
|
||||||
|
if importlib.util.find_spec('PyQt5') or importlib.util.find_spec('PySide6'):
|
||||||
|
backend_available = True
|
||||||
|
elif importlib.util.find_spec('gi'):
|
||||||
|
backend_available = True
|
||||||
|
|
||||||
|
if not backend_available:
|
||||||
|
logger.debug('pywebview backend (Qt/GTK) not available; disabling native mode')
|
||||||
|
native_mode = False
|
||||||
|
except Exception as e:
|
||||||
|
logger.debug(f'pywebview import/check failed: {e}; disabling native mode')
|
||||||
|
native_mode = False
|
||||||
|
|
||||||
|
# Try to run with the requested mode. If native startup fails at
|
||||||
|
# runtime, catch and fall back to browser mode so the app stays usable.
|
||||||
|
try:
|
||||||
|
ui.run(reload=False, show=True, port=port, native=native_mode)
|
||||||
|
except Exception as e:
|
||||||
|
logger.error(f"Failed to start UI in native mode: {e}")
|
||||||
|
if native_mode:
|
||||||
|
logger.debug("Retrying without native mode (browser fallback)")
|
||||||
|
try:
|
||||||
|
ui.run(reload=False, show=True, port=port, native=False)
|
||||||
|
except Exception as e2:
|
||||||
|
logger.error(f"Fallback to browser mode failed: {e2}")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(e)
|
logger.error(e)
|
||||||
pass
|
pass
|
||||||
|
|||||||
@@ -0,0 +1,67 @@
|
|||||||
|
# -*- mode: python ; coding: utf-8 -*-
|
||||||
|
|
||||||
|
import os
|
||||||
|
import importlib.util
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
here = Path(__file__).resolve().parent
|
||||||
|
project_root = here
|
||||||
|
|
||||||
|
# Prepare datas and binaries lists; include icon resources
|
||||||
|
datas = [
|
||||||
|
(str(project_root / 'icon'), 'icon'),
|
||||||
|
(str(project_root / 'icon' / 'icon1.png'), 'icon'),
|
||||||
|
# include user-supplied CA bundle folder (all files) so frozen app can use it
|
||||||
|
(str(project_root / 'cert'), 'cert')
|
||||||
|
]
|
||||||
|
binaries = []
|
||||||
|
|
||||||
|
# Try to include pythonnet runtime DLLs so Python.Runtime.dll is available
|
||||||
|
try:
|
||||||
|
spec_py = importlib.util.find_spec("pythonnet")
|
||||||
|
if spec_py and spec_py.submodule_search_locations:
|
||||||
|
pn_path = Path(spec_py.submodule_search_locations[0])
|
||||||
|
runtime_dir = pn_path / "runtime"
|
||||||
|
if runtime_dir.exists():
|
||||||
|
for f in runtime_dir.rglob("*.dll"):
|
||||||
|
# destination folder inside the frozen app
|
||||||
|
binaries.append((str(f), "pythonnet/runtime"))
|
||||||
|
except Exception:
|
||||||
|
# best-effort; if we can't locate pythonnet at build time, user can add it manually
|
||||||
|
pass
|
||||||
|
|
||||||
|
a = Analysis(
|
||||||
|
['nice.py'],
|
||||||
|
pathex=[str(project_root)],
|
||||||
|
binaries=binaries,
|
||||||
|
datas=datas,
|
||||||
|
hiddenimports=['bs4', 'requests', 'nicegui', 'pynng'],
|
||||||
|
hookspath=[],
|
||||||
|
hooksconfig={},
|
||||||
|
runtime_hooks=[],
|
||||||
|
excludes=[],
|
||||||
|
noarchive=False,
|
||||||
|
optimize=0,
|
||||||
|
)
|
||||||
|
pyz = PYZ(a.pure)
|
||||||
|
|
||||||
|
exe = EXE(
|
||||||
|
pyz,
|
||||||
|
a.scripts,
|
||||||
|
a.binaries,
|
||||||
|
a.datas,
|
||||||
|
[],
|
||||||
|
name='nice_uploader',
|
||||||
|
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,
|
||||||
|
)
|
||||||
@@ -0,0 +1,3 @@
|
|||||||
|
import webview
|
||||||
|
webview.create_window('Hello world', 'https://pywebview.flowrl.com/hello')
|
||||||
|
webview.start()
|
||||||
+5
-3
@@ -1,12 +1,14 @@
|
|||||||
typer
|
typer
|
||||||
wxpython
|
|
||||||
pydicom
|
pydicom
|
||||||
dicognito
|
dicognito
|
||||||
python-datauri
|
python-datauri
|
||||||
bs4
|
bs4
|
||||||
requests
|
requests
|
||||||
zmq
|
pynng
|
||||||
blake3
|
blake3
|
||||||
loguru
|
loguru
|
||||||
dicom2jpg
|
dicom2jpg
|
||||||
pyinstaller
|
pyinstaller
|
||||||
|
nicegui
|
||||||
|
pywebview
|
||||||
|
psutil
|
||||||
|
|||||||
+259
@@ -0,0 +1,259 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Standalone SSL check utility to be frozen for testing.
|
||||||
|
|
||||||
|
Usage:
|
||||||
|
python ssl_check.py --url https://www.penracourses.org.uk --cert-dir ./cert
|
||||||
|
|
||||||
|
What it does:
|
||||||
|
- Runs `openssl s_client -showcerts` against the host (if openssl on PATH).
|
||||||
|
- Extracts PEM cert blocks from the output and inspects each with `openssl x509`.
|
||||||
|
- Loads cert files from --cert-dir (PEM or DER) and converts DER to PEM as needed.
|
||||||
|
- Compares server chain Issuers with Subjects of bundled certs and reports which CA certs are missing.
|
||||||
|
|
||||||
|
This script is intentionally small and self-contained so it can be frozen with PyInstaller
|
||||||
|
for quick testing on Windows where the system openssl may be too old.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import base64
|
||||||
|
import re
|
||||||
|
import shutil
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import List, Tuple
|
||||||
|
|
||||||
|
|
||||||
|
def find_openssl() -> str | None:
|
||||||
|
return shutil.which("openssl")
|
||||||
|
|
||||||
|
|
||||||
|
def run_openssl_s_client(host: str, port: int = 443, tls_flag: str | None = None, timeout: int = 60) -> str:
|
||||||
|
openssl = find_openssl()
|
||||||
|
if not openssl:
|
||||||
|
raise RuntimeError("openssl not found on PATH")
|
||||||
|
cmd = [openssl, "s_client", "-showcerts", "-connect", f"{host}:{port}", "-servername", host]
|
||||||
|
if tls_flag:
|
||||||
|
cmd.append(tls_flag)
|
||||||
|
# give a newline so s_client won't wait for interactive input
|
||||||
|
proc = subprocess.run(cmd, input="\n", capture_output=True, text=True, timeout=timeout)
|
||||||
|
return proc.stdout + "\n" + proc.stderr
|
||||||
|
|
||||||
|
|
||||||
|
def fetch_cert_via_proxy(host: str, port: int, proxy: str, timeout: int = 20) -> List[str]:
|
||||||
|
"""Connect to proxy, issue CONNECT, perform TLS handshake and return PEM of the leaf cert."""
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
import socket
|
||||||
|
import ssl
|
||||||
|
import base64
|
||||||
|
|
||||||
|
parsed = urlparse(proxy)
|
||||||
|
proxy_host = parsed.hostname
|
||||||
|
proxy_port = parsed.port or (443 if parsed.scheme == 'https' else 80)
|
||||||
|
|
||||||
|
auth_header = None
|
||||||
|
if parsed.username:
|
||||||
|
user = parsed.username
|
||||||
|
pwd = parsed.password or ""
|
||||||
|
cred = f"{user}:{pwd}".encode("utf-8")
|
||||||
|
auth_header = base64.b64encode(cred).decode("ascii")
|
||||||
|
|
||||||
|
s = socket.create_connection((proxy_host, proxy_port), timeout=timeout)
|
||||||
|
try:
|
||||||
|
connect_req = f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n"
|
||||||
|
if auth_header:
|
||||||
|
connect_req += f"Proxy-Authorization: Basic {auth_header}\r\n"
|
||||||
|
connect_req += "\r\n"
|
||||||
|
s.sendall(connect_req.encode("ascii"))
|
||||||
|
|
||||||
|
# read response headers
|
||||||
|
resp = b""
|
||||||
|
while b"\r\n\r\n" not in resp:
|
||||||
|
chunk = s.recv(4096)
|
||||||
|
if not chunk:
|
||||||
|
break
|
||||||
|
resp += chunk
|
||||||
|
if len(resp) > 65536:
|
||||||
|
break
|
||||||
|
|
||||||
|
header = resp.split(b"\r\n\r\n", 1)[0].decode("ascii", errors="ignore")
|
||||||
|
# Simple status check
|
||||||
|
first_line = header.splitlines()[0] if header.splitlines() else ""
|
||||||
|
if not first_line.startswith("HTTP/") or ("200" not in first_line):
|
||||||
|
raise RuntimeError(f"Proxy CONNECT failed: {first_line}")
|
||||||
|
|
||||||
|
# Wrap the socket with SSL and fetch peer cert (leaf)
|
||||||
|
ctx = ssl.create_default_context()
|
||||||
|
with ctx.wrap_socket(s, server_hostname=host) as ss:
|
||||||
|
der = ss.getpeercert(True)
|
||||||
|
pem = ssl.DER_cert_to_PEM_cert(der)
|
||||||
|
return [pem]
|
||||||
|
except Exception:
|
||||||
|
s.close()
|
||||||
|
raise
|
||||||
|
|
||||||
|
|
||||||
|
def extract_pem_blocks(s: str) -> List[str]:
|
||||||
|
pattern = re.compile(r"-----BEGIN CERTIFICATE-----(?:.|\n)*?-----END CERTIFICATE-----", re.M)
|
||||||
|
return pattern.findall(s)
|
||||||
|
|
||||||
|
|
||||||
|
def write_pems_to_temp(pems: List[str]) -> List[Path]:
|
||||||
|
out = []
|
||||||
|
for i, pem in enumerate(pems):
|
||||||
|
f = Path(tempfile.gettempdir()) / f"ssl_check_cert_{i}.pem"
|
||||||
|
f.write_text(pem)
|
||||||
|
out.append(f)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def inspect_cert(path: Path) -> Tuple[str, str, str]:
|
||||||
|
"""Return (subject, issuer, fingerprint) by calling openssl x509."""
|
||||||
|
openssl = find_openssl()
|
||||||
|
if not openssl:
|
||||||
|
raise RuntimeError("openssl not found on PATH")
|
||||||
|
proc = subprocess.run([openssl, "x509", "-in", str(path), "-noout", "-subject", "-issuer", "-fingerprint"], capture_output=True, text=True)
|
||||||
|
out = proc.stdout + proc.stderr
|
||||||
|
subj = re.search(r"subject=\s*(.*)", out)
|
||||||
|
issu = re.search(r"issuer=\s*(.*)", out)
|
||||||
|
fp = re.search(r"Fingerprint=([A-F0-9:]+)", out)
|
||||||
|
return (subj.group(1).strip() if subj else "", issu.group(1).strip() if issu else "", fp.group(1).strip() if fp else "")
|
||||||
|
|
||||||
|
|
||||||
|
def load_cert_files_from_dir(cert_dir: Path) -> List[Path]:
|
||||||
|
files = []
|
||||||
|
if not cert_dir.exists():
|
||||||
|
return files
|
||||||
|
for pattern in ("*.pem", "*.crt", "*.cer"):
|
||||||
|
files.extend(sorted(cert_dir.glob(pattern)))
|
||||||
|
|
||||||
|
out = []
|
||||||
|
for f in files:
|
||||||
|
data = f.read_bytes()
|
||||||
|
if b"-----BEGIN CERTIFICATE-----" in data:
|
||||||
|
# already PEM
|
||||||
|
out.append(f)
|
||||||
|
continue
|
||||||
|
# assume DER -> convert to PEM
|
||||||
|
b64 = base64.encodebytes(data).decode("ascii")
|
||||||
|
pem = "-----BEGIN CERTIFICATE-----\n" + b64 + "-----END CERTIFICATE-----\n"
|
||||||
|
tmp = Path(tempfile.gettempdir()) / f"ssl_check_bundle_{f.name}.pem"
|
||||||
|
tmp.write_text(pem)
|
||||||
|
out.append(tmp)
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def main(argv: List[str]) -> int:
|
||||||
|
p = argparse.ArgumentParser()
|
||||||
|
p.add_argument("--url", required=True, help="URL or host to check (eg https://www.penracourses.org.uk)")
|
||||||
|
p.add_argument("--cert-dir", default="cert", help="Directory with bundled certs")
|
||||||
|
p.add_argument("--force-tls12", action="store_true", help="Force -tls1_2 on openssl if initial attempt failed")
|
||||||
|
args = p.parse_args(argv)
|
||||||
|
|
||||||
|
# parse host/port
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
|
parsed = urlparse(args.url if "//" in args.url else "//" + args.url)
|
||||||
|
host = parsed.hostname or args.url
|
||||||
|
port = parsed.port or 443
|
||||||
|
|
||||||
|
openssl = find_openssl()
|
||||||
|
if not openssl:
|
||||||
|
print("openssl not found on PATH; please install Git for Windows or OpenSSL and re-run.")
|
||||||
|
# we still attempt Python fallback but chain won't be available
|
||||||
|
else:
|
||||||
|
print(f"Using openssl: {openssl}")
|
||||||
|
|
||||||
|
out = ""
|
||||||
|
if openssl:
|
||||||
|
try:
|
||||||
|
out = run_openssl_s_client(host, port)
|
||||||
|
if "tlsv1 alert protocol version" in out or "SSL23_GET_SERVER_HELLO" in out:
|
||||||
|
print("Server rejected legacy TLS; retrying with -tls1_2")
|
||||||
|
out2 = run_openssl_s_client(host, port, tls_flag="-tls1_2")
|
||||||
|
out = out + "\n--- retry result ---\n" + out2
|
||||||
|
except subprocess.TimeoutExpired:
|
||||||
|
print("OpenSSL check timed out")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"OpenSSL check failed: {e}")
|
||||||
|
|
||||||
|
# If openssl couldn't fetch a chain (or wasn't available), attempt a proxy-aware
|
||||||
|
# fetch if HTTPS_PROXY or HTTP_PROXY is set. This helps in corporate networks
|
||||||
|
# where direct TLS is blocked and a proxy must be used.
|
||||||
|
if not extract_pem_blocks(out):
|
||||||
|
proxy = None
|
||||||
|
for key in ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy"):
|
||||||
|
proxy = os.environ.get(key)
|
||||||
|
if proxy:
|
||||||
|
break
|
||||||
|
if proxy:
|
||||||
|
try:
|
||||||
|
print(f"Attempting proxy CONNECT via {proxy}")
|
||||||
|
pems = fetch_cert_via_proxy(host, port, proxy)
|
||||||
|
if pems:
|
||||||
|
print(f"Fetched {len(pems)} PEM(s) via proxy")
|
||||||
|
server_paths = write_pems_to_temp(pems)
|
||||||
|
server_info = [inspect_cert(p) for p in server_paths]
|
||||||
|
print("Server chain (proxy fetched):")
|
||||||
|
for i, (s, iss, fp) in enumerate(server_info):
|
||||||
|
print(f"[{i}] SUBJECT: {s}")
|
||||||
|
print(f" ISSUER: {iss}")
|
||||||
|
print(f" FP: {fp}")
|
||||||
|
else:
|
||||||
|
print("Proxy CONNECT succeeded but no certs found")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Proxy CONNECT fetch failed: {e}")
|
||||||
|
|
||||||
|
# extract certs from openssl output if any
|
||||||
|
pems = extract_pem_blocks(out)
|
||||||
|
if pems:
|
||||||
|
print(f"Found {len(pems)} PEM blocks in openssl output")
|
||||||
|
server_paths = write_pems_to_temp(pems)
|
||||||
|
server_info = [inspect_cert(p) for p in server_paths]
|
||||||
|
print("Server chain:")
|
||||||
|
for i, (s, iss, fp) in enumerate(server_info):
|
||||||
|
print(f"[{i}] SUBJECT: {s}")
|
||||||
|
print(f" ISSUER: {iss}")
|
||||||
|
print(f" FP: {fp}")
|
||||||
|
else:
|
||||||
|
print("No certificate chain found from openssl (server may have rejected the handshake or openssl is incompatible)")
|
||||||
|
server_info = []
|
||||||
|
|
||||||
|
# load bundled certs
|
||||||
|
cert_dir = Path(args.cert_dir)
|
||||||
|
bundled = load_cert_files_from_dir(cert_dir)
|
||||||
|
print(f"Found {len(bundled)} bundled cert files in {cert_dir}")
|
||||||
|
bundled_info = [inspect_cert(p) for p in bundled]
|
||||||
|
subjects = {info[0]: p for info, p in zip(bundled_info, bundled)}
|
||||||
|
|
||||||
|
# Compare: ensure each server issuer is present in bundled subjects
|
||||||
|
missing = set()
|
||||||
|
for s, iss, fp in server_info:
|
||||||
|
if iss and iss not in subjects:
|
||||||
|
missing.add(iss)
|
||||||
|
|
||||||
|
if not server_info:
|
||||||
|
print("No server certs to compare; check network/openssl compatibility")
|
||||||
|
return 2
|
||||||
|
|
||||||
|
if not missing:
|
||||||
|
print("All server issuers are present in the bundled certs. OK")
|
||||||
|
return 0
|
||||||
|
else:
|
||||||
|
print("Missing issuer certificates in bundled certs:")
|
||||||
|
for m in missing:
|
||||||
|
print(" - ", m)
|
||||||
|
return 3
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
try:
|
||||||
|
rc = main(sys.argv[1:])
|
||||||
|
except Exception as e:
|
||||||
|
print(f"ssl_check failed: {e}")
|
||||||
|
rc = 1
|
||||||
|
sys.exit(rc)
|
||||||
@@ -2,19 +2,17 @@
|
|||||||
|
|
||||||
|
|
||||||
a = Analysis(
|
a = Analysis(
|
||||||
['anon_gui.py'],
|
['ssl_check.py'],
|
||||||
pathex=[],
|
pathex=[],
|
||||||
binaries=[],
|
binaries=[],
|
||||||
datas=[
|
datas=[],
|
||||||
('icon\\icon1.ico', 'icon'),
|
|
||||||
('icon\\icon1.png', 'icon'),
|
|
||||||
],
|
|
||||||
hiddenimports=[],
|
hiddenimports=[],
|
||||||
hookspath=[],
|
hookspath=[],
|
||||||
hooksconfig={},
|
hooksconfig={},
|
||||||
runtime_hooks=[],
|
runtime_hooks=[],
|
||||||
excludes=[],
|
excludes=[],
|
||||||
noarchive=False,
|
noarchive=False,
|
||||||
|
optimize=0,
|
||||||
)
|
)
|
||||||
pyz = PYZ(a.pure)
|
pyz = PYZ(a.pure)
|
||||||
|
|
||||||
@@ -24,18 +22,17 @@ exe = EXE(
|
|||||||
a.binaries,
|
a.binaries,
|
||||||
a.datas,
|
a.datas,
|
||||||
[],
|
[],
|
||||||
name='anon_gui',
|
name='ssl_check',
|
||||||
debug=False,
|
debug=False,
|
||||||
bootloader_ignore_signals=False,
|
bootloader_ignore_signals=False,
|
||||||
strip=False,
|
strip=False,
|
||||||
upx=True,
|
upx=True,
|
||||||
upx_exclude=[],
|
upx_exclude=[],
|
||||||
runtime_tmpdir=None,
|
runtime_tmpdir=None,
|
||||||
console=False,
|
console=True,
|
||||||
disable_windowed_traceback=False,
|
disable_windowed_traceback=False,
|
||||||
argv_emulation=False,
|
argv_emulation=False,
|
||||||
target_arch=None,
|
target_arch=None,
|
||||||
codesign_identity=None,
|
codesign_identity=None,
|
||||||
entitlements_file=None,
|
entitlements_file=None,
|
||||||
icon='icon\\icon1.ico'
|
|
||||||
)
|
)
|
||||||
@@ -0,0 +1,6 @@
|
|||||||
|
PS T:\rad-tools\uploader\Uploader_test> .\ssl_check.exe --url http://www.penracourses.org.uk --cert-dir ./cert
|
||||||
|
Using openssl: C:\Sybase\OCS-15_0\bin\openssl.EXE
|
||||||
|
Server rejected legacy TLS; retrying with -tls1_2
|
||||||
|
No certificate chain found from openssl (server may have rejected the handshake or openssl is incompatible)
|
||||||
|
Found 2 bundled cert files in cert
|
||||||
|
No server certs to compare; check network/openssl compatibility
|
||||||
@@ -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()
|
|
||||||
Reference in New Issue
Block a user