697 lines
32 KiB
Python
697 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
FireMapper - Session Post-Processing (graphical interface)
|
|
==========================================================
|
|
|
|
Graphical front-end for the FireMapper post-processing workflow:
|
|
|
|
1. Embed GPS & metadata - writes each frame's JSON sidecar (GPS, orientation,
|
|
capture time, lens) into the EXIF/XMP of tagged image copies for use in
|
|
mapping software. (wraps embed_metadata.py)
|
|
|
|
2. Thermal stretch - rescales the session's 16-bit radiometric thermal frames
|
|
to a single shared brightness window and saves 8-bit images in the chosen
|
|
palette. (wraps stretch_thermal.py)
|
|
|
|
3. Map - plots trigger points and image footprints on OpenStreetMap.
|
|
(wraps session_map.py)
|
|
|
|
Launch with: python firemapper_gui.py
|
|
|
|
Select a session folder, review the description on each tab, set the options,
|
|
and start the operation. Long-running tasks execute on a background thread with
|
|
a progress indicator. Original files are never modified.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import math
|
|
import os
|
|
import queue
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
import tkinter as tk
|
|
from tkinter import filedialog, messagebox, scrolledtext, ttk
|
|
|
|
from PIL import Image, ImageDraw, ImageTk
|
|
|
|
# import the worker modules that live next to this file
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(APP_DIR))
|
|
import embed_metadata as embed # noqa: E402
|
|
import stretch_thermal as stretch # noqa: E402
|
|
import session_map as smap # noqa: E402
|
|
import map_export # noqa: E402
|
|
|
|
try:
|
|
from tkintermapview import TkinterMapView # noqa: E402
|
|
except Exception: # noqa: BLE001 (optional dependency - the Map tab degrades gracefully)
|
|
TkinterMapView = None
|
|
|
|
PAD = 10
|
|
HEADING = ("Segoe UI Semibold", 11)
|
|
EXPLAIN_WRAP = 720
|
|
|
|
|
|
def autodetect_session() -> str:
|
|
"""Newest session folder near this script (also one level into parent folders)."""
|
|
sessions = smap.list_sessions(APP_DIR)
|
|
return str(max(sessions, key=lambda p: p.stat().st_mtime)) if sessions else ""
|
|
|
|
|
|
def open_in_explorer(path: Path) -> None:
|
|
if hasattr(os, "startfile"):
|
|
try:
|
|
os.startfile(path) # type: ignore[attr-defined]
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
class FireMapperGUI(tk.Tk):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.title("FireMapper - Session Post-Processing")
|
|
self.geometry("960x780")
|
|
self.minsize(820, 640)
|
|
|
|
# style must be created with THIS root as master - never before it,
|
|
# or tkinter spins up a second, stray default root and the StringVars
|
|
# bind to the wrong one (entries then render blank).
|
|
style = ttk.Style(self)
|
|
if "vista" in style.theme_names():
|
|
style.theme_use("vista")
|
|
style.configure("Card.TFrame", background="#f5f6f8", relief="solid", borderwidth=1)
|
|
|
|
self.q: queue.Queue = queue.Queue()
|
|
self.running = False
|
|
self.action_buttons: list[ttk.Button] = []
|
|
self.logs: dict[str, scrolledtext.ScrolledText] = {}
|
|
self.bars: dict[str, ttk.Progressbar] = {}
|
|
self.status_vars: dict[str, tk.StringVar] = {}
|
|
self._preview_imgtk = None # keep a reference so Tk doesn't GC it
|
|
self.workers = tk.IntVar(self, value=embed.default_workers())
|
|
|
|
self._build_header()
|
|
nb = ttk.Notebook(self)
|
|
nb.pack(fill="both", expand=True, padx=PAD, pady=(0, PAD))
|
|
self._build_embed_tab(nb)
|
|
self._build_thermal_tab(nb)
|
|
self._build_map_tab(nb)
|
|
|
|
self.after(100, self._poll_queue)
|
|
|
|
# ----- shared UI bits -------------------------------------------------
|
|
def _build_header(self):
|
|
head = ttk.Frame(self, padding=(PAD, PAD, PAD, 4))
|
|
head.pack(fill="x")
|
|
ttk.Label(head, text="FireMapper - Session Post-Processing",
|
|
font=("Segoe UI Semibold", 15)).pack(anchor="w")
|
|
ttk.Label(head, foreground="#555", wraplength=900, justify="left",
|
|
text="Convert a raw capture session into mapping-ready imagery. Select a "
|
|
"session folder and run any step. Original files are never modified; "
|
|
"all results are written to separate output folders.").pack(anchor="w")
|
|
ttk.Separator(self).pack(fill="x", padx=PAD, pady=(6, 8))
|
|
|
|
@staticmethod
|
|
def _explain(parent, text):
|
|
box = ttk.Frame(parent, padding=8)
|
|
box.configure(style="Card.TFrame")
|
|
ttk.Label(box, text=text, wraplength=EXPLAIN_WRAP, justify="left",
|
|
foreground="#333").pack(anchor="w")
|
|
return box
|
|
|
|
def _folder_row(self, parent, label, var, browse_dir=True):
|
|
row = ttk.Frame(parent)
|
|
ttk.Label(row, text=label, width=16).pack(side="left")
|
|
ttk.Entry(row, textvariable=var).pack(side="left", fill="x", expand=True)
|
|
|
|
def browse():
|
|
if browse_dir:
|
|
p = filedialog.askdirectory(title=label,
|
|
initialdir=var.get() or str(APP_DIR))
|
|
else:
|
|
p = filedialog.askopenfilename(title=label,
|
|
initialdir=var.get() or str(APP_DIR))
|
|
if p:
|
|
var.set(p)
|
|
ttk.Button(row, text="Browse...", command=browse).pack(side="left", padx=(6, 0))
|
|
return row
|
|
|
|
def _workers_row(self, parent):
|
|
row = ttk.Frame(parent)
|
|
ttk.Label(row, text="Parallel workers", width=16).pack(side="left")
|
|
ttk.Spinbox(row, from_=1, to=(os.cpu_count() or 32), increment=1, width=6,
|
|
textvariable=self.workers).pack(side="left")
|
|
ttk.Label(row, foreground="#777",
|
|
text=f" concurrent workers for copying, rendering and exiftool "
|
|
f"({os.cpu_count()} logical cores available)"
|
|
).pack(side="left", padx=6)
|
|
return row
|
|
|
|
def _progress_block(self, parent, which):
|
|
frame = ttk.Frame(parent)
|
|
bar = ttk.Progressbar(frame, mode="determinate", maximum=1000)
|
|
bar.pack(fill="x")
|
|
sv = tk.StringVar(self, value="Idle.")
|
|
ttk.Label(frame, textvariable=sv, foreground="#555").pack(anchor="w", pady=(2, 0))
|
|
log = scrolledtext.ScrolledText(frame, height=11, wrap="word",
|
|
font=("Consolas", 9))
|
|
log.pack(fill="both", expand=True, pady=(6, 0))
|
|
self.bars[which] = bar
|
|
self.status_vars[which] = sv
|
|
self.logs[which] = log
|
|
return frame
|
|
|
|
# ----- tab 1: embed ---------------------------------------------------
|
|
def _build_embed_tab(self, nb):
|
|
tab = ttk.Frame(nb, padding=PAD)
|
|
nb.add(tab, text=" 1. Embed GPS & Metadata ")
|
|
|
|
ttk.Label(tab, text="Embed GPS & Metadata", font=HEADING).pack(anchor="w")
|
|
self._explain(tab,
|
|
"Creates a tagged copy of every image in the session and writes the "
|
|
"corresponding JSON values into its EXIF/XMP metadata: GPS position, camera "
|
|
"heading, pitch and roll, UTC capture time, lens and exposure, and the complete "
|
|
"JSON record in the comment field. The tagged copies are compatible with "
|
|
"photogrammetry and GIS applications (Pix4D, Metashape, QGIS), which read the "
|
|
"embedded GPS and camera orientation to position each image. It also writes a "
|
|
"metashape_reference.csv (GPS + yaw/pitch/roll for every frame) into the output "
|
|
"folder, which you can load directly via Metashape's Import Reference if you "
|
|
"prefer a CSV over the embedded tags. Requires exiftool, which is located "
|
|
"automatically and installed on first use if not already present."
|
|
).pack(fill="x", pady=(4, 10))
|
|
|
|
self.embed_session = tk.StringVar(self, value=autodetect_session())
|
|
self.embed_out = tk.StringVar(self, value="")
|
|
self.embed_gps = tk.StringVar(self, value="position")
|
|
self.embed_exif = tk.StringVar(self, value="")
|
|
|
|
self._folder_row(tab, "Session folder", self.embed_session).pack(fill="x", pady=3)
|
|
self._folder_row(tab, "Output folder", self.embed_out).pack(fill="x", pady=3)
|
|
ttk.Label(tab, text="If left blank, output is written to <session>_exif beside the session.",
|
|
foreground="#777").pack(anchor="w", padx=(16, 0))
|
|
|
|
gps = ttk.LabelFrame(tab, text="GPS source for camera frames", padding=8)
|
|
gps.pack(fill="x", pady=(10, 4))
|
|
ttk.Radiobutton(gps, variable=self.embed_gps, value="position",
|
|
text="Fused INS (position) - highest accuracy (recommended)").pack(anchor="w")
|
|
ttk.Radiobutton(gps, variable=self.embed_gps, value="gps",
|
|
text="Raw GNSS (gps) - uncorrected satellite fix").pack(anchor="w")
|
|
ttk.Label(gps, foreground="#777",
|
|
text="Thermal frames have no INS solution and always use raw GNSS.").pack(anchor="w")
|
|
|
|
adv = ttk.Frame(tab)
|
|
adv.pack(fill="x", pady=(8, 2))
|
|
ttk.Label(adv, text="exiftool path", width=16).pack(side="left")
|
|
ttk.Entry(adv, textvariable=self.embed_exif).pack(side="left", fill="x", expand=True)
|
|
ttk.Label(adv, text="(optional; auto-detected)", foreground="#777").pack(side="left", padx=6)
|
|
|
|
self._workers_row(tab).pack(fill="x", pady=(8, 2))
|
|
|
|
btn = ttk.Button(tab, text="Embed metadata", command=self._start_embed)
|
|
btn.pack(anchor="w", pady=10)
|
|
self.action_buttons.append(btn)
|
|
|
|
self._progress_block(tab, "embed").pack(fill="both", expand=True)
|
|
|
|
# ----- tab 2: thermal -------------------------------------------------
|
|
def _build_thermal_tab(self, nb):
|
|
tab = ttk.Frame(nb, padding=PAD)
|
|
nb.add(tab, text=" 2. Thermal Stretch ")
|
|
|
|
ttk.Label(tab, text="Stretch Thermal Frames for Visualisation", font=HEADING).pack(anchor="w")
|
|
self._explain(tab,
|
|
"The thermal camera records 16-bit radiometric frames that occupy only a small "
|
|
"portion of the available range; consequently the raw files appear nearly black "
|
|
"and vary in brightness from frame to frame. This step determines a single "
|
|
"brightness window for the entire session and rescales every frame into it, "
|
|
"producing viewable 8-bit images. Because all frames share one window, warm and "
|
|
"cool areas remain consistent across the flight. Pixel values represent "
|
|
"radiometric signal (proportional to temperature), not calibrated degrees. Use "
|
|
"'Preview palettes' to compare the available palettes before processing."
|
|
).pack(fill="x", pady=(4, 10))
|
|
|
|
self.th_session = tk.StringVar(self, value=autodetect_session())
|
|
self.th_out = tk.StringVar(self, value="")
|
|
self.th_cmap = tk.StringVar(self, value="inferno")
|
|
self.th_lo = tk.DoubleVar(self, value=1.0)
|
|
self.th_hi = tk.DoubleVar(self, value=99.0)
|
|
self.th_absolute = tk.BooleanVar(self, value=False)
|
|
|
|
self._folder_row(tab, "Session folder", self.th_session).pack(fill="x", pady=3)
|
|
self._folder_row(tab, "Output folder", self.th_out).pack(fill="x", pady=3)
|
|
ttk.Label(tab, text="If left blank, output is written to <session>/thermal_stretched.",
|
|
foreground="#777").pack(anchor="w", padx=(16, 0))
|
|
|
|
opts = ttk.Frame(tab)
|
|
opts.pack(fill="x", pady=(10, 4))
|
|
ttk.Label(opts, text="Palette", width=16).pack(side="left")
|
|
ttk.Combobox(opts, textvariable=self.th_cmap, width=12, state="readonly",
|
|
values=["inferno", "ironbow", "gray"]).pack(side="left")
|
|
ttk.Label(opts, text=" Window low %").pack(side="left")
|
|
ttk.Spinbox(opts, from_=0, to=49, increment=0.5, width=6,
|
|
textvariable=self.th_lo).pack(side="left", padx=4)
|
|
ttk.Label(opts, text="high %").pack(side="left")
|
|
ttk.Spinbox(opts, from_=51, to=100, increment=0.5, width=6,
|
|
textvariable=self.th_hi).pack(side="left", padx=4)
|
|
ttk.Checkbutton(opts, text="Absolute min/max", variable=self.th_absolute).pack(side="left", padx=10)
|
|
|
|
self.th_embed = tk.BooleanVar(self, value=True)
|
|
ttk.Checkbutton(tab, variable=self.th_embed,
|
|
text="Embed GPS and orientation metadata into the stretched PNGs "
|
|
"(requires exiftool)").pack(anchor="w", pady=(6, 0))
|
|
|
|
self._workers_row(tab).pack(fill="x", pady=(6, 2))
|
|
|
|
bar = ttk.Frame(tab)
|
|
bar.pack(fill="x", pady=10)
|
|
pv = ttk.Button(bar, text="Preview palettes", command=self._start_preview)
|
|
pv.pack(side="left")
|
|
run = ttk.Button(bar, text="Stretch all frames", command=self._start_stretch)
|
|
run.pack(side="left", padx=8)
|
|
self.action_buttons += [pv, run]
|
|
|
|
ttk.Label(tab, text="Preview (middle frame) gray | inferno | ironbow",
|
|
foreground="#777").pack(anchor="w")
|
|
self.preview_label = ttk.Label(tab, anchor="center")
|
|
self.preview_label.pack(fill="x", pady=(2, 6))
|
|
|
|
self._progress_block(tab, "thermal").pack(fill="both", expand=True)
|
|
|
|
# ----- tab 3: map -----------------------------------------------------
|
|
def _build_map_tab(self, nb):
|
|
tab = ttk.Frame(nb, padding=PAD)
|
|
nb.add(tab, text=" 3. Map ")
|
|
|
|
if TkinterMapView is None:
|
|
ttk.Label(tab, foreground="#a00000", wraplength=600, justify="left",
|
|
text="The map requires the 'tkintermapview' package.\n\n"
|
|
"Install it from a terminal:\n"
|
|
" pip install tkintermapview\n\n"
|
|
"then restart the application.").pack(anchor="w", pady=20)
|
|
return
|
|
|
|
ttk.Label(tab, text="Trigger Points & Image Footprints on OpenStreetMap",
|
|
font=HEADING).pack(anchor="w")
|
|
self._explain(tab,
|
|
"Select the sessions to display and click 'Show on map'. Each image is shown as "
|
|
"a trigger point (its GPS position) and, optionally, as an oblique footprint: "
|
|
"the ground area the image covers, projected from the camera off-nadir angle "
|
|
"(cam25 = 25 deg, cam45 = 45 deg), the cross-track scan angle, the aircraft "
|
|
"attitude and the height above ground. Footprints assume flat terrain at the "
|
|
"elevation specified below. An internet connection is required for the map tiles."
|
|
).pack(fill="x", pady=(4, 8))
|
|
|
|
body = ttk.Frame(tab)
|
|
body.pack(fill="both", expand=True)
|
|
left = ttk.Frame(body)
|
|
left.pack(side="left", fill="y", padx=(0, 8))
|
|
right = ttk.Frame(body)
|
|
right.pack(side="left", fill="both", expand=True)
|
|
|
|
sess_box = ttk.LabelFrame(left, text="Sessions", padding=6)
|
|
sess_box.pack(fill="x")
|
|
self.session_list_frame = ttk.Frame(sess_box)
|
|
self.session_list_frame.pack(fill="x")
|
|
self.session_vars: dict[str, tk.BooleanVar] = {}
|
|
btns = ttk.Frame(sess_box)
|
|
btns.pack(anchor="w", pady=(4, 0))
|
|
ttk.Button(btns, text="All", width=5,
|
|
command=lambda: self._set_all_sessions(True)).pack(side="left")
|
|
ttk.Button(btns, text="None", width=5,
|
|
command=lambda: self._set_all_sessions(False)).pack(side="left", padx=4)
|
|
ttk.Button(btns, text="Refresh", width=8,
|
|
command=self._refresh_sessions).pack(side="left")
|
|
|
|
opt = ttk.LabelFrame(left, text="Options", padding=6)
|
|
opt.pack(fill="x", pady=(8, 0))
|
|
self.map_ground = tk.StringVar(self, value="110")
|
|
self.map_tfov_h = tk.StringVar(self, value=str(smap.THERMAL_FOV_DEFAULT[0])) # FLIR A65 25deg
|
|
self.map_tfov_v = tk.StringVar(self, value=str(smap.THERMAL_FOV_DEFAULT[1]))
|
|
self.map_toff = tk.StringVar(self, value="0")
|
|
self.map_step = tk.StringVar(self, value="auto")
|
|
self.map_markers = tk.BooleanVar(self, value=True)
|
|
self.map_foot = tk.BooleanVar(self, value=True)
|
|
self.map_cam = tk.BooleanVar(self, value=True)
|
|
self.map_thermal = tk.BooleanVar(self, value=True)
|
|
|
|
def field(label, var):
|
|
r = ttk.Frame(opt)
|
|
r.pack(fill="x", pady=1)
|
|
ttk.Label(r, text=label, width=17).pack(side="left")
|
|
ttk.Entry(r, textvariable=var, width=8).pack(side="left")
|
|
|
|
field("Ground elev (m)", self.map_ground)
|
|
field("Thermal FOV H (deg)", self.map_tfov_h)
|
|
field("Thermal FOV V (deg)", self.map_tfov_v)
|
|
field("Thermal off-nadir", self.map_toff)
|
|
field("Plot every Nth", self.map_step)
|
|
ttk.Label(opt, foreground="#777",
|
|
text="Thermal footprints require the field-of-view values above.").pack(anchor="w", pady=(2, 4))
|
|
ttk.Checkbutton(opt, text="Trigger points", variable=self.map_markers).pack(anchor="w")
|
|
ttk.Checkbutton(opt, text="Footprints", variable=self.map_foot).pack(anchor="w")
|
|
ttk.Checkbutton(opt, text="Cameras (blue)", variable=self.map_cam).pack(anchor="w")
|
|
ttk.Checkbutton(opt, text="Thermal (orange)", variable=self.map_thermal).pack(anchor="w")
|
|
|
|
self.map_btn = ttk.Button(left, text="Show on map", command=self._start_plot_map)
|
|
self.map_btn.pack(fill="x", pady=(8, 2))
|
|
self.action_buttons.append(self.map_btn)
|
|
|
|
self.map_export_btn = ttk.Button(
|
|
left, text="Open full map in browser", command=self._start_export_map)
|
|
self.map_export_btn.pack(fill="x")
|
|
self.action_buttons.append(self.map_export_btn)
|
|
ttk.Label(left, foreground="#777", wraplength=230, justify="left",
|
|
text="The in-app map above thins dense flights for speed. The browser map "
|
|
"renders every trigger and footprint, with per-session and per-kind "
|
|
"toggles and clustering. Needs internet.").pack(anchor="w", pady=(2, 6))
|
|
|
|
self.map_status = tk.StringVar(self, value="Idle.")
|
|
ttk.Label(left, textvariable=self.map_status, wraplength=230,
|
|
foreground="#555").pack(anchor="w")
|
|
|
|
# small dot icons so hundreds of trigger points stay legible (the default
|
|
# tkintermapview pins are large and merge into a blob at this density)
|
|
self._dot_icons = {"cam": self._dot("#1565c0"), "thermal": self._dot("#e65100")}
|
|
|
|
self.map_widget = TkinterMapView(right, corner_radius=0)
|
|
self.map_widget.pack(fill="both", expand=True)
|
|
self.map_widget.set_tile_server(
|
|
"https://tile.openstreetmap.org/{z}/{x}/{y}.png", max_zoom=19)
|
|
self.map_widget.set_position(49.32, 8.43) # Speyer area, until data is plotted
|
|
self.map_widget.set_zoom(9)
|
|
|
|
self._refresh_sessions()
|
|
|
|
def _set_all_sessions(self, value: bool):
|
|
for v in self.session_vars.values():
|
|
v.set(value)
|
|
|
|
def _refresh_sessions(self):
|
|
for w in self.session_list_frame.winfo_children():
|
|
w.destroy()
|
|
self.session_vars = {}
|
|
sessions = smap.list_sessions(APP_DIR)
|
|
for s in sessions:
|
|
v = tk.BooleanVar(self, value=True) # default: show all strips
|
|
self.session_vars[str(s)] = v
|
|
ttk.Checkbutton(self.session_list_frame, text=s.name, variable=v).pack(anchor="w")
|
|
if not sessions:
|
|
ttk.Label(self.session_list_frame, text="(no sessions found)",
|
|
foreground="#777").pack(anchor="w")
|
|
|
|
def _read_map_inputs(self):
|
|
"""Parse the shared map options, or None (after warning) if invalid."""
|
|
sel = [s for s, v in self.session_vars.items() if v.get()]
|
|
if not sel:
|
|
messagebox.showwarning("No sessions", "Select at least one session to display.")
|
|
return None
|
|
try:
|
|
ground = float(self.map_ground.get())
|
|
except ValueError:
|
|
messagebox.showwarning("Ground elevation", "Ground elevation must be a number (m).")
|
|
return None
|
|
tfov = None
|
|
if self.map_tfov_h.get().strip() and self.map_tfov_v.get().strip():
|
|
try:
|
|
tfov = (float(self.map_tfov_h.get()), float(self.map_tfov_v.get()))
|
|
except ValueError:
|
|
messagebox.showwarning("Thermal FOV", "Thermal FOV must be numbers (degrees).")
|
|
return None
|
|
try:
|
|
toff = float(self.map_toff.get() or 0)
|
|
except ValueError:
|
|
toff = 0.0
|
|
return sel, ground, tfov, toff
|
|
|
|
def _start_plot_map(self):
|
|
if self.running:
|
|
return
|
|
parsed = self._read_map_inputs()
|
|
if parsed is None:
|
|
return
|
|
sel, ground, tfov, toff = parsed
|
|
opts = dict(sessions=sel, ground=ground, tfov=tfov, toff=toff,
|
|
step_raw=self.map_step.get().strip().lower(),
|
|
markers=self.map_markers.get(), foot=self.map_foot.get(),
|
|
inc_cam=self.map_cam.get(), inc_thermal=self.map_thermal.get())
|
|
self._set_running(True)
|
|
self.map_status.set("Reading sessions ...")
|
|
|
|
def work():
|
|
try:
|
|
frames = []
|
|
for s in opts["sessions"]:
|
|
frames += smap.iter_session_frames(
|
|
s, include_cam=opts["inc_cam"], include_thermal=opts["inc_thermal"],
|
|
thermal_off_nadir=opts["toff"])
|
|
if not frames:
|
|
self.q.put(("maperror", "No frames with GPS in the selected sessions."))
|
|
return
|
|
total = len(frames)
|
|
if opts["step_raw"] in ("", "auto", "0"):
|
|
step = max(1, math.ceil(total / 600))
|
|
else:
|
|
step = max(1, int(float(opts["step_raw"])))
|
|
use = frames[::step]
|
|
markers = [(f["lat"], f["lon"], f["kind"]) for f in use] if opts["markers"] else []
|
|
polys = []
|
|
if opts["foot"]:
|
|
for f in use:
|
|
c = smap.footprint(f, opts["ground"], thermal_fov_deg=opts["tfov"])
|
|
if c:
|
|
polys.append((c, f["kind"]))
|
|
self.q.put(("map", {"markers": markers, "polys": polys,
|
|
"bounds": smap.bounds(use), "total": total,
|
|
"shown": len(use), "step": step}))
|
|
except Exception as e: # noqa: BLE001
|
|
self.q.put(("maperror", str(e)))
|
|
threading.Thread(target=work, daemon=True).start()
|
|
|
|
def _start_export_map(self):
|
|
"""Render *every* trigger/footprint to a standalone Leaflet page and open it."""
|
|
if self.running:
|
|
return
|
|
parsed = self._read_map_inputs()
|
|
if parsed is None:
|
|
return
|
|
sel, ground, tfov, toff = parsed
|
|
opts = dict(sessions=sel, ground=ground, tfov=tfov, toff=toff,
|
|
inc_cam=self.map_cam.get(), inc_thermal=self.map_thermal.get(),
|
|
markers=self.map_markers.get(), foot=self.map_foot.get())
|
|
self._set_running(True)
|
|
self.map_status.set("Building full map (all frames) ...")
|
|
|
|
def work():
|
|
try:
|
|
frames = []
|
|
for s in opts["sessions"]:
|
|
frames += smap.iter_session_frames(
|
|
s, include_cam=opts["inc_cam"], include_thermal=opts["inc_thermal"],
|
|
thermal_off_nadir=opts["toff"])
|
|
if not frames:
|
|
self.q.put(("maperror", "No frames with GPS in the selected sessions."))
|
|
return
|
|
res = map_export.export(
|
|
frames, ground_elev=opts["ground"], thermal_fov_deg=opts["tfov"],
|
|
include_markers=opts["markers"], include_footprints=opts["foot"],
|
|
title="FireMapper - " + ", ".join(Path(s).name for s in opts["sessions"]))
|
|
self.q.put(("mapexported", res))
|
|
except Exception as e: # noqa: BLE001
|
|
self.q.put(("maperror", str(e)))
|
|
threading.Thread(target=work, daemon=True).start()
|
|
|
|
@staticmethod
|
|
def _dot(color, d=9):
|
|
"""A small round marker icon (PhotoImage) of the given colour."""
|
|
img = Image.new("RGBA", (d, d), (0, 0, 0, 0))
|
|
ImageDraw.Draw(img).ellipse([0, 0, d - 1, d - 1], fill=color, outline="#ffffff")
|
|
return ImageTk.PhotoImage(img)
|
|
|
|
def _render_map(self, data):
|
|
self._set_running(False)
|
|
mw = self.map_widget
|
|
mw.delete_all_marker()
|
|
mw.delete_all_polygon()
|
|
colors = {"cam": "#1565c0", "thermal": "#e65100"}
|
|
for corners, kind in data["polys"]:
|
|
mw.set_polygon(corners, outline_color=colors.get(kind, "#333333"), border_width=1)
|
|
for lat, lon, kind in data["markers"]:
|
|
mw.set_marker(lat, lon, text="", icon=self._dot_icons.get(kind), icon_anchor="center")
|
|
mn_lat, mn_lon, mx_lat, mx_lon = data["bounds"]
|
|
try:
|
|
if mx_lat > mn_lat and mx_lon > mn_lon:
|
|
mw.fit_bounding_box((mx_lat, mn_lon), (mn_lat, mx_lon))
|
|
else:
|
|
mw.set_position((mn_lat + mx_lat) / 2, (mn_lon + mx_lon) / 2)
|
|
mw.set_zoom(14)
|
|
except Exception: # noqa: BLE001
|
|
pass
|
|
self.map_status.set(
|
|
f"Plotted {data['shown']} of {data['total']} frames (every {data['step']}): "
|
|
f"{len(data['markers'])} points, {len(data['polys'])} footprints.")
|
|
|
|
# ----- run helpers ----------------------------------------------------
|
|
def _set_running(self, busy: bool):
|
|
self.running = busy
|
|
for b in self.action_buttons:
|
|
b.config(state="disabled" if busy else "normal")
|
|
|
|
def _callbacks(self, which):
|
|
return (lambda frac, msg: self.q.put(("progress", which, frac, msg)),
|
|
lambda msg: self.q.put(("log", which, msg)))
|
|
|
|
def _guard(self, session: str, which: str) -> bool:
|
|
if self.running:
|
|
return False
|
|
if not session or not Path(session).is_dir():
|
|
messagebox.showwarning("No session", "Please choose a valid session folder.")
|
|
return False
|
|
self.logs[which].delete("1.0", "end")
|
|
self.bars[which]["value"] = 0
|
|
self._set_running(True)
|
|
return True
|
|
|
|
def _spawn(self, fn):
|
|
threading.Thread(target=fn, daemon=True).start()
|
|
|
|
def _start_embed(self):
|
|
session = self.embed_session.get().strip()
|
|
if not self._guard(session, "embed"):
|
|
return
|
|
out = self.embed_out.get().strip() or None
|
|
gps = self.embed_gps.get()
|
|
et = self.embed_exif.get().strip() or None
|
|
workers = max(1, int(self.workers.get()))
|
|
progress, log = self._callbacks("embed")
|
|
|
|
def work():
|
|
try:
|
|
res = embed.embed_session(session, out, gps, et, progress=progress, log=log,
|
|
workers=workers)
|
|
self.q.put(("done", "embed", res))
|
|
except Exception as e: # noqa: BLE001
|
|
self.q.put(("error", "embed", str(e)))
|
|
self._spawn(work)
|
|
|
|
def _read_window_params(self):
|
|
return dict(lo_pct=float(self.th_lo.get()), hi_pct=float(self.th_hi.get()),
|
|
absolute=bool(self.th_absolute.get()))
|
|
|
|
def _start_preview(self):
|
|
session = self.th_session.get().strip()
|
|
if not self._guard(session, "thermal"):
|
|
return
|
|
params = self._read_window_params()
|
|
workers = max(1, int(self.workers.get()))
|
|
progress, log = self._callbacks("thermal")
|
|
|
|
def work():
|
|
try:
|
|
pngs = stretch.list_frames(Path(session))
|
|
win = stretch.compute_window(pngs, progress=progress, workers=workers, **params)
|
|
log(f"Frames: {len(pngs)} | absolute signal {win['abs_lo']}-{win['abs_hi']}")
|
|
log(f"Window: {win['lo']}-{win['hi']} ({win['how']})")
|
|
img = stretch.make_sample_image(pngs, win["lo"], win["hi"])
|
|
self.q.put(("preview", img))
|
|
self.q.put(("done", "thermal", {"kind": "preview", **win}))
|
|
except Exception as e: # noqa: BLE001
|
|
self.q.put(("error", "thermal", str(e)))
|
|
self._spawn(work)
|
|
|
|
def _start_stretch(self):
|
|
session = self.th_session.get().strip()
|
|
if not self._guard(session, "thermal"):
|
|
return
|
|
out = self.th_out.get().strip() or None
|
|
cmap = self.th_cmap.get()
|
|
params = self._read_window_params()
|
|
workers = max(1, int(self.workers.get()))
|
|
embed_exif = bool(self.th_embed.get())
|
|
progress, log = self._callbacks("thermal")
|
|
|
|
def work():
|
|
try:
|
|
pngs = stretch.list_frames(Path(session))
|
|
win = stretch.compute_window(pngs, progress=progress, workers=workers, **params)
|
|
log(f"Window: {win['lo']}-{win['hi']} ({win['how']})")
|
|
res = stretch.stretch_session(session, out, cmap, win["lo"], win["hi"],
|
|
embed_exif=embed_exif,
|
|
progress=progress, log=log, workers=workers)
|
|
self.q.put(("done", "thermal", {"kind": "stretch", **res}))
|
|
except Exception as e: # noqa: BLE001
|
|
self.q.put(("error", "thermal", str(e)))
|
|
self._spawn(work)
|
|
|
|
# ----- preview rendering ---------------------------------------------
|
|
def _show_preview(self, img: Image.Image):
|
|
maxw = max(self.preview_label.winfo_width() - 8, 600)
|
|
if img.width > maxw:
|
|
h = round(img.height * maxw / img.width)
|
|
img = img.resize((maxw, h), Image.LANCZOS)
|
|
self._preview_imgtk = ImageTk.PhotoImage(img)
|
|
self.preview_label.config(image=self._preview_imgtk)
|
|
|
|
# ----- queue pump (runs on the Tk main thread) ------------------------
|
|
def _poll_queue(self):
|
|
try:
|
|
while True:
|
|
msg = self.q.get_nowait()
|
|
kind = msg[0]
|
|
if kind == "log":
|
|
_, which, text = msg
|
|
self.logs[which].insert("end", text + "\n")
|
|
self.logs[which].see("end")
|
|
elif kind == "progress":
|
|
_, which, frac, m = msg
|
|
self.bars[which]["value"] = max(0, min(1000, int(frac * 1000)))
|
|
self.status_vars[which].set(m)
|
|
elif kind == "preview":
|
|
self._show_preview(msg[1])
|
|
elif kind == "map":
|
|
self._render_map(msg[1])
|
|
elif kind == "mapexported":
|
|
self._set_running(False)
|
|
res = msg[1]
|
|
self.map_status.set(
|
|
f"Opened browser map: {res['markers']} points, "
|
|
f"{res['footprints']} footprints.\n{res['path']}")
|
|
elif kind == "maperror":
|
|
self._set_running(False)
|
|
self.map_status.set("Error.")
|
|
messagebox.showerror("FireMapper - map", msg[1])
|
|
elif kind == "done":
|
|
self._on_done(msg[1], msg[2])
|
|
elif kind == "error":
|
|
_, which, m = msg
|
|
self._set_running(False)
|
|
self.status_vars[which].set("Error.")
|
|
messagebox.showerror("FireMapper - error", m)
|
|
except queue.Empty:
|
|
pass
|
|
self.after(100, self._poll_queue)
|
|
|
|
def _on_done(self, which, res):
|
|
self._set_running(False)
|
|
if which == "thermal" and res.get("kind") == "preview":
|
|
self.status_vars["thermal"].set(
|
|
f"Preview ready - window {res['lo']}-{res['hi']} ({res['how']}).")
|
|
return
|
|
self.bars[which]["value"] = 1000
|
|
self.status_vars[which].set("Done.")
|
|
if which == "embed":
|
|
out = Path(res["out_root"])
|
|
text = f"Tagged {res['updated']} of {res['pairs']} images.\n\nOutput:\n{out}"
|
|
else:
|
|
out = Path(res["out_dir"])
|
|
text = f"Wrote {res['frames']} thermal frames.\n\nOutput:\n{out}"
|
|
if messagebox.askyesno("FireMapper - done", text + "\n\nOpen the output folder?"):
|
|
open_in_explorer(out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
FireMapperGUI().mainloop()
|