diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/LamNI_webpage_generator.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/LamNI_webpage_generator.py
index a5a1708..ead6fe3 100644
--- a/csaxs_bec/bec_ipython_client/plugins/LamNI/LamNI_webpage_generator.py
+++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/LamNI_webpage_generator.py
@@ -1,11 +1,11 @@
"""
-LamNI/webpage_generator.py
-===========================
+LamNI/LamNI_webpage_generator.py
+==================================
LamNI-specific webpage generator subclass.
Integration (inside the LamNI __init__ / startup):
---------------------------------------------------
- from csaxs_bec.bec_ipython_client.plugins.LamNI.webpage_generator import (
+ from csaxs_bec.bec_ipython_client.plugins.LamNI.LamNI_webpage_generator import (
LamniWebpageGenerator,
)
self._webpage_gen = LamniWebpageGenerator(
@@ -16,7 +16,7 @@ Integration (inside the LamNI __init__ / startup):
Or use the factory (auto-selects by session name "lamni"):
----------------------------------------------------------
- from csaxs_bec.bec_ipython_client.plugins.flomni.webpage_generator import (
+ from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.webpage_generator_base import (
make_webpage_generator,
)
self._webpage_gen = make_webpage_generator(bec, output_dir="~/data/raw/webpage/")
@@ -32,11 +32,8 @@ Interactive helpers:
from pathlib import Path
-from csaxs_bec.bec_ipython_client.plugins.flomni.webpage_generator import (
+from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.webpage_generator_base import (
WebpageGeneratorBase,
- _safe_get,
- _safe_float,
- _gvar,
)
@@ -49,57 +46,92 @@ class LamniWebpageGenerator(WebpageGeneratorBase):
sample name, and measurement settings.
"""
- # LamNI has no persisted parameter-snapshot job queue (unlike flomni
- # and, later, omny) — the "Tomo queue" card and its global-var read
- # are skipped entirely for this setup.
- HAS_TOMO_QUEUE = False
+ # LamNI's tomo-queue jobs (added via TomoQueueMixin, see
+ # OMNY_shared/tomo_queue_mixin.py) are persisted under the same
+ # "tomo_queue" global var key flomni uses, so the shared "Tomo queue"
+ # UI card and its global-var read work unmodified for this setup too.
+ HAS_TOMO_QUEUE = True
- # LamNI's tomo scans come in two flavours: a 2-subtomo and an
- # 8-subtomo variant, both using the same equally-spaced-grid formula
- # as flomni's type 1 (just with a different sub-tomo count).
- # TODO: replace the placeholder keys (1, 2) with whatever values
- # progress["tomo_type"] actually takes for LamNI once the producer
- # side (LamNI equivalent of flomni.py) defines them.
+ # LamNI's tomo_scan() (lamni.py) implements the same 3-mode scheme as
+ # flomni: type 1 = equally-spaced grid, always 8 sub-tomograms
+ # (`for ii in range(subtomo_start, 9)`), types 2 and 3 = golden-ratio
+ # variants capped by golden_max_number_of_projections. Identical to
+ # FlomniWebpageGenerator's TOMO_TYPES -- confirmed against lamni.py's
+ # actual tomo_scan()/sub_tomo_scan() implementation, not a guess.
TOMO_TYPES = {
- 1: {"kind": "equally_spaced_grid", "n_subtomos": 2},
- 2: {"kind": "equally_spaced_grid", "n_subtomos": 8},
+ 1: {"kind": "equally_spaced_grid", "n_subtomos": 8},
+ 2: {"kind": "golden_capped"},
+ 3: {"kind": "golden_capped"},
}
- # TODO: fill in LamNI-specific device paths
+ # LamNI's own names for the tomo-queue job / "Current measurement" param
+ # table, mirroring LamNI._TOMO_SCAN_PARAM_NAMES (lamni.py) rather than
+ # flomni's fovx/fovy, stitch_x/y (lamni is laminography: a single
+ # circular FOV, no 180 mode, no single-point acquisition).
+ TQ_PARAM_DISPLAY = [
+ ("tomo_type", "Type"),
+ ("tomo_countingtime", "Exposure (s)"),
+ ("tomo_circfov", "Circular FOV (µm)"),
+ ("tomo_angle_stepsize", "Angle step (°)"),
+ ("tomo_shellstep", "Shell step (µm)"),
+ ("lamni_stitch_x", "Stitch x"),
+ ("lamni_stitch_y", "Stitch y"),
+ ("tomo_stitch_overlap", "Stitch overlap"),
+ ("corridor_size", "Corridor size"),
+ ("frames_per_trigger", "Frames/trigger"),
+ ("golden_ratio_bunch_size", "Golden bunch size"),
+ ("golden_max_number_of_projections", "Golden max projections"),
+ ("golden_projections_at_0_deg_for_damage_estimation", "0° damage-check shots"),
+ ("ptycho_reconstruct_foldername", "Recon folder"),
+ ]
+
+ # Global-var names of the tomo parameters to read live for the
+ # 'Current measurement' section -- the TQ_PARAM_DISPLAY keys plus
+ # at_each_angle_hook (rendered specially, only when set; see
+ # buildParamRows() in the shared HTML/JS template). Any name that is
+ # not actually a global var is skipped silently and just does not
+ # appear as a row.
+ _CURRENT_PARAM_KEYS = [key for key, _label in TQ_PARAM_DISPLAY] + ["at_each_angle_hook"]
+
+ # LamNI has no temperature/humidity monitoring device configured
+ # (unlike flomni's flomni_temphum.*) -- leave empty until one exists.
# label -> dotpath under device_manager.devices
- _TEMP_MAP = {
- # "Sample": "lamni_temphum.temperature_sample",
- # "OSA": "lamni_temphum.temperature_osa",
- }
+ _TEMP_MAP = {}
def _logo_path(self):
return Path(__file__).parent / "LamNI.png"
def _collect_setup_data(self) -> dict:
- # ── LamNI-specific data goes here ─────────────────────────────
- # Uncomment and adapt when device names are known:
- #
- # dm = self._bec.device_manager
- # sample_name = _safe_get(dm, "lamni_samples.sample_names.sample0") or "N/A"
- # temperatures = {
- # label: _safe_float(_safe_get(dm, path))
- # for label, path in self._TEMP_MAP.items()
- # }
- # settings = {
- # "Sample name": sample_name,
- # "FOV x / y": ...,
- # "Exposure time": _gvar(self._bec, "tomo_countingtime", ".3f", " s"),
- # "Angle step": _gvar(self._bec, "tomo_angle_stepsize", ".2f", "\u00b0"),
- # }
- # return {
- # "type": "lamni",
- # "sample_name": sample_name,
- # "temperatures": temperatures,
- # "settings": settings,
- # }
+ g = self._bec
+
+ # ── Sample name ───────────────────────────────────────────────
+ # LamNI has no automatic sample changer/device (unlike flomni's
+ # gripper-backed flomni_samples); the active sample is tracked
+ # manually via the sample_name global var (see LamNI.sample_name
+ # property in lamni.py), same default as that property uses.
+ sample_name = g.get_global_var("sample_name") or "bec_test_sample"
+
+ # ── Temperatures ──────────────────────────────────────────────
+ # No lamni temp/humidity device exists yet -- always empty.
+ temperatures = {}
+
+ # ── Current measurement parameters (read live) ────────────────
+ # Same keys the tomo queue stores in job.params, so the 'Current
+ # measurement' section on the page renders through the identical
+ # param-table path (buildParamRows -> fmtTqParam / calcProjections)
+ # and always matches the queue's level of detail.
+ current_params = {}
+ for key in self._CURRENT_PARAM_KEYS:
+ try:
+ val = g.get_global_var(key)
+ except Exception:
+ val = None
+ if val is not None:
+ current_params[key] = val
- # Placeholder — returns minimal info until implemented
return {
- "type": "lamni",
- # LamNI-specific data here
- }
\ No newline at end of file
+ "type": "lamni",
+ "sample_name": sample_name,
+ "temperatures": temperatures,
+ "current_params": current_params,
+ }
diff --git a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py
index c11a96e..7340e18 100644
--- a/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py
+++ b/csaxs_bec/bec_ipython_client/plugins/LamNI/lamni.py
@@ -218,6 +218,22 @@ class LamNI(TomoQueueMixin, LamNIAlignmentMixin, LamNIOpticsMixin, LamniGuiTools
self._progress_proxy = _ProgressProxy(self.client)
self._init_tomo_queue()
+ from csaxs_bec.bec_ipython_client.plugins.LamNI.LamNI_webpage_generator import (
+ LamniWebpageGenerator,
+ )
+
+ self._webpage_gen = LamniWebpageGenerator(
+ bec_client=client,
+ output_dir="~/data/raw/webpage/",
+ upload_url="https://v1p0zyg2w9n2k9c1.myfritz.net/upload.php",
+ local_port=8080,
+ )
+ self._webpage_gen.start()
+
+ def set_web_password(self, password: str) -> None:
+ """Set the web password for the current BEC account."""
+ self._webpage_gen.set_web_password(password)
+
# ------------------------------------------------------------------
# Special angles
# ------------------------------------------------------------------
diff --git a/csaxs_bec/bec_ipython_client/plugins/OMNY_shared/webpage_generator_base.py b/csaxs_bec/bec_ipython_client/plugins/OMNY_shared/webpage_generator_base.py
new file mode 100644
index 0000000..8e694af
--- /dev/null
+++ b/csaxs_bec/bec_ipython_client/plugins/OMNY_shared/webpage_generator_base.py
@@ -0,0 +1,3027 @@
+"""
+OMNY_shared/webpage_generator_base.py
+======================================
+Background thread that reads tomo progress from the BEC global variable store
+and writes status.json (every cycle) + status.html (once at startup) to a
+staging directory. An optional HttpUploader sends those files to a web host
+after every cycle, running in a separate daemon thread so uploads never block
+the generator cycle. A built-in LocalHttpServer always serves the output
+directory locally (default port 8080) so the page can be accessed on the
+lab network without any extra setup.
+
+Extracted from flomni/flomni_webpage_generator.py so the common logic can be
+shared across setups (flomni, lamni, ...), the same way the tomo-queue
+backend was shared via OMNY_shared/tomo_queue_mixin.py -- a
+behavior-preserving refactor (moved code, not changed logic), aside from
+making TQ_PARAM_DISPLAY a per-setup override alongside the existing
+HAS_TOMO_QUEUE / TOMO_TYPES flags.
+
+Architecture
+------------
+LocalHttpServer -- built-in HTTP server; serves output_dir on port 8080.
+ Always started at _launch(); URL printed to console.
+HttpUploader -- non-blocking HTTP uploader (fire-and-forget thread).
+ Tracks file mtimes; only uploads changed files.
+ Sends a cleanup request to the server when the
+ ptycho scan ID changes (removes old S*_*.png/jpg).
+WebpageGeneratorBase -- all common logic: queue, progress, idle detection,
+ reconstruction queue, HTML, audio, phone numbers,
+ outdated-page warning. Import this in subclasses.
+make_webpage_generator() -- factory; selects the right class by session name.
+ Lazy-imports setup-specific subclasses to avoid
+ circular dependencies.
+
+Per-setup subclasses live next to their plugin (e.g.
+flomni/flomni_webpage_generator.py's FlomniWebpageGenerator,
+LamNI/LamNI_webpage_generator.py's LamniWebpageGenerator) and import
+WebpageGeneratorBase from this module.
+
+Integration (inside a setup's plugin __init__, after progress reset):
+-----------------------------------------------------------------------
+ from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import (
+ FlomniWebpageGenerator,
+ )
+ self._webpage_gen = FlomniWebpageGenerator(
+ bec_client=client,
+ output_dir="~/data/raw/webpage/",
+ upload_url="http://omny.online/upload.php", # optional
+ local_port=8080, # optional, default 8080
+ )
+ self._webpage_gen.start()
+ # On start(), the console prints:
+ # ➜ Status page: http://hostname:8080/status.html
+
+Interactive helpers (optional, in the iPython session):
+-------------------------------------------------------
+ flomni._webpage_gen.status() # print current status + local URL
+ flomni._webpage_gen.verbosity = 2 # VERBOSE: one-line summary per cycle
+ flomni._webpage_gen.verbosity = 3 # DEBUG: full JSON per cycle
+ flomni._webpage_gen.stop() # release lock, stop local server
+ flomni._webpage_gen.start() # restart after stop()
+ flomni._webpage_gen.set_web_password("pw") # set web password for current e-account
+
+Session authentication:
+-----------------------
+Two htpasswd files are used on the remote server:
+ users.htpasswd -- static, managed manually with htpasswd -B
+ session.htpasswd -- managed by the generator; one entry for the current e-account
+
+At _launch(), the generator queries session_query.php to find which account was
+previously active. If the account has changed the session.htpasswd is cleared
+immediately (old user loses access) and uploaded. The new user calls
+set_web_password() to set their own password and gain access.
+
+Account-mismatch fallback (dev/sim sessions):
+----------------------------------------------
+If bec_client.active_account doesn't match the OS login (_check_account_match)
+-- typically a leftover session under the wrong experiment account, or a dev/sim
+session run under a personal account -- start() no longer does nothing. It
+starts a *local-only* status page on a separate port (default 8081, see
+mismatch_local_port) for local debugging: no singleton lock (so it can never
+block the real session's takeover), no remote upload, no session.htpasswd
+changes. Look for "[local-only: account mismatch]" in the startup log / status().
+"""
+
+import datetime
+import functools
+import http.server
+import json
+import os
+import shutil
+import socket
+import threading
+import time
+from pathlib import Path
+
+try:
+ from PIL import Image as _PILImage
+ _PIL_AVAILABLE = True
+except ImportError:
+ _PIL_AVAILABLE = False
+
+from bec_lib import bec_logger
+
+logger = bec_logger.logger
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+_LOCK_VAR_KEY = "webpage_generator_lock"
+_LOCK_STALE_AFTER_S = 35 # just over 2 missed cycles (2×15s+5s margin)
+_CYCLE_INTERVAL_S = 15
+_TOMO_HEARTBEAT_STALE_S = 90
+
+# Fallback port used when the BEC account doesn't match the system user (see
+# _check_account_match / start()). Deliberately different from the default
+# local_port so a local-only fallback session never collides with a
+# legitimately-started one on the same host.
+_MISMATCH_LOCAL_PORT = 8081
+
+# Path where online ptycho reconstructions are written by the analysis pipeline.
+# Subfolders follow the pattern dset_NNNNN and contain S#####_*_phase.png etc.
+_PTYCHO_ONLINE_DIR = "~/data/raw/analysis/online/ptycho"
+_PTYCHO_THUMB_PX = 200 # longest edge of generated thumbnail in pixels
+
+VERBOSITY_SILENT = 0
+VERBOSITY_NORMAL = 1
+VERBOSITY_VERBOSE = 2
+VERBOSITY_DEBUG = 3
+
+_PHONE_NUMBERS = [
+ ("Beamline", "+41 56 310 5845"),
+ ("Control room", "+41 56 310 5503"),
+]
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _now_iso() -> str:
+ return datetime.datetime.now().isoformat(timespec="seconds")
+
+def _epoch() -> float:
+ return time.time()
+
+def _heartbeat_age_s(iso_str) -> float:
+ if iso_str is None:
+ return float("inf")
+ try:
+ ts = datetime.datetime.fromisoformat(iso_str)
+ return (datetime.datetime.now() - ts).total_seconds()
+ except Exception:
+ return float("inf")
+
+def _format_duration(seconds) -> str:
+ if seconds is None:
+ return "N/A"
+ try:
+ seconds = int(float(seconds))
+ except (TypeError, ValueError):
+ return "N/A"
+ h, remainder = divmod(seconds, 3600)
+ m, s = divmod(remainder, 60)
+ if h > 0:
+ return f"{h}h {m:02d}m {s:02d}s"
+ if m > 0:
+ return f"{m}m {s:02d}s"
+ return f"{s}s"
+
+def _check_account_match(bec_client) -> bool:
+ try:
+ active = bec_client.active_account
+ system_user = os.getenv("USER") or os.getlogin()
+ return active[1:] == system_user[1:]
+ except Exception:
+ return True
+
+def _safe_get(device_manager, dotpath: str):
+ """Navigate a dotted device path and call .get(), returning None on any error."""
+ try:
+ obj = device_manager.devices
+ for part in dotpath.split("."):
+ obj = getattr(obj, part)
+ return obj.get(cached=True)
+ except Exception:
+ return None
+
+def _safe_float(val, ndigits: int = 2):
+ try:
+ return round(float(val), ndigits)
+ except (TypeError, ValueError):
+ return None
+
+def _gvar(bec_client, key, fmt=None, suffix=""):
+ val = bec_client.get_global_var(key)
+ if fmt is None:
+ return val
+ if val is None:
+ return "N/A"
+ try:
+ return f"{val:{fmt}}{suffix}"
+ except Exception:
+ return str(val)
+
+def _read_beamline_states_info(bec_client) -> dict:
+ """Return a diagnostic snapshot of all configured beamline states, for
+ display only (NOT used to derive the 'blocked' experiment status — that
+ is computed from the queue's own locks via _read_queue_locks(), which is
+ the authoritative source; see its docstring for why).
+
+ Returns:
+ {
+ "enabled": bool | None, # bec.builtin_actors.scan_interlock.enabled
+ "states": [
+ {
+ "name": str,
+ "status": str, # 'valid' | 'invalid' | 'warning' | 'unknown'
+ "label": str,
+ "watched": bool, # True if present in states_watched
+ "accepted": list[str] | None, # accepted statuses if watched
+ "mismatched": bool, # True if watched AND status not in accepted
+ },
+ ...
+ ],
+ }
+
+ Beamline states are registered dynamically and can differ from
+ experiment to experiment, so names are read from the manager's _states
+ dict (the same one show_all() iterates over internally — there is no
+ public "list names" method as of this BEC version). states_watched and
+ enabled are read through the public scan_interlock HLI properties.
+ """
+ manager = getattr(bec_client, "beamline_states", None)
+ if manager is None:
+ return {"enabled": None, "states": []}
+
+ try:
+ names = list(getattr(manager, "_states", {}).keys())
+ except Exception:
+ names = []
+
+ try:
+ interlock = bec_client.builtin_actors.scan_interlock
+ enabled = interlock.enabled
+ states_watched = interlock.states_watched or {}
+ except Exception:
+ enabled = None
+ states_watched = {}
+
+ states = []
+ for name in names:
+ try:
+ state_obj = getattr(manager, name, None)
+ if state_obj is None:
+ continue
+ info = state_obj.get() # {"status": ..., "label": ...}
+ status = info.get("status", "unknown")
+ label = info.get("label", name)
+ except Exception:
+ status, label = "unknown", name
+
+ accepted = states_watched.get(name)
+ watched = accepted is not None
+ mismatched = watched and status not in accepted
+
+ states.append({
+ "name": name,
+ "status": status,
+ "label": label,
+ "watched": watched,
+ "accepted": accepted,
+ "mismatched": mismatched,
+ })
+
+ return {"enabled": enabled, "states": states}
+
+
+def _read_queue_locks(primary) -> list:
+ """Return a list of {"identifier": str, "reason": str} for every lock
+ currently applied to the primary scan queue.
+
+ BEC's scan_queue.status becomes "LOCKED" (saving the prior status to
+ restore later) whenever any lock is added, e.g. by the ScanInterlockActor
+ when a watched beamline state goes out of spec (see scan_interlock.py:
+ add_queue_lock(queue="primary", reason=..., lock_id="ScanInterlockActor")).
+ Reading the lock list directly is more robust than re-deriving "blocked"
+ from beamline states ourselves: it reflects whatever is actually holding
+ the queue, regardless of which actor (interlock or otherwise) caused it,
+ and survives changes to which beamline states are configured/watched.
+ """
+ if primary is None:
+ return []
+ try:
+ return [
+ {"identifier": lock.identifier, "reason": lock.reason}
+ for lock in (primary.locks or [])
+ ]
+ except Exception:
+ return []
+
+
+# ---------------------------------------------------------------------------
+# Status derivation
+# ---------------------------------------------------------------------------
+
+def _derive_status(
+ progress: dict,
+ queue_has_active_scan: bool,
+ last_active_time,
+ had_activity: bool,
+ queue_locks: list | None = None,
+ beamline_blocking: bool = False,
+) -> str:
+ """
+ Returns one of:
+ scanning -- tomo heartbeat fresh (< _TOMO_HEARTBEAT_STALE_S), OR
+ heartbeat recently seen AND queue still has active scan
+ (handles long individual projections > heartbeat timeout)
+ blocked -- the experiment is being held up externally: EITHER a
+ watched beamline state is out of its accepted range while
+ the scan interlock would act on it (beamline_blocking),
+ OR the primary queue has at least one explicit lock
+ applied (queue_locks). Reported regardless of whether a
+ scan is currently executing: when a block hits, the
+ running scan is interrupted and its repeat waits for the
+ block to clear, so active_request_block (and hence
+ queue_has_active_scan) may already be cleared even though
+ the beamline is still blocking. Only 'scanning' (fresh
+ heartbeat = data still flowing) takes precedence.
+ running -- queue has active scan but heartbeat has never been seen,
+ and nothing is blocking
+ idle -- not scanning, not blocked, last_active_time known
+ unknown -- no activity ever seen since generator started
+
+ 'unknown' is ONLY returned before any scan activity has been observed.
+ Once activity has been seen the status goes directly:
+ scanning/running/blocked -> idle
+ never through 'unknown'.
+ """
+ hb_age = _heartbeat_age_s(progress.get("heartbeat"))
+ if hb_age < _TOMO_HEARTBEAT_STALE_S:
+ return "scanning"
+ # External block takes precedence over the remaining scan/idle states.
+ # A blocked scan is interrupted (active_request_block cleared) while its
+ # repeat waits for the block to clear, so we must NOT require an active
+ # scan here — that was the previous behaviour and it made a blocked-but-
+ # interrupted experiment fall through to 'idle'.
+ if beamline_blocking or queue_locks:
+ return "blocked"
+ # Heartbeat stale but queue still active and heartbeat was seen recently
+ # enough (within 10× the stale window) — likely a long projection, not idle.
+ # Report as 'scanning' so audio system does not trigger a false alarm.
+ if queue_has_active_scan and hb_age < _TOMO_HEARTBEAT_STALE_S * 10:
+ return "scanning"
+ if queue_has_active_scan:
+ return "running"
+ if last_active_time is not None or had_activity:
+ return "idle"
+ return "unknown"
+
+
+# ---------------------------------------------------------------------------
+# HTTP Uploader (non-blocking, fire-and-forget)
+# ---------------------------------------------------------------------------
+
+class HttpUploader:
+ """
+ Uploads files from a local directory to a remote server via HTTP POST.
+
+ Uploads run in a daemon thread so they never block the generator cycle.
+ If an upload is already in progress when the next cycle fires, the new
+ upload request is dropped (logged at DEBUG level) rather than queuing up.
+
+ File tracking:
+ - Tracks mtime of each file; only re-uploads files that have changed.
+ - upload_dir() -- uploads ALL eligible files (called once at start).
+ - upload_changed() -- uploads only files whose mtime has changed.
+
+ Scan change cleanup:
+ - cleanup_ptycho_images() -- sends action=cleanup POST to the server,
+ asking it to delete all S*_*.png and S*_*.jpg files.
+ Called automatically by the generator when the ptycho scan ID changes.
+
+ The server-side upload.php must:
+ - Accept POST with multipart file upload (field name 'file').
+ - Accept POST with action=cleanup to delete ptycho image files.
+ - Enforce IP-based access control (129.129.122.x).
+ - Validate filename with regex to block path traversal.
+ """
+
+ _UPLOAD_SUFFIXES = {".html", ".json", ".jpg", ".png", ".pdf", ".txt"}
+
+ def __init__(self, url: str, timeout: float = 20.0):
+ self._url = url
+ self._timeout = timeout
+ self._uploaded: dict[str, float] = {} # abs path -> mtime at last upload
+ self._lock = threading.Lock()
+ self._busy = False # True while an upload thread is running
+ self._warn_at: dict[str, float] = {} # key -> epoch of last warning
+
+ _WARN_COOLDOWN_S = 600 # only repeat the same warning once per minute
+
+ def _warn(self, key: str, msg: str) -> None:
+ """Log a warning at most once per _WARN_COOLDOWN_S for a given key."""
+ now = _epoch()
+ if now - self._warn_at.get(key, 0) >= self._WARN_COOLDOWN_S:
+ self._warn_at[key] = now
+ logger.warning(msg)
+
+ # ── Public API ──────────────────────────────────────────────────────────
+
+ def upload_dir_async(self, directory: Path) -> None:
+ """Upload ALL eligible files in directory, in a background thread."""
+ files = self._eligible_files(directory)
+ self._dispatch(self._upload_files, files, True)
+
+ def upload_changed_async(self, directory: Path) -> None:
+ """Upload only changed files in directory, in a background thread."""
+ files = self._changed_files(directory)
+ if files:
+ self._dispatch(self._upload_files, files, False)
+
+ def upload_file_async(self, path: Path) -> None:
+ """Upload a single specific file, bypassing suffix whitelist and mtime check."""
+ self._dispatch(self._upload_files, [Path(path)], True)
+
+ def cleanup_ptycho_images_async(self) -> None:
+ """Ask the server to delete all S*_*.png / S*_*.jpg files (background)."""
+ self._dispatch(self._do_cleanup)
+
+ # ── Internal ────────────────────────────────────────────────────────────
+
+ def _dispatch(self, fn, *args) -> None:
+ """Run fn(*args) in a daemon thread. Drops the request if already busy."""
+ with self._lock:
+ if self._busy:
+ logger.debug("HttpUploader: previous upload still running, skipping")
+ return
+ self._busy = True
+
+ def _run():
+ try:
+ fn(*args)
+ finally:
+ with self._lock:
+ self._busy = False
+
+ t = threading.Thread(target=_run, name="HttpUploader", daemon=True)
+ t.start()
+
+ def _eligible_files(self, directory: Path) -> list:
+ result = []
+ for path in Path(directory).iterdir():
+ if path.is_file() and path.suffix in self._UPLOAD_SUFFIXES:
+ result.append(path)
+ return result
+
+ def _changed_files(self, directory: Path) -> list:
+ result = []
+ for path in self._eligible_files(directory):
+ try:
+ mtime = path.stat().st_mtime
+ except OSError:
+ continue
+ if self._uploaded.get(str(path)) != mtime:
+ result.append(path)
+ return result
+
+ def _upload_files(self, files: list, force: bool = False) -> None:
+ try:
+ import requests as _requests
+ import urllib3
+ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+ except ImportError:
+ self._warn("no_requests", "HttpUploader: 'requests' library not installed")
+ return
+
+ for path in files:
+ try:
+ mtime = path.stat().st_mtime
+ except OSError:
+ continue
+ # Double-check mtime unless forced (upload_dir uses force=True)
+ if not force and self._uploaded.get(str(path)) == mtime:
+ continue
+ try:
+ with open(path, "rb") as f:
+ r = _requests.post(
+ self._url,
+ files={"file": (path.name, f)},
+ timeout=self._timeout,
+ verify=False, # accept self-signed / untrusted certs
+ )
+ if r.status_code == 200:
+ self._uploaded[str(path)] = mtime
+ self._warn_at.pop(f"upload_{path.name}", None) # clear on success
+ logger.debug(f"HttpUploader: OK {path.name}")
+ else:
+ self._warn(
+ f"upload_{path.name}",
+ f"HttpUploader: {path.name} -> HTTP {r.status_code}: "
+ f"{r.text[:120]}"
+ )
+ except Exception as exc:
+ self._warn(f"upload_{path.name}", f"HttpUploader: {path.name} failed: {exc}")
+
+ def _do_cleanup(self) -> None:
+ try:
+ import requests as _requests
+ import urllib3
+ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+ except ImportError:
+ return
+ try:
+ r = _requests.post(
+ self._url,
+ data={"action": "cleanup"},
+ timeout=self._timeout,
+ verify=False, # accept self-signed / untrusted certs
+ )
+ logger.info(f"HttpUploader cleanup: {r.text[:120]}")
+ self._warn_at.pop("cleanup", None) # clear on success
+ # Forget mtime records for ptycho files so they get re-uploaded
+ with self._lock:
+ to_remove = [k for k in self._uploaded if "/S" in k or "\\S" in k]
+ for k in to_remove:
+ self._uploaded.pop(k, None)
+ except Exception as exc:
+ self._warn("cleanup", f"HttpUploader cleanup failed: {exc}")
+
+
+# ---------------------------------------------------------------------------
+# Local HTTP server (serves output_dir over http://hostname:port/)
+# ---------------------------------------------------------------------------
+
+class LocalHttpServer:
+ """
+ Serves the generator's output directory over plain HTTP in a daemon thread.
+
+ Uses Python's built-in http.server — no extra dependencies.
+ Request logging is suppressed so the BEC console stays clean.
+ The server survives stop()/start() cycles: _launch() creates a fresh
+ instance each time start() is called.
+
+ Usage:
+ srv = LocalHttpServer(output_dir, port=8080)
+ srv.start()
+ print(srv.url) # http://hostname:8080/status.html
+ srv.stop()
+ """
+
+ def __init__(self, directory: Path, port: int = 8080):
+ self._directory = Path(directory)
+ self._port = port
+ self._server = None
+ self._thread = None
+
+ # ── silence the per-request log lines in the iPython console ──────────
+ class _QuietHandler(http.server.SimpleHTTPRequestHandler):
+ def log_message(self, *args):
+ pass
+ # handle_error here does nothing — wrong class
+
+ class _QuietHTTPServer(http.server.HTTPServer):
+ def handle_error(self, request, client_address):
+ pass # suppress BrokenPipeError and all other per-connection noise
+
+ def start(self) -> None:
+ Handler = functools.partial(
+ self._QuietHandler,
+ directory=str(self._directory),
+ )
+ try:
+ self._server = self._QuietHTTPServer(("", self._port), Handler)
+ except OSError as exc:
+ raise RuntimeError(
+ f"LocalHttpServer: cannot bind port {self._port}: {exc}"
+ ) from exc
+ self._thread = threading.Thread(
+ target=self._server.serve_forever,
+ name="LocalHttpServer",
+ daemon=True,
+ )
+ self._thread.start()
+
+ def stop(self) -> None:
+ if self._server is not None:
+ self._server.shutdown() # blocks until serve_forever() returns
+ self._server = None
+
+ def is_alive(self) -> bool:
+ return self._thread is not None and self._thread.is_alive()
+
+ @property
+ def port(self) -> int:
+ return self._port
+
+ @property
+ def url(self) -> str:
+ """Best-guess URL for printing. Uses the machine's hostname."""
+ return f"http://{socket.gethostname()}:{self._port}/status.html"
+
+
+# ---------------------------------------------------------------------------
+# Base generator
+# ---------------------------------------------------------------------------
+
+class WebpageGeneratorBase:
+ """
+ Common webpage generator. Subclass and override:
+ _collect_setup_data() -- return dict of instrument-specific data
+ _logo_path() -- return Path to logo PNG, or None for text fallback
+
+ Setup capability flags (override on subclass as needed):
+ HAS_TOMO_QUEUE -- whether this instrument exposes the persisted
+ parameter-snapshot job queue (global var
+ "tomo_queue" + the "Tomo queue" UI card).
+ NOT to be confused with BEC's own primary scan
+ queue (queue_status / queue_locks in _cycle()),
+ which every setup has and which stays common.
+ TOMO_TYPES -- dict describing this instrument's tomography
+ scan types, keyed by the value progress["tomo_type"]
+ takes. Used by the UI to compute total-projection
+ counts for queued jobs without hardcoding a
+ per-instrument formula in JS. Recognised "kind"s:
+ "equally_spaced_grid" (params: n_subtomos)
+ total = floor(180 / angle_stepsize) * n_subtomos
+ (independent of angle_range - see
+ Flomni._subtomo_angle_plan()'s docstring for
+ why: in 360-degree mode, angle_range only
+ splits the n_subtomos into complementary
+ low/high halves, it does not change the
+ total count)
+ "golden_capped"
+ total = golden_max_number_of_projections
+ TQ_PARAM_DISPLAY -- list of (global_var_key, display_label) pairs, in
+ display order, for the tomo-queue job detail table
+ and the "Current measurement" section. Override on
+ subclasses whose global-var names differ (e.g.
+ lamni's tomo_circfov / lamni_stitch_x/y instead of
+ flomni's fovx/fovy / stitch_x/y).
+ """
+
+ HAS_TOMO_QUEUE = True
+ TOMO_TYPES = {
+ 1: {"kind": "equally_spaced_grid", "n_subtomos": 8},
+ 2: {"kind": "golden_capped"},
+ 3: {"kind": "golden_capped"},
+ }
+ TQ_PARAM_DISPLAY = [
+ ("tomo_type", "Type"),
+ ("tomo_countingtime", "Exposure (s)"),
+ ("fovx", "FOV x (µm)"),
+ ("fovy", "FOV y (µm)"),
+ ("tomo_angle_stepsize", "Angle step (°)"),
+ ("tomo_angle_range", "Angle range (°)"),
+ ("tomo_shellstep", "Shell step (µm)"),
+ ("stitch_x", "Stitch x"),
+ ("stitch_y", "Stitch y"),
+ ("frames_per_trigger", "Frames/trigger"),
+ ("single_point_instead_of_fermat_scan", "Single point"),
+ ("ptycho_reconstruct_foldername", "Recon folder"),
+ ]
+
+ def __init__(
+ self,
+ bec_client,
+ output_dir: str = "~/data/raw/webpage/",
+ cycle_interval: float = _CYCLE_INTERVAL_S,
+ verbosity: int = VERBOSITY_NORMAL,
+ upload_url: str = None,
+ local_port: int = 8080,
+ mismatch_local_port: int = _MISMATCH_LOCAL_PORT,
+ ):
+ self._bec = bec_client
+ self._output_dir = Path(output_dir).expanduser().resolve()
+ self._cycle_interval = cycle_interval
+ self._verbosity = verbosity
+ self._uploader = HttpUploader(upload_url) if upload_url else None
+ self._local_port = local_port
+ self._mismatch_local_port = mismatch_local_port
+ self._local_server = None # created fresh each _launch()
+ self._local_only = False # True while running the account-mismatch fallback
+
+ # Derive companion URLs from upload_url
+ if upload_url is not None:
+ self._session_query_url = upload_url.replace("upload.php", "session_query.php")
+ self._set_password_url = upload_url.replace("upload.php", "set_password.php")
+ else:
+ self._session_query_url = None
+ self._set_password_url = None
+
+ self._thread = None
+ self._stop_event = threading.Event()
+ self._last_active_time = None # epoch of last tomo/queue activity
+ self._last_ptycho_scan = None # scan id of last copied ptycho images
+ self._had_activity = False # True once any activity has been observed
+ self._last_queue_id = None # tracks queue history changes between cycles
+ self._owner_id = f"{socket.gethostname()}:{os.getpid()}"
+
+ # ------------------------------------------------------------------
+ # Public API
+ # ------------------------------------------------------------------
+
+ def start(self) -> None:
+ """Start the generator.
+
+ If this session already holds the thread, does nothing.
+ If another session holds a fresh lock, a background watcher thread
+ is launched that polls every 3 s and takes over as soon as the lock
+ goes stale — no blocking, no second call to start() needed.
+ Progress messages during the wait are printed at verbosity >= 2 only.
+ """
+ if not _check_account_match(self._bec):
+ if self._thread is not None and self._thread.is_alive():
+ self._log(VERBOSITY_NORMAL, "WebpageGenerator already running in this session.")
+ return
+ self._log(VERBOSITY_NORMAL,
+ "WebpageGenerator: BEC account does not match system user. "
+ f"Starting a local-only status page on port "
+ f"{self._mismatch_local_port} instead (no singleton lock, no "
+ "remote upload, no session-auth changes) — likely a dev/sim "
+ "session running under a different account.", level="warning")
+ self._launch(local_only=True)
+ return
+
+ if self._thread is not None and self._thread.is_alive():
+ self._log(VERBOSITY_NORMAL, "WebpageGenerator already running in this session.")
+ return
+
+ if not self._acquire_lock():
+ # Lock is fresh — _acquire_lock() has spawned a background watcher
+ # that will call _launch() automatically once the lock expires.
+ return
+
+ self._launch()
+
+ def _launch(self, local_only: bool = False) -> None:
+ """Actually start the worker thread and write static files. Called by
+ start() on the normal path, or by the lock-watcher thread after takeover.
+
+ local_only=True is the account-mismatch fallback (see start()): serves
+ the status page on a separate port for local debugging only, and skips
+ everything that assumes the active BEC account is the "real" one for
+ this host — the singleton lock (a mismatched account shouldn't contend
+ for it against the legitimate session) and remote upload / session-auth
+ (writing session.htpasswd or uploading under the wrong account could
+ step on the real experiment's web page).
+ """
+ self._local_only = local_only
+ self._output_dir.mkdir(parents=True, exist_ok=True)
+
+ # Copy logo once at startup — HTML is also written once here,
+ # not every cycle, since it is a static shell that only loads status.json.
+ logo_filename = self._copy_logo()
+ (self._output_dir / "status.html").write_text(
+ _render_html(
+ _PHONE_NUMBERS,
+ self.HAS_TOMO_QUEUE,
+ self.TOMO_TYPES,
+ self.TQ_PARAM_DISPLAY,
+ logo_filename,
+ )
+ )
+
+ if not local_only:
+ # Check whether the active BEC account matches the session user on
+ # the remote server. If not, clear session.htpasswd so the old
+ # user loses web access immediately. The new user then calls
+ # set_web_password().
+ self._check_session_user()
+
+ # Start local HTTP server (always on; a fresh instance per _launch()).
+ port = self._mismatch_local_port if local_only else self._local_port
+ if self._local_server is not None and self._local_server.is_alive():
+ self._local_server.stop()
+ self._local_server = LocalHttpServer(self._output_dir, port)
+ try:
+ self._local_server.start()
+ local_url_msg = f" local={self._local_server.url}"
+ except RuntimeError as exc:
+ local_url_msg = f" local=ERROR({exc})"
+ self._log(VERBOSITY_NORMAL, str(exc), level="warning")
+
+ # Upload static files (html + logo) once at startup
+ if not local_only and self._uploader is not None:
+ self._uploader.upload_dir_async(self._output_dir)
+
+ self._stop_event.clear()
+ self._thread = threading.Thread(
+ target=self._run, name="WebpageGenerator", daemon=True
+ )
+ self._thread.start()
+ mode_note = " [local-only: account mismatch]" if local_only else ""
+ self._log(VERBOSITY_NORMAL,
+ f"WebpageGenerator started{mode_note} owner={self._owner_id} "
+ f"output={self._output_dir} interval={self._cycle_interval}s"
+ + (f" upload={self._uploader._url}"
+ if (self._uploader and not local_only) else " upload=disabled")
+ + f"\n ➜ Status page:{local_url_msg}")
+
+ def stop(self) -> None:
+ """Stop the generator thread, local HTTP server, and release the singleton lock."""
+ self._stop_event.set()
+ if self._thread is not None:
+ self._thread.join(timeout=self._cycle_interval + 5)
+ # Always clear the reference so start() gets a clean slate,
+ # even if the thread did not exit within the join timeout.
+ self._thread = None
+ if self._local_server is not None:
+ self._local_server.stop()
+ self._local_server = None
+ self._release_lock()
+ self._local_only = False
+ self._log(VERBOSITY_NORMAL, "WebpageGenerator stopped.")
+
+ @property
+ def verbosity(self) -> int:
+ return self._verbosity
+
+ @verbosity.setter
+ def verbosity(self, val: int) -> None:
+ self._verbosity = val
+ self._log(VERBOSITY_NORMAL, f"WebpageGenerator verbosity -> {val}")
+
+ def status(self) -> None:
+ """Print a human-readable status summary to the console."""
+ lock = self._read_lock()
+ running = self._thread is not None and self._thread.is_alive()
+ local = self._local_server.url if (self._local_server and self._local_server.is_alive()) else "stopped"
+ remote_user = self._get_remote_session_user() if self._session_query_url else "n/a"
+ try:
+ active_account = self._bec.active_account
+ except Exception:
+ active_account = "unknown"
+ print(
+ f"WebpageGenerator\n"
+ f" This session running : {running}"
+ + (" [local-only: account mismatch]" if self._local_only else "") + "\n"
+ f" Lock owner : {lock.get('owner_id', 'none')}\n"
+ f" Lock heartbeat : {lock.get('heartbeat', 'never')}\n"
+ f" Output dir : {self._output_dir}\n"
+ f" Local URL : {local}\n"
+ f" Cycle interval : {self._cycle_interval}s\n"
+ f" Upload URL : {self._uploader._url if self._uploader else 'disabled'}\n"
+ f" Session query URL : {self._session_query_url or 'disabled'}\n"
+ f" BEC active account : {active_account}\n"
+ f" Server session user : {remote_user}\n"
+ f" Verbosity : {self._verbosity}\n"
+ )
+
+ # ------------------------------------------------------------------
+ # Logo
+ # ------------------------------------------------------------------
+
+ def _logo_path(self):
+ """
+ Return a Path to the logo PNG for this setup, or None for text fallback.
+ Override in subclasses.
+ """
+ return None
+
+ def _copy_logo(self) -> str | None:
+ """Copy the setup logo into output_dir under its own filename (the
+ source PNG's basename, e.g. "flOMNI.png" / "LamNI.png"), and return
+ that filename for the HTML to reference -- or None if unavailable
+ (renders the text fallback instead).
+
+ Deliberately NOT a fixed "logo.png" for every setup: flomni and lamni
+ share the same output_dir/port (they never run simultaneously), so a
+ shared filename means a browser that already cached logo.png from one
+ instrument's session keeps showing those cached bytes after the other
+ instrument overwrites the file -- same URL, same cache key, no way
+ for the browser to tell the content changed short of a conditional
+ GET matching exactly (which even a `Last-Modified` bump doesn't
+ reliably guarantee, e.g. proxies/extensions with more aggressive
+ heuristic caching). Giving each setup's logo a distinct filename
+ means the two instruments never share a cache key at all -- switching
+ setups is a different URL, so there is nothing to go stale.
+ """
+ src = self._logo_path()
+ if src is None:
+ return None
+ src = Path(src)
+ if not src.exists():
+ self._log(VERBOSITY_VERBOSE,
+ f"Logo not found at {src}, using text fallback.", level="warning")
+ return None
+ try:
+ shutil.copy(src, self._output_dir / src.name)
+ self._log(VERBOSITY_VERBOSE, f"Logo copied from {src}")
+ return src.name
+ except Exception as exc:
+ self._log(VERBOSITY_NORMAL,
+ f"Failed to copy logo: {exc}", level="warning")
+ return None
+
+ # ------------------------------------------------------------------
+ # Session authentication
+ # ------------------------------------------------------------------
+
+ def _get_remote_session_user(self) -> "str | None":
+ """GET session_query.php and return the username in session.htpasswd, or None."""
+ if self._session_query_url is None:
+ return None
+ try:
+ import requests as _requests
+ import urllib3
+ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+ r = _requests.get(self._session_query_url, timeout=10, verify=False)
+ if r.status_code == 200:
+ return r.json().get("user") # str or None
+ self._log(VERBOSITY_VERBOSE,
+ f"session_query: HTTP {r.status_code}", level="warning")
+ except Exception as exc:
+ self._log(VERBOSITY_VERBOSE,
+ f"session_query failed: {exc}", level="warning")
+ return None
+
+ def _check_session_user(self) -> None:
+ """Compare BEC active account with the server session user.
+
+ - Match: silent, nothing to do.
+ - No remote user: current user has no web auth yet — print reminder.
+ - Mismatch: clear session.htpasswd immediately (old user loses access),
+ then print reminder that current user has no web auth yet.
+ """
+ if self._session_query_url is None:
+ return # no remote server configured
+
+ try:
+ current_account = self._bec.active_account # e.g. 'p23092'
+ except Exception as exc:
+ self._log(VERBOSITY_VERBOSE,
+ f"session check: cannot read active_account: {exc}", level="warning")
+ return
+
+ remote_user = self._get_remote_session_user()
+
+ if remote_user == current_account:
+ return # all good, stay silent
+
+ if remote_user is not None:
+ # Different user still has web access — remove it immediately.
+ print(f"WebpageGenerator: account changed "
+ f"'{remote_user}' -> '{current_account}' "
+ f"— removing web access for '{remote_user}'")
+ self._clear_session_htpasswd()
+
+ # Either no previous user or a different user — either way the
+ # current account has no web authentication yet.
+ _BOLD_RED = "\033[1;31m"
+ _RESET = "\033[0m"
+ print(f"{_BOLD_RED}***\nWebpageGenerator: '{current_account}' has no web authentication yet.{_RESET}")
+ print(f"{_BOLD_RED} To set a password run: flomni.set_web_password(\"your_password\")\n***{_RESET}")
+
+ def _clear_session_htpasswd(self) -> None:
+ """Write an empty session.htpasswd locally and upload it."""
+ session_path = self._output_dir / "session.htpasswd"
+ session_path.write_text("")
+ if self._uploader is not None:
+ self._uploader.upload_file_async(session_path)
+
+ def set_web_password(self, password: str) -> None:
+ """Set the web password for the current BEC account.
+
+ Sends the plaintext password to set_password.php on the server
+ (IP-restricted, HTTPS). PHP generates the bcrypt hash and writes
+ session.htpasswd directly — no Python bcrypt package needed.
+
+ The username is taken from bec.active_account (e.g. 'p23092').
+
+ Example::
+ flomni.webpage_gen.set_web_password("my_secret_password")
+ """
+ if self._set_password_url is None:
+ print("set_web_password: no upload URL configured — cannot reach server")
+ return
+
+ try:
+ account = self._bec.active_account
+ except Exception as exc:
+ print(f"set_web_password: cannot determine active account: {exc}")
+ return
+
+ try:
+ import requests as _requests
+ import urllib3
+ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+ r = _requests.post(
+ self._set_password_url,
+ data={"username": account, "password": password},
+ timeout=15,
+ verify=False,
+ )
+ if r.status_code == 200:
+ resp = r.json()
+ if resp.get("ok"):
+ print(f"set_web_password: password set for '{account}' on server")
+ else:
+ print(f"set_web_password: server error: {resp.get('error', '?')}")
+ else:
+ print(f"set_web_password: HTTP {r.status_code}: {r.text[:120]}")
+ except Exception as exc:
+ print(f"set_web_password: request failed: {exc}")
+
+ # ------------------------------------------------------------------
+ # Singleton lock
+ # ------------------------------------------------------------------
+
+ def _acquire_lock(self) -> bool:
+ """Try to acquire the singleton lock.
+
+ Returns True immediately if the lock is free or stale.
+ Returns False and launches a background watcher thread if the lock is
+ held by another session — the watcher will call _launch() automatically
+ once the lock expires (no user action needed).
+ """
+ _POLL = 3 # seconds between retries in the watcher thread
+
+ lock = self._read_lock()
+ owner = lock.get("owner_id", "unknown") if lock else "none"
+ age = _heartbeat_age_s(lock.get("heartbeat")) if lock else float("inf")
+
+ if age >= _LOCK_STALE_AFTER_S:
+ if lock:
+ self._log(VERBOSITY_NORMAL,
+ f"Stale lock (owner: '{owner}', {age:.0f}s ago). Taking over.")
+ self._write_lock()
+ return True
+
+ # Lock is fresh — spin up a background watcher and return immediately.
+ self._log(VERBOSITY_NORMAL,
+ f"WebpageGenerator: lock held by '{owner}' ({age:.0f}s ago). "
+ f"Will take over automatically when it expires.")
+
+ def _watcher():
+ while True:
+ time.sleep(_POLL)
+ lk = self._read_lock()
+ age = _heartbeat_age_s(lk.get("heartbeat")) if lk else float("inf")
+ self._log(VERBOSITY_VERBOSE,
+ f" …waiting for lock, age {age:.0f}s / {_LOCK_STALE_AFTER_S}s")
+ if age >= _LOCK_STALE_AFTER_S:
+ break
+ owner2 = lk.get("owner_id", "unknown") if lk else "none"
+ self._log(VERBOSITY_VERBOSE,
+ f"Lock expired (was '{owner2}'). Taking over.")
+ self._write_lock()
+ self._launch()
+
+ t = threading.Thread(target=_watcher, name="WebpageGeneratorWatcher", daemon=True)
+ t.start()
+ return False
+
+ def _write_lock(self) -> None:
+ self._bec.set_global_var(_LOCK_VAR_KEY, {
+ "owner_id": self._owner_id,
+ "heartbeat": _now_iso(),
+ "pid": os.getpid(),
+ "hostname": socket.gethostname(),
+ })
+
+ def _read_lock(self) -> dict:
+ val = self._bec.get_global_var(_LOCK_VAR_KEY)
+ return val if isinstance(val, dict) else {}
+
+ def _release_lock(self) -> None:
+ lock = self._read_lock()
+ if lock.get("owner_id") == self._owner_id:
+ self._bec.delete_global_var(_LOCK_VAR_KEY)
+
+ # ------------------------------------------------------------------
+ # Main loop
+ # ------------------------------------------------------------------
+
+ def _run(self) -> None:
+ while not self._stop_event.is_set():
+ t0 = _epoch()
+ try:
+ self._cycle()
+ except Exception as exc:
+ self._log(VERBOSITY_NORMAL,
+ f"WebpageGenerator cycle error: {exc}", level="warning")
+ if not self._local_only:
+ # The singleton lock coordinates real sessions for this
+ # account; a mismatched-account fallback session must never
+ # claim it (see _launch()'s local_only docstring).
+ try:
+ self._write_lock()
+ except Exception:
+ pass
+ self._stop_event.wait(max(0.0, self._cycle_interval - (_epoch() - t0)))
+
+ def _cycle(self) -> None:
+ """One generator cycle: read state -> derive status -> write files -> upload."""
+
+ # ── Progress ─────────────────────────────────────────────────
+ progress = self._bec.get_global_var("tomo_progress") or {}
+
+ # ── Queue ────────────────────────────────────────────────────
+ # queue_status reflects BEC's ScanQueueStatus: 'RUNNING', 'PAUSED',
+ # or 'LOCKED' (the latter set whenever any lock — e.g. from
+ # ScanInterlockActor — is applied to the queue; the prior status is
+ # restored once the lock is released).
+ # A scan is actually executing only when info is non-empty AND
+ # active_request_block is set on the first entry.
+ try:
+ qi = self._bec.queue.queue_storage.current_scan_queue
+ primary = qi.get("primary")
+ queue_status = primary.status if primary is not None else "unknown"
+ queue_has_active_scan = (
+ primary is not None
+ and len(primary.info) > 0
+ and primary.info[0].active_request_block is not None
+ )
+ queue_locks = _read_queue_locks(primary)
+ except Exception:
+ queue_status = "unknown"
+ queue_has_active_scan = False
+ queue_locks = []
+
+ # ── Beamline states ──────────────────────────────────────────
+ # Diagnostic display AND (via beamline_blocking below) one of the two
+ # signals that drive the 'blocked' experiment_status; the other is
+ # queue_locks. A watched state that is out of its accepted range
+ # (mismatched) means the scan interlock is / would be holding the
+ # queue, even if the running scan has already been interrupted and
+ # active_request_block cleared.
+ try:
+ beamline_states_info = _read_beamline_states_info(self._bec)
+ except Exception as exc:
+ self._log(VERBOSITY_VERBOSE,
+ f"beamline_states read error: {exc}", level="warning")
+ beamline_states_info = {"enabled": None, "states": []}
+
+ # True if the scan interlock is enabled AND at least one watched state
+ # is out of its accepted range. Gated on `enabled`: a mismatched state
+ # while the interlock is disabled does not actually block a scan, so it
+ # must not raise the 'blocked' status. Kept in lock-step with the
+ # beamline-states card summary (renderBeamlineStates), which likewise
+ # only calls mismatched states "blocking" when the interlock is enabled,
+ # so the top status pill and that card can never disagree.
+ beamline_blocking = bool(
+ beamline_states_info.get("enabled") is True
+ and any(s.get("mismatched") for s in beamline_states_info.get("states", []))
+ )
+
+ # Current scan number (the BEC scan in progress right now, e.g. for
+ # comparison against ptycho reconstruction filenames like S06770).
+ # Only meaningful while a scan is actually active; None otherwise.
+ current_scan_number = None
+ if queue_has_active_scan:
+ try:
+ current_scan_number = primary.info[0].active_request_block.scan_number
+ except Exception:
+ current_scan_number = None
+
+ # ── Idle tracking ────────────────────────────────────────────
+ hb_age = _heartbeat_age_s(progress.get("heartbeat"))
+ tomo_active = hb_age < _TOMO_HEARTBEAT_STALE_S
+
+ # Detect scans that start and finish entirely between two polls
+ # by watching for a new queue_id at the top of history.
+ try:
+ history = self._bec.queue.queue_storage.queue_history
+ latest_queue_id = history[0].info.queue_id if history else None
+ except Exception:
+ latest_queue_id = None
+
+ history_changed = (
+ latest_queue_id is not None
+ and latest_queue_id != self._last_queue_id
+ )
+ self._last_queue_id = latest_queue_id
+
+ if (tomo_active or queue_has_active_scan or history_changed
+ or beamline_blocking):
+ self._last_active_time = _epoch()
+ self._had_activity = True
+
+ exp_status = _derive_status(
+ progress, queue_has_active_scan,
+ self._last_active_time, self._had_activity,
+ queue_locks, beamline_blocking,
+ )
+ idle_for_s = (
+ None if self._last_active_time is None
+ else max(0.0, _epoch() - self._last_active_time)
+ )
+
+ # ── Tomo queue (setup-specific; see HAS_TOMO_QUEUE) ─────────────
+ # Persisted list of parameter-snapshot jobs (global var "tomo_queue").
+ # Each job: {"label": str, "params": {...},
+ # "status": pending|running|incomplete|done, "added_at": ISO str}
+ # Instruments without this feature (e.g. LamNI) never touch the
+ # global var and always report an empty queue.
+ if self.HAS_TOMO_QUEUE:
+ tomo_queue = self._bec.get_global_var("tomo_queue") or []
+ else:
+ tomo_queue = []
+
+ # ── Reconstruction queue ──────────────────────────────────────
+ recon = self._collect_recon_data()
+
+ # ── Ptychography images ───────────────────────────────────────
+ # Snapshot scan id BEFORE collecting so we can detect changes
+ prev_ptycho_scan = self._last_ptycho_scan
+
+ ptycho = {}
+ try:
+ ptycho = self._collect_ptycho_images()
+ except Exception as exc:
+ self._log(VERBOSITY_VERBOSE, f"ptycho image error: {exc}", level="warning")
+
+ # ── Setup-specific data (subclass hook) ───────────────────────
+ setup = {}
+ try:
+ setup = self._collect_setup_data()
+ except Exception as exc:
+ self._log(VERBOSITY_VERBOSE, f"setup data error: {exc}", level="warning")
+
+ # ── Payload ───────────────────────────────────────────────────
+ payload = {
+ "generated_at": _now_iso(),
+ "generated_at_epoch": _epoch(),
+ "experiment_status": exp_status,
+ "queue_status": queue_status,
+ "queue_has_active_scan": queue_has_active_scan,
+ "idle_for_s": idle_for_s,
+ "idle_for_human": _format_duration(idle_for_s),
+ "queue_locks": queue_locks,
+ "beamline_states": beamline_states_info,
+ "tomo_queue": tomo_queue,
+ "progress": {
+ "tomo_type": progress.get("tomo_type", "N/A"),
+ "projection": progress.get("projection", 0),
+ "total_projections": progress.get("total_projections", 0),
+ "subtomo": progress.get("subtomo", 0),
+ "subtomo_projection": progress.get("subtomo_projection", 0),
+ "subtomo_total_projections": progress.get("subtomo_total_projections", 1),
+ "angle": progress.get("angle", 0),
+ "tomo_start_time": progress.get("tomo_start_time"),
+ "tomo_start_scan_number": progress.get("tomo_start_scan_number"),
+ "current_scan_number": current_scan_number,
+ "estimated_remaining_s": progress.get("estimated_remaining_time"),
+ "estimated_remaining_human": _format_duration(
+ progress.get("estimated_remaining_time")
+ ),
+ "estimated_finish_time": progress.get("estimated_finish_time"),
+ "tomo_heartbeat": progress.get("heartbeat"),
+ "tomo_heartbeat_age_s": round(hb_age, 1) if hb_age != float("inf") else None,
+ },
+ "recon": recon,
+ "ptycho": ptycho,
+ "setup": setup,
+ "generator": {
+ "owner_id": self._owner_id,
+ "cycle_interval_s": self._cycle_interval,
+ "active_account": getattr(self._bec, "active_account", None) or "",
+ },
+ }
+
+ # ── Write status.json ─────────────────────────────────────────
+ # (HTML is static, written once at _launch())
+ (self._output_dir / "status.json").write_text(
+ json.dumps(payload, indent=2, default=str)
+ )
+
+ # ── Upload ────────────────────────────────────────────────────
+ # Uploads run in a background daemon thread — never blocks this cycle.
+ # If the server is unreachable the cycle continues normally; failures
+ # are logged as warnings. If an upload is still in progress from the
+ # previous cycle the new request is dropped (logged at DEBUG level).
+ # Never upload from the account-mismatch fallback (local_only) — the
+ # active account isn't the one the remote page's auth is set up for.
+ if self._uploader is not None and not self._local_only:
+ new_scan = ptycho.get("scan_id")
+ if new_scan and new_scan != prev_ptycho_scan:
+ # Scan changed: ask server to delete old S*_*.png/jpg first,
+ # then upload all current files (including new ptycho images).
+ self._log(VERBOSITY_VERBOSE,
+ f"Ptycho scan changed {prev_ptycho_scan} -> {new_scan}, "
+ f"triggering remote cleanup + full upload")
+ self._uploader.cleanup_ptycho_images_async()
+ # Small delay so cleanup runs before the file uploads start.
+ # Both are async, so we schedule the full upload on a short timer.
+ def _delayed_full_upload():
+ time.sleep(2)
+ self._uploader.upload_dir_async(self._output_dir)
+ threading.Thread(target=_delayed_full_upload,
+ name="HttpUploaderDelayed", daemon=True).start()
+ else:
+ # Normal cycle: only upload files that have changed since last cycle.
+ self._uploader.upload_changed_async(self._output_dir)
+
+ # ── Console feedback ──────────────────────────────────────────
+ blocked_summary = (
+ f" locked_by={','.join(l['identifier'] for l in queue_locks)}"
+ if queue_locks else ""
+ )
+ self._log(VERBOSITY_VERBOSE,
+ f"[{_now_iso()}] {exp_status:<12} active={queue_has_active_scan} "
+ f"proj={payload['progress']['projection']}/"
+ f"{payload['progress']['total_projections']} "
+ f"hb={payload['progress']['tomo_heartbeat_age_s']}s "
+ f"idle={_format_duration(idle_for_s)}" + blocked_summary)
+ self._log(VERBOSITY_DEBUG,
+ f" payload:\n{json.dumps(payload, indent=4, default=str)}")
+
+ # ------------------------------------------------------------------
+ # Reconstruction queue
+ # ------------------------------------------------------------------
+
+ def _collect_recon_data(self) -> dict:
+ foldername = (
+ self._bec.get_global_var("ptycho_reconstruct_foldername") or "ptycho_reconstruct"
+ )
+ base = Path("~/data/raw/logs/reconstruction_queue").expanduser()
+ queue_dir = base / foldername
+ failed_dir = queue_dir / "failed"
+
+ waiting = failed = 0
+ try:
+ if queue_dir.exists():
+ waiting = len(list(queue_dir.glob("*.dat")))
+ except Exception:
+ pass
+ try:
+ if failed_dir.exists():
+ failed = len(list(failed_dir.glob("*.dat")))
+ except Exception:
+ pass
+
+ return {
+ "foldername": foldername,
+ "folder_path": str(queue_dir),
+ "waiting": waiting,
+ "failed": failed,
+ }
+
+ # ------------------------------------------------------------------
+ # Ptychography images
+ # ------------------------------------------------------------------
+
+ def _collect_ptycho_images(self) -> dict:
+ """Find the latest ptycho reconstruction, copy/thumbnail changed images.
+
+ Scans _PTYCHO_ONLINE_DIR for the most-recently-modified subfolder whose
+ name contains an underscore (e.g. dset_00005). Within that folder looks
+ for PNG files matching S*_phase.png, S*_probe.png, S*_amplitude.png,
+ S*_err.png, S*_spectrum.png.
+
+ When the scan changes (new S##### prefix detected):
+ - Thumbnails are generated with Pillow and saved to output_dir
+ - Full-size PNGs are also copied to output_dir (for click-to-full)
+ - Old ptycho files from previous scans are removed from output_dir
+
+ Returns a dict consumed by the JS ptycho renderer:
+ scan_id str e.g. "S06666"
+ images list [{"role": "phase", "thumb": "..._thumb.jpg",
+ "full": "..._phase.png"}, ...]
+ folder str source folder path (for diagnostics)
+ """
+ base = Path(_PTYCHO_ONLINE_DIR).expanduser()
+ if not base.exists():
+ return {}
+
+ # Find latest subfolder by modification time.
+ candidates = sorted(
+ [d for d in base.iterdir() if d.is_dir()],
+ key=lambda d: d.stat().st_mtime,
+ )
+ if not candidates:
+ return {}
+ latest_dir = candidates[-1]
+
+ # Roles we care about — phase and probe are primary (always shown),
+ # amplitude/err/spectrum are secondary (shown on expand)
+ _ROLES = [
+ ("phase", "S*phase.png", True),
+ ("probe", "S*probe.png", True),
+ ("amplitude", "S*amplitude.png", False),
+ ("error", "S*err.png", False),
+ ("spectrum", "S*spectrum.png", False),
+ ]
+
+ # Collect available source files
+ found = {}
+ scan_id = None
+ for role, pattern, _primary in _ROLES:
+ matches = sorted(latest_dir.glob(pattern))
+ if matches:
+ src = matches[-1]
+ stem = src.stem # e.g. S06666_400x400_b0_run_1_phase
+ sid = stem.split("_")[0]
+ if scan_id is None:
+ scan_id = sid
+ found[role] = src
+
+ if not scan_id:
+ return {}
+
+ # If scan unchanged, thumbs exist, and source hasn't changed, skip all work
+ thumb_key = f"{scan_id}_phase_thumb.jpg"
+ thumb_path = self._output_dir / thumb_key
+ src_phase = found.get("phase")
+ thumb_fresh = (
+ thumb_path.exists()
+ and src_phase is not None
+ and thumb_path.stat().st_mtime >= src_phase.stat().st_mtime
+ )
+ if (scan_id == self._last_ptycho_scan and thumb_fresh):
+ # Rebuild result dict from existing files without doing I/O
+ images = []
+ for role, _pat, primary in _ROLES:
+ thumb = self._output_dir / f"{scan_id}_{role}_thumb.jpg"
+ full = self._output_dir / f"{scan_id}_{role}.png"
+ if thumb.exists() and full.exists():
+ images.append({
+ "role": role,
+ "thumb": thumb.name,
+ "full": full.name,
+ "primary": primary,
+ })
+ return {"scan_id": scan_id, "images": images, "folder": str(latest_dir)}
+
+ # Scan changed (or first run) — clean up old ptycho files in output_dir
+ self._log(VERBOSITY_VERBOSE,
+ f"New ptycho scan detected: {scan_id} in {latest_dir.name}")
+ for old_file in self._output_dir.glob("S*_*_thumb.jpg"):
+ if not old_file.name.startswith(scan_id + "_"):
+ old_file.unlink(missing_ok=True)
+ for old_file in self._output_dir.glob("S*_*.png"):
+ if not old_file.name.startswith(scan_id + "_"):
+ for role, _, _ in _ROLES:
+ if f"_{role}." in old_file.name or f"_{role}_" in old_file.name:
+ old_file.unlink(missing_ok=True)
+ break
+
+ # Generate thumbnails and copy full-size for each found role
+ images = []
+ for role, _pat, primary in _ROLES:
+ src = found.get(role)
+ if src is None:
+ continue
+ thumb_name = f"{scan_id}_{role}_thumb.jpg"
+ full_name = f"{scan_id}_{role}.png"
+ thumb_path = self._output_dir / thumb_name
+ full_path = self._output_dir / full_name
+ try:
+ if _PIL_AVAILABLE:
+ img = _PILImage.open(src)
+ img.thumbnail((_PTYCHO_THUMB_PX, _PTYCHO_THUMB_PX))
+ img.save(thumb_path, "JPEG", quality=85)
+ else:
+ shutil.copy2(src, thumb_path)
+ shutil.copy2(src, full_path)
+ images.append({
+ "role": role,
+ "thumb": thumb_name,
+ "full": full_name,
+ "primary": primary,
+ })
+ self._log(VERBOSITY_VERBOSE, f" ptycho {role}: thumb={thumb_name}")
+ except Exception as exc:
+ self._log(VERBOSITY_VERBOSE,
+ f" ptycho {role} failed: {exc}", level="warning")
+
+ self._last_ptycho_scan = scan_id
+ return {"scan_id": scan_id, "images": images, "folder": str(latest_dir)}
+
+ # ------------------------------------------------------------------
+ # Subclass hooks
+ # ------------------------------------------------------------------
+
+ def _collect_setup_data(self) -> dict:
+ """
+ Override in subclasses to return a dict of setup-specific data.
+ The dict is embedded in the JSON payload as 'setup' and rendered
+ by the HTML page as the 'Instrument details' section.
+
+ Expected keys (all optional):
+ type (str) -- setup identifier, e.g. "flomni"
+ sample_name (str) -- current sample name
+ temperatures (dict) -- label -> float (degrees C) or None
+ current_params (dict) -- raw tomo parameters read live (keyed by
+ the same global-var names the tomo queue
+ stores in job.params, e.g. tomo_countingtime,
+ fovx, tomo_angle_stepsize, ...). Rendered by
+ the page through the same param-table path
+ as a queue job (buildParamRows), so the
+ 'Current measurement' section always matches
+ the queue's level of detail -- and shows the
+ details even for a scan started outside the
+ queue. Values are left raw; the page formats
+ them (fmtTqParam) and computes the projection
+ count (calcProjections).
+ """
+ return {}
+
+ # ------------------------------------------------------------------
+ # Logging
+ # ------------------------------------------------------------------
+
+ def _log(self, min_verbosity: int, msg: str, level: str = "info") -> None:
+ if self._verbosity < min_verbosity:
+ return
+ if level == "warning":
+ logger.warning(msg)
+ elif level == "error":
+ logger.error(msg)
+ else:
+ print(msg)
+
+
+
+# ---------------------------------------------------------------------------
+# Factory
+# ---------------------------------------------------------------------------
+
+def make_webpage_generator(bec_client, **kwargs):
+ """
+ Select the appropriate WebpageGenerator subclass based on the BEC session
+ name (bec._ip.prompts.session_name) and return a ready-to-start instance.
+
+ Usage (generic startup script):
+ from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.webpage_generator_base import (
+ make_webpage_generator,
+ )
+ gen = make_webpage_generator(bec, output_dir="~/data/raw/webpage/",
+ upload_url="http://omny.online/upload.php")
+ gen.start()
+ """
+ try:
+ session = bec_client._ip.prompts.session_name
+ except Exception:
+ session = "unknown"
+
+ if session == "flomni":
+ from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import (
+ FlomniWebpageGenerator,
+ )
+ return FlomniWebpageGenerator(bec_client, **kwargs)
+
+ elif session == "lamni":
+ from csaxs_bec.bec_ipython_client.plugins.LamNI.LamNI_webpage_generator import (
+ LamniWebpageGenerator,
+ )
+ return LamniWebpageGenerator(bec_client, **kwargs)
+
+ elif session == "omny":
+ # omny has no real webpage-generator subclass yet (its
+ # omny_webpage_generator.py is an unwired, stale duplicate) --
+ # fall through to the base class until that's built.
+ return WebpageGeneratorBase(bec_client, **kwargs)
+
+ else:
+ return WebpageGeneratorBase(bec_client, **kwargs)
+
+
+# ---------------------------------------------------------------------------
+# HTML template (written once at start(), not every cycle)
+# Three themes: dark (default), light, auto (follows OS preference).
+# Theme is stored in localStorage and applied via data-theme on .
+# ---------------------------------------------------------------------------
+
+def _render_html(
+ phone_numbers: list,
+ has_tomo_queue: bool = True,
+ tomo_types: dict = None,
+ tq_param_display: list = None,
+ logo_filename: str | None = None,
+) -> str:
+ # Falls back to a filename that will 404 (triggering the
+ # text-logo fallback already in the template) when a setup has no logo --
+ # e.g. WebpageGeneratorBase's own default _logo_path() returning None.
+ logo_src = logo_filename or "logo.png"
+
+ phones_html = "\n".join(
+ f'
'
+ f'{label}'
+ f'{num}
'
+ for label, num in phone_numbers
+ )
+
+ # Tomo queue card + its slot in the default card order are only emitted
+ # for setups that actually have the feature (see HAS_TOMO_QUEUE).
+ if has_tomo_queue:
+ tomo_queue_card_html = """
+
+
+
⋮⋮
+
Tomo queue
+
+
+"""
+ else:
+ tomo_queue_card_html = ""
+
+ default_order = ['audio', 'recon-queue', 'ptycho', 'instrument', 'blstates', 'contacts']
+ if has_tomo_queue:
+ default_order.insert(default_order.index('contacts'), 'tomo-queue')
+ default_order_json = json.dumps(default_order)
+
+ # Declarative tomo-type -> projection-count formula, consumed by the
+ # generic calcProjections() in JS instead of a hardcoded per-instrument
+ # branch. See WebpageGeneratorBase.TOMO_TYPES for the schema.
+ tomo_types_json = json.dumps(tomo_types or {})
+
+ # Declarative param-key -> display-label table for the tomo-queue job
+ # detail table and the "Current measurement" section, consumed by
+ # buildParamRows() in JS instead of a hardcoded per-instrument list.
+ # See WebpageGeneratorBase.TQ_PARAM_DISPLAY for the schema.
+ tq_param_display_json = json.dumps(tq_param_display or [])
+
+ return f"""
+
+
+
+
+ Experiment Status
+
+
+
+
+
+
+ ⚠ Status page data is outdated — generator may have stopped
+
+
+
+
+
+ STATUS
+ · STATUS
+
+
+
+
+
+ Theme
+
+
+
+
+
updating…
+
+
+
+
+
+
-
+
Loading…
+
+
+
+
Tomography progress
+
+
+
+ 0%
+ OVERALL
+
+
+
+
Projection-
+
Sub-tomo-
+
Angle-
+
Tomo type-
+
Remaining-
+
Finish time-
+
Started-
+
Queue job-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
⋮⋮
+
+
+
+
+ System
+
+
+
+ Watch
+
+
+
+ Blocked
+
+
+
+ Live
+
+
+ Audio disabled
+
+
+
+
+
+
+
+
+
+
+
+
⋮⋮
+
Reconstruction queue
+
+
Waiting-
+
Failed-
+
Queue name-
+
+
+
+
+
+
⋮⋮
+
Ptychography reconstructions
+
+
No reconstruction found
+
+
+
+
+
+
⋮⋮
+
Instrument details
+
+
+
+
+
+
⋮⋮
+
Beamline states
+
+
+
+
State
Status
Watched
+
+
+
+
+
+{tomo_queue_card_html}
+
+
+
⋮⋮
+
Contacts
+
+{phones_html}
+
+
+
+
+
+
+
+
+
+
+
+"""
\ No newline at end of file
diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/FLOMNI_TO_LAMNI_COMPARISON.md b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/FLOMNI_TO_LAMNI_COMPARISON.md
index 4f3aaf5..92d3a1f 100644
--- a/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/FLOMNI_TO_LAMNI_COMPARISON.md
+++ b/csaxs_bec/bec_ipython_client/plugins/flomni/AI_docs/FLOMNI_TO_LAMNI_COMPARISON.md
@@ -5,10 +5,10 @@ which flomni features (GUI, status webpage, tomo-parameter widget, angle
distribution) are worth porting to lamni. This is a comparison only — no
migration had been implemented yet as of when it was originally written.
-**Status update (2026-07-16): items 2–4 of §5's recommended order are now
-done.** Only **item 1, the webpage generator (§4), remains open** — the rest
-of this document (originally a forward-looking comparison/plan) is now a
-historical snapshot of the *starting* gap, not the current state. See:
+**Status update (2026-07-16): all of §5's recommended order (items 1–4) is
+now done**, including the webpage generator (§4) — the rest of this document
+(originally a forward-looking comparison/plan) is now a historical snapshot
+of the *starting* gap, not the current state. See:
- `csaxs_bec/bec_ipython_client/plugins/LamNI/AI_docs/TOMO_QUEUE_PORT.md` —
tomo-queue/job-list backend ported to lamni via a shared `TomoQueueMixin`
@@ -19,11 +19,27 @@ historical snapshot of the *starting* gap, not the current state. See:
button), plus several real `lamni.py` bugs found and fixed along the way
(two `sub_tomo_scan()` angle bugs, three `tomo_scan()`/`write_pdf_report()`
crashes, progress-GUI-auto-show + stale-busy-heartbeat).
-- **Remaining**: `LamNI_webpage_generator.py` is still the stub described in
- §4 below — `_collect_setup_data()` commented out, `_TEMP_MAP` empty,
- `HAS_TOMO_QUEUE = False` (now incorrect — a queue backend exists, see
- above), `TOMO_TYPES` placeholders, and never wired into
- `lamni.py.__init__`. §4's 5-step plan is otherwise still accurate.
+- **Webpage generator (§4) — done.** The shared base
+ (`WebpageGeneratorBase`, `HttpUploader`, `LocalHttpServer`,
+ `make_webpage_generator()`, the HTML/JS template) was extracted out of
+ `flomni/flomni_webpage_generator.py` into
+ `OMNY_shared/webpage_generator_base.py`, mirroring the tomo-queue-mixin
+ precedent — `flomni_webpage_generator.py` now only holds
+ `FlomniWebpageGenerator`. `LamNI_webpage_generator.py`'s broken import
+ (`flomni.webpage_generator`, a module that never existed) is fixed to
+ import from the shared module. `TOMO_TYPES` confirmed identical to
+ flomni's 3-mode dict by reading `lamni.py`'s actual `tomo_scan()`.
+ `HAS_TOMO_QUEUE` is now `True` (the tomo-queue mixin backs it). The HTML/JS
+ template's tomo-queue param-display table (`TQ_PARAM_DISPLAY`), previously
+ hardcoded to flomni's param names, is now a per-subclass override the same
+ way `TOMO_TYPES` already was, so lamni's page shows its own param names
+ (`tomo_circfov`, `lamni_stitch_x/y`, ...). Sample name comes from lamni's
+ `sample_name` global var (no automatic sample changer exists on lamni, so
+ no device-backed lookup); temperatures stay empty (no lamni temp/humidity
+ device configured yet). Wired into `LamNI.__init__` the same way
+ `FlomniWebpageGenerator` is wired into `Flomni.__init__`. omny was left
+ untouched (`omny_webpage_generator.py` remains dead/unwired duplicate
+ code — separate future task if omny ever needs a real one).
- Also still explicitly deferred (confirmed with Mirko, separate future
tasks): `zero_deg_reference_at_each_subtomo` for lamni (not implemented at
all yet), and sharing angle-calculation code between flomni/lamni
diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
index 796cf23..beae086 100644
--- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
+++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni.py
@@ -32,14 +32,6 @@ from csaxs_bec.bec_ipython_client.plugins.omny.omny_general_tools import (
from csaxs_bec.devices.omny.galil.galil_ophyd import GalilError
from csaxs_bec.devices.omny.sample_desc_codec import pack_desc, unpack_desc
-# from csaxs_bec.bec_ipython_client.plugins.flomni.webpage_generator import (
-# FlomniWebpageGenerator,
-# VERBOSITY_SILENT, # 0 — no output
-# VERBOSITY_NORMAL, # 1 — startup / stop messages only (default)
-# VERBOSITY_VERBOSE, # 2 — one-line summary per cycle
-# VERBOSITY_DEBUG, # 3 — full JSON payload per cycle
-# )
-
logger = bec_logger.logger
if builtins.__dict__.get("bec") is not None:
diff --git a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py
index d39ebe5..0fe6885 100644
--- a/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py
+++ b/csaxs_bec/bec_ipython_client/plugins/flomni/flomni_webpage_generator.py
@@ -1,34 +1,19 @@
"""
-flomni/webpage_generator.py
-============================
-Background thread that reads tomo progress from the BEC global variable store
-and writes status.json (every cycle) + status.html (once at startup) to a
-staging directory. An optional HttpUploader sends those files to a web host
-after every cycle, running in a separate daemon thread so uploads never block
-the generator cycle. A built-in LocalHttpServer always serves the output
-directory locally (default port 8080) so the page can be accessed on the
-lab network without any extra setup.
+flomni/flomni_webpage_generator.py
+====================================
+flOMNI-specific webpage generator subclass.
-Architecture
-------------
-LocalHttpServer -- built-in HTTP server; serves output_dir on port 8080.
- Always started at _launch(); URL printed to console.
-HttpUploader -- non-blocking HTTP uploader (fire-and-forget thread).
- Tracks file mtimes; only uploads changed files.
- Sends a cleanup request to the server when the
- ptycho scan ID changes (removes old S*_*.png/jpg).
-WebpageGeneratorBase -- all common logic: queue, progress, idle detection,
- reconstruction queue, HTML, audio, phone numbers,
- outdated-page warning. Import this in subclasses.
-FlomniWebpageGenerator -- flomni-specific: temperatures, sample name, settings,
- flOMNI logo.
-make_webpage_generator() -- factory; selects the right class by session name.
- Lazy-imports LamNI / omny subclasses to avoid
- circular dependencies.
+The shared implementation (WebpageGeneratorBase, HttpUploader,
+LocalHttpServer, make_webpage_generator(), the HTML/JS template) now lives in
+csaxs_bec.bec_ipython_client.plugins.OMNY_shared.webpage_generator_base --
+extracted so it can be shared with lamni (and, later, omny), the same way the
+tomo-queue backend was shared via OMNY_shared/tomo_queue_mixin.py. This file
+keeps only the flomni-specific subclass: temperatures, sample name, and live
+current-measurement parameters from global vars.
Integration (inside Flomni.__init__, after self._progress_proxy.reset()):
--------------------------------------------------------------------------
- from csaxs_bec.bec_ipython_client.plugins.flomni.webpage_generator import (
+ from csaxs_bec.bec_ipython_client.plugins.flomni.flomni_webpage_generator import (
FlomniWebpageGenerator,
)
self._webpage_gen = FlomniWebpageGenerator(
@@ -48,1422 +33,17 @@ Interactive helpers (optional, in the iPython session):
flomni._webpage_gen.verbosity = 3 # DEBUG: full JSON per cycle
flomni._webpage_gen.stop() # release lock, stop local server
flomni._webpage_gen.start() # restart after stop()
- flomni.webpage_gen.set_web_password("pw") # set web password for current e-account
-
-Session authentication:
------------------------
-Two htpasswd files are used on the remote server:
- users.htpasswd -- static, managed manually with htpasswd -B
- session.htpasswd -- managed by the generator; one entry for the current e-account
-
-At _launch(), the generator queries session_query.php to find which account was
-previously active. If the account has changed the session.htpasswd is cleared
-immediately (old user loses access) and uploaded. The new user calls
-set_web_password() to set their own password and gain access.
-
-Account-mismatch fallback (dev/sim sessions):
-----------------------------------------------
-If bec_client.active_account doesn't match the OS login (_check_account_match)
--- typically a leftover session under the wrong experiment account, or a dev/sim
-session run under a personal account -- start() no longer does nothing. It
-starts a *local-only* status page on a separate port (default 8081, see
-mismatch_local_port) for local debugging: no singleton lock (so it can never
-block the real session's takeover), no remote upload, no session.htpasswd
-changes. Look for "[local-only: account mismatch]" in the startup log / status().
+ flomni._webpage_gen.set_web_password("pw") # set web password for current e-account
"""
-import datetime
-import functools
-import http.server
-import json
-import os
-import shutil
-import socket
-import threading
-import time
from pathlib import Path
-try:
- from PIL import Image as _PILImage
- _PIL_AVAILABLE = True
-except ImportError:
- _PIL_AVAILABLE = False
+from csaxs_bec.bec_ipython_client.plugins.OMNY_shared.webpage_generator_base import (
+ WebpageGeneratorBase,
+ _safe_get,
+ _safe_float,
+)
-from bec_lib import bec_logger
-
-logger = bec_logger.logger
-
-# ---------------------------------------------------------------------------
-# Constants
-# ---------------------------------------------------------------------------
-
-_LOCK_VAR_KEY = "webpage_generator_lock"
-_LOCK_STALE_AFTER_S = 35 # just over 2 missed cycles (2×15s+5s margin)
-_CYCLE_INTERVAL_S = 15
-_TOMO_HEARTBEAT_STALE_S = 90
-
-# Fallback port used when the BEC account doesn't match the system user (see
-# _check_account_match / start()). Deliberately different from the default
-# local_port so a local-only fallback session never collides with a
-# legitimately-started one on the same host.
-_MISMATCH_LOCAL_PORT = 8081
-
-# Path where online ptycho reconstructions are written by the analysis pipeline.
-# Subfolders follow the pattern dset_NNNNN and contain S#####_*_phase.png etc.
-_PTYCHO_ONLINE_DIR = "~/data/raw/analysis/online/ptycho"
-_PTYCHO_THUMB_PX = 200 # longest edge of generated thumbnail in pixels
-
-VERBOSITY_SILENT = 0
-VERBOSITY_NORMAL = 1
-VERBOSITY_VERBOSE = 2
-VERBOSITY_DEBUG = 3
-
-_PHONE_NUMBERS = [
- ("Beamline", "+41 56 310 5845"),
- ("Control room", "+41 56 310 5503"),
-]
-
-
-# ---------------------------------------------------------------------------
-# Helpers
-# ---------------------------------------------------------------------------
-
-def _now_iso() -> str:
- return datetime.datetime.now().isoformat(timespec="seconds")
-
-def _epoch() -> float:
- return time.time()
-
-def _heartbeat_age_s(iso_str) -> float:
- if iso_str is None:
- return float("inf")
- try:
- ts = datetime.datetime.fromisoformat(iso_str)
- return (datetime.datetime.now() - ts).total_seconds()
- except Exception:
- return float("inf")
-
-def _format_duration(seconds) -> str:
- if seconds is None:
- return "N/A"
- try:
- seconds = int(float(seconds))
- except (TypeError, ValueError):
- return "N/A"
- h, remainder = divmod(seconds, 3600)
- m, s = divmod(remainder, 60)
- if h > 0:
- return f"{h}h {m:02d}m {s:02d}s"
- if m > 0:
- return f"{m}m {s:02d}s"
- return f"{s}s"
-
-def _check_account_match(bec_client) -> bool:
- try:
- active = bec_client.active_account
- system_user = os.getenv("USER") or os.getlogin()
- return active[1:] == system_user[1:]
- except Exception:
- return True
-
-def _safe_get(device_manager, dotpath: str):
- """Navigate a dotted device path and call .get(), returning None on any error."""
- try:
- obj = device_manager.devices
- for part in dotpath.split("."):
- obj = getattr(obj, part)
- return obj.get(cached=True)
- except Exception:
- return None
-
-def _safe_float(val, ndigits: int = 2):
- try:
- return round(float(val), ndigits)
- except (TypeError, ValueError):
- return None
-
-def _gvar(bec_client, key, fmt=None, suffix=""):
- val = bec_client.get_global_var(key)
- if fmt is None:
- return val
- if val is None:
- return "N/A"
- try:
- return f"{val:{fmt}}{suffix}"
- except Exception:
- return str(val)
-
-def _read_beamline_states_info(bec_client) -> dict:
- """Return a diagnostic snapshot of all configured beamline states, for
- display only (NOT used to derive the 'blocked' experiment status — that
- is computed from the queue's own locks via _read_queue_locks(), which is
- the authoritative source; see its docstring for why).
-
- Returns:
- {
- "enabled": bool | None, # bec.builtin_actors.scan_interlock.enabled
- "states": [
- {
- "name": str,
- "status": str, # 'valid' | 'invalid' | 'warning' | 'unknown'
- "label": str,
- "watched": bool, # True if present in states_watched
- "accepted": list[str] | None, # accepted statuses if watched
- "mismatched": bool, # True if watched AND status not in accepted
- },
- ...
- ],
- }
-
- Beamline states are registered dynamically and can differ from
- experiment to experiment, so names are read from the manager's _states
- dict (the same one show_all() iterates over internally — there is no
- public "list names" method as of this BEC version). states_watched and
- enabled are read through the public scan_interlock HLI properties.
- """
- manager = getattr(bec_client, "beamline_states", None)
- if manager is None:
- return {"enabled": None, "states": []}
-
- try:
- names = list(getattr(manager, "_states", {}).keys())
- except Exception:
- names = []
-
- try:
- interlock = bec_client.builtin_actors.scan_interlock
- enabled = interlock.enabled
- states_watched = interlock.states_watched or {}
- except Exception:
- enabled = None
- states_watched = {}
-
- states = []
- for name in names:
- try:
- state_obj = getattr(manager, name, None)
- if state_obj is None:
- continue
- info = state_obj.get() # {"status": ..., "label": ...}
- status = info.get("status", "unknown")
- label = info.get("label", name)
- except Exception:
- status, label = "unknown", name
-
- accepted = states_watched.get(name)
- watched = accepted is not None
- mismatched = watched and status not in accepted
-
- states.append({
- "name": name,
- "status": status,
- "label": label,
- "watched": watched,
- "accepted": accepted,
- "mismatched": mismatched,
- })
-
- return {"enabled": enabled, "states": states}
-
-
-def _read_queue_locks(primary) -> list:
- """Return a list of {"identifier": str, "reason": str} for every lock
- currently applied to the primary scan queue.
-
- BEC's scan_queue.status becomes "LOCKED" (saving the prior status to
- restore later) whenever any lock is added, e.g. by the ScanInterlockActor
- when a watched beamline state goes out of spec (see scan_interlock.py:
- add_queue_lock(queue="primary", reason=..., lock_id="ScanInterlockActor")).
- Reading the lock list directly is more robust than re-deriving "blocked"
- from beamline states ourselves: it reflects whatever is actually holding
- the queue, regardless of which actor (interlock or otherwise) caused it,
- and survives changes to which beamline states are configured/watched.
- """
- if primary is None:
- return []
- try:
- return [
- {"identifier": lock.identifier, "reason": lock.reason}
- for lock in (primary.locks or [])
- ]
- except Exception:
- return []
-
-
-# ---------------------------------------------------------------------------
-# Status derivation
-# ---------------------------------------------------------------------------
-
-def _derive_status(
- progress: dict,
- queue_has_active_scan: bool,
- last_active_time,
- had_activity: bool,
- queue_locks: list | None = None,
- beamline_blocking: bool = False,
-) -> str:
- """
- Returns one of:
- scanning -- tomo heartbeat fresh (< _TOMO_HEARTBEAT_STALE_S), OR
- heartbeat recently seen AND queue still has active scan
- (handles long individual projections > heartbeat timeout)
- blocked -- the experiment is being held up externally: EITHER a
- watched beamline state is out of its accepted range while
- the scan interlock would act on it (beamline_blocking),
- OR the primary queue has at least one explicit lock
- applied (queue_locks). Reported regardless of whether a
- scan is currently executing: when a block hits, the
- running scan is interrupted and its repeat waits for the
- block to clear, so active_request_block (and hence
- queue_has_active_scan) may already be cleared even though
- the beamline is still blocking. Only 'scanning' (fresh
- heartbeat = data still flowing) takes precedence.
- running -- queue has active scan but heartbeat has never been seen,
- and nothing is blocking
- idle -- not scanning, not blocked, last_active_time known
- unknown -- no activity ever seen since generator started
-
- 'unknown' is ONLY returned before any scan activity has been observed.
- Once activity has been seen the status goes directly:
- scanning/running/blocked -> idle
- never through 'unknown'.
- """
- hb_age = _heartbeat_age_s(progress.get("heartbeat"))
- if hb_age < _TOMO_HEARTBEAT_STALE_S:
- return "scanning"
- # External block takes precedence over the remaining scan/idle states.
- # A blocked scan is interrupted (active_request_block cleared) while its
- # repeat waits for the block to clear, so we must NOT require an active
- # scan here — that was the previous behaviour and it made a blocked-but-
- # interrupted experiment fall through to 'idle'.
- if beamline_blocking or queue_locks:
- return "blocked"
- # Heartbeat stale but queue still active and heartbeat was seen recently
- # enough (within 10× the stale window) — likely a long projection, not idle.
- # Report as 'scanning' so audio system does not trigger a false alarm.
- if queue_has_active_scan and hb_age < _TOMO_HEARTBEAT_STALE_S * 10:
- return "scanning"
- if queue_has_active_scan:
- return "running"
- if last_active_time is not None or had_activity:
- return "idle"
- return "unknown"
-
-
-# ---------------------------------------------------------------------------
-# HTTP Uploader (non-blocking, fire-and-forget)
-# ---------------------------------------------------------------------------
-
-class HttpUploader:
- """
- Uploads files from a local directory to a remote server via HTTP POST.
-
- Uploads run in a daemon thread so they never block the generator cycle.
- If an upload is already in progress when the next cycle fires, the new
- upload request is dropped (logged at DEBUG level) rather than queuing up.
-
- File tracking:
- - Tracks mtime of each file; only re-uploads files that have changed.
- - upload_dir() -- uploads ALL eligible files (called once at start).
- - upload_changed() -- uploads only files whose mtime has changed.
-
- Scan change cleanup:
- - cleanup_ptycho_images() -- sends action=cleanup POST to the server,
- asking it to delete all S*_*.png and S*_*.jpg files.
- Called automatically by the generator when the ptycho scan ID changes.
-
- The server-side upload.php must:
- - Accept POST with multipart file upload (field name 'file').
- - Accept POST with action=cleanup to delete ptycho image files.
- - Enforce IP-based access control (129.129.122.x).
- - Validate filename with regex to block path traversal.
- """
-
- _UPLOAD_SUFFIXES = {".html", ".json", ".jpg", ".png", ".pdf", ".txt"}
-
- def __init__(self, url: str, timeout: float = 20.0):
- self._url = url
- self._timeout = timeout
- self._uploaded: dict[str, float] = {} # abs path -> mtime at last upload
- self._lock = threading.Lock()
- self._busy = False # True while an upload thread is running
- self._warn_at: dict[str, float] = {} # key -> epoch of last warning
-
- _WARN_COOLDOWN_S = 600 # only repeat the same warning once per minute
-
- def _warn(self, key: str, msg: str) -> None:
- """Log a warning at most once per _WARN_COOLDOWN_S for a given key."""
- now = _epoch()
- if now - self._warn_at.get(key, 0) >= self._WARN_COOLDOWN_S:
- self._warn_at[key] = now
- logger.warning(msg)
-
- # ── Public API ──────────────────────────────────────────────────────────
-
- def upload_dir_async(self, directory: Path) -> None:
- """Upload ALL eligible files in directory, in a background thread."""
- files = self._eligible_files(directory)
- self._dispatch(self._upload_files, files, True)
-
- def upload_changed_async(self, directory: Path) -> None:
- """Upload only changed files in directory, in a background thread."""
- files = self._changed_files(directory)
- if files:
- self._dispatch(self._upload_files, files, False)
-
- def upload_file_async(self, path: Path) -> None:
- """Upload a single specific file, bypassing suffix whitelist and mtime check."""
- self._dispatch(self._upload_files, [Path(path)], True)
-
- def cleanup_ptycho_images_async(self) -> None:
- """Ask the server to delete all S*_*.png / S*_*.jpg files (background)."""
- self._dispatch(self._do_cleanup)
-
- # ── Internal ────────────────────────────────────────────────────────────
-
- def _dispatch(self, fn, *args) -> None:
- """Run fn(*args) in a daemon thread. Drops the request if already busy."""
- with self._lock:
- if self._busy:
- logger.debug("HttpUploader: previous upload still running, skipping")
- return
- self._busy = True
-
- def _run():
- try:
- fn(*args)
- finally:
- with self._lock:
- self._busy = False
-
- t = threading.Thread(target=_run, name="HttpUploader", daemon=True)
- t.start()
-
- def _eligible_files(self, directory: Path) -> list:
- result = []
- for path in Path(directory).iterdir():
- if path.is_file() and path.suffix in self._UPLOAD_SUFFIXES:
- result.append(path)
- return result
-
- def _changed_files(self, directory: Path) -> list:
- result = []
- for path in self._eligible_files(directory):
- try:
- mtime = path.stat().st_mtime
- except OSError:
- continue
- if self._uploaded.get(str(path)) != mtime:
- result.append(path)
- return result
-
- def _upload_files(self, files: list, force: bool = False) -> None:
- try:
- import requests as _requests
- import urllib3
- urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
- except ImportError:
- self._warn("no_requests", "HttpUploader: 'requests' library not installed")
- return
-
- for path in files:
- try:
- mtime = path.stat().st_mtime
- except OSError:
- continue
- # Double-check mtime unless forced (upload_dir uses force=True)
- if not force and self._uploaded.get(str(path)) == mtime:
- continue
- try:
- with open(path, "rb") as f:
- r = _requests.post(
- self._url,
- files={"file": (path.name, f)},
- timeout=self._timeout,
- verify=False, # accept self-signed / untrusted certs
- )
- if r.status_code == 200:
- self._uploaded[str(path)] = mtime
- self._warn_at.pop(f"upload_{path.name}", None) # clear on success
- logger.debug(f"HttpUploader: OK {path.name}")
- else:
- self._warn(
- f"upload_{path.name}",
- f"HttpUploader: {path.name} -> HTTP {r.status_code}: "
- f"{r.text[:120]}"
- )
- except Exception as exc:
- self._warn(f"upload_{path.name}", f"HttpUploader: {path.name} failed: {exc}")
-
- def _do_cleanup(self) -> None:
- try:
- import requests as _requests
- import urllib3
- urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
- except ImportError:
- return
- try:
- r = _requests.post(
- self._url,
- data={"action": "cleanup"},
- timeout=self._timeout,
- verify=False, # accept self-signed / untrusted certs
- )
- logger.info(f"HttpUploader cleanup: {r.text[:120]}")
- self._warn_at.pop("cleanup", None) # clear on success
- # Forget mtime records for ptycho files so they get re-uploaded
- with self._lock:
- to_remove = [k for k in self._uploaded if "/S" in k or "\\S" in k]
- for k in to_remove:
- self._uploaded.pop(k, None)
- except Exception as exc:
- self._warn("cleanup", f"HttpUploader cleanup failed: {exc}")
-
-
-# ---------------------------------------------------------------------------
-# Local HTTP server (serves output_dir over http://hostname:port/)
-# ---------------------------------------------------------------------------
-
-class LocalHttpServer:
- """
- Serves the generator's output directory over plain HTTP in a daemon thread.
-
- Uses Python's built-in http.server — no extra dependencies.
- Request logging is suppressed so the BEC console stays clean.
- The server survives stop()/start() cycles: _launch() creates a fresh
- instance each time start() is called.
-
- Usage:
- srv = LocalHttpServer(output_dir, port=8080)
- srv.start()
- print(srv.url) # http://hostname:8080/status.html
- srv.stop()
- """
-
- def __init__(self, directory: Path, port: int = 8080):
- self._directory = Path(directory)
- self._port = port
- self._server = None
- self._thread = None
-
- # ── silence the per-request log lines in the iPython console ──────────
- class _QuietHandler(http.server.SimpleHTTPRequestHandler):
- def log_message(self, *args):
- pass
- # handle_error here does nothing — wrong class
-
- class _QuietHTTPServer(http.server.HTTPServer):
- def handle_error(self, request, client_address):
- pass # suppress BrokenPipeError and all other per-connection noise
-
- def start(self) -> None:
- Handler = functools.partial(
- self._QuietHandler,
- directory=str(self._directory),
- )
- try:
- self._server = self._QuietHTTPServer(("", self._port), Handler)
- except OSError as exc:
- raise RuntimeError(
- f"LocalHttpServer: cannot bind port {self._port}: {exc}"
- ) from exc
- self._thread = threading.Thread(
- target=self._server.serve_forever,
- name="LocalHttpServer",
- daemon=True,
- )
- self._thread.start()
-
- def stop(self) -> None:
- if self._server is not None:
- self._server.shutdown() # blocks until serve_forever() returns
- self._server = None
-
- def is_alive(self) -> bool:
- return self._thread is not None and self._thread.is_alive()
-
- @property
- def port(self) -> int:
- return self._port
-
- @property
- def url(self) -> str:
- """Best-guess URL for printing. Uses the machine's hostname."""
- return f"http://{socket.gethostname()}:{self._port}/status.html"
-
-
-# ---------------------------------------------------------------------------
-# Base generator
-# ---------------------------------------------------------------------------
-
-class WebpageGeneratorBase:
- """
- Common webpage generator. Subclass and override:
- _collect_setup_data() -- return dict of instrument-specific data
- _logo_path() -- return Path to logo PNG, or None for text fallback
-
- Setup capability flags (override on subclass as needed):
- HAS_TOMO_QUEUE -- whether this instrument exposes the persisted
- parameter-snapshot job queue (global var
- "tomo_queue" + the "Tomo queue" UI card).
- NOT to be confused with BEC's own primary scan
- queue (queue_status / queue_locks in _cycle()),
- which every setup has and which stays common.
- TOMO_TYPES -- dict describing this instrument's tomography
- scan types, keyed by the value progress["tomo_type"]
- takes. Used by the UI to compute total-projection
- counts for queued jobs without hardcoding a
- per-instrument formula in JS. Recognised "kind"s:
- "equally_spaced_grid" (params: n_subtomos)
- total = floor(180 / angle_stepsize) * n_subtomos
- (independent of angle_range - see
- Flomni._subtomo_angle_plan()'s docstring for
- why: in 360-degree mode, angle_range only
- splits the n_subtomos into complementary
- low/high halves, it does not change the
- total count)
- "golden_capped"
- total = golden_max_number_of_projections
- """
-
- HAS_TOMO_QUEUE = True
- TOMO_TYPES = {
- 1: {"kind": "equally_spaced_grid", "n_subtomos": 8},
- 2: {"kind": "golden_capped"},
- 3: {"kind": "golden_capped"},
- }
-
- def __init__(
- self,
- bec_client,
- output_dir: str = "~/data/raw/webpage/",
- cycle_interval: float = _CYCLE_INTERVAL_S,
- verbosity: int = VERBOSITY_NORMAL,
- upload_url: str = None,
- local_port: int = 8080,
- mismatch_local_port: int = _MISMATCH_LOCAL_PORT,
- ):
- self._bec = bec_client
- self._output_dir = Path(output_dir).expanduser().resolve()
- self._cycle_interval = cycle_interval
- self._verbosity = verbosity
- self._uploader = HttpUploader(upload_url) if upload_url else None
- self._local_port = local_port
- self._mismatch_local_port = mismatch_local_port
- self._local_server = None # created fresh each _launch()
- self._local_only = False # True while running the account-mismatch fallback
-
- # Derive companion URLs from upload_url
- if upload_url is not None:
- self._session_query_url = upload_url.replace("upload.php", "session_query.php")
- self._set_password_url = upload_url.replace("upload.php", "set_password.php")
- else:
- self._session_query_url = None
- self._set_password_url = None
-
- self._thread = None
- self._stop_event = threading.Event()
- self._last_active_time = None # epoch of last tomo/queue activity
- self._last_ptycho_scan = None # scan id of last copied ptycho images
- self._had_activity = False # True once any activity has been observed
- self._last_queue_id = None # tracks queue history changes between cycles
- self._owner_id = f"{socket.gethostname()}:{os.getpid()}"
-
- # ------------------------------------------------------------------
- # Public API
- # ------------------------------------------------------------------
-
- def start(self) -> None:
- """Start the generator.
-
- If this session already holds the thread, does nothing.
- If another session holds a fresh lock, a background watcher thread
- is launched that polls every 3 s and takes over as soon as the lock
- goes stale — no blocking, no second call to start() needed.
- Progress messages during the wait are printed at verbosity >= 2 only.
- """
- if not _check_account_match(self._bec):
- if self._thread is not None and self._thread.is_alive():
- self._log(VERBOSITY_NORMAL, "WebpageGenerator already running in this session.")
- return
- self._log(VERBOSITY_NORMAL,
- "WebpageGenerator: BEC account does not match system user. "
- f"Starting a local-only status page on port "
- f"{self._mismatch_local_port} instead (no singleton lock, no "
- "remote upload, no session-auth changes) — likely a dev/sim "
- "session running under a different account.", level="warning")
- self._launch(local_only=True)
- return
-
- if self._thread is not None and self._thread.is_alive():
- self._log(VERBOSITY_NORMAL, "WebpageGenerator already running in this session.")
- return
-
- if not self._acquire_lock():
- # Lock is fresh — _acquire_lock() has spawned a background watcher
- # that will call _launch() automatically once the lock expires.
- return
-
- self._launch()
-
- def _launch(self, local_only: bool = False) -> None:
- """Actually start the worker thread and write static files. Called by
- start() on the normal path, or by the lock-watcher thread after takeover.
-
- local_only=True is the account-mismatch fallback (see start()): serves
- the status page on a separate port for local debugging only, and skips
- everything that assumes the active BEC account is the "real" one for
- this host — the singleton lock (a mismatched account shouldn't contend
- for it against the legitimate session) and remote upload / session-auth
- (writing session.htpasswd or uploading under the wrong account could
- step on the real experiment's web page).
- """
- self._local_only = local_only
- self._output_dir.mkdir(parents=True, exist_ok=True)
-
- # Copy logo once at startup — HTML is also written once here,
- # not every cycle, since it is a static shell that only loads status.json.
- self._copy_logo()
- (self._output_dir / "status.html").write_text(
- _render_html(_PHONE_NUMBERS, self.HAS_TOMO_QUEUE, self.TOMO_TYPES)
- )
-
- if not local_only:
- # Check whether the active BEC account matches the session user on
- # the remote server. If not, clear session.htpasswd so the old
- # user loses web access immediately. The new user then calls
- # set_web_password().
- self._check_session_user()
-
- # Start local HTTP server (always on; a fresh instance per _launch()).
- port = self._mismatch_local_port if local_only else self._local_port
- if self._local_server is not None and self._local_server.is_alive():
- self._local_server.stop()
- self._local_server = LocalHttpServer(self._output_dir, port)
- try:
- self._local_server.start()
- local_url_msg = f" local={self._local_server.url}"
- except RuntimeError as exc:
- local_url_msg = f" local=ERROR({exc})"
- self._log(VERBOSITY_NORMAL, str(exc), level="warning")
-
- # Upload static files (html + logo) once at startup
- if not local_only and self._uploader is not None:
- self._uploader.upload_dir_async(self._output_dir)
-
- self._stop_event.clear()
- self._thread = threading.Thread(
- target=self._run, name="WebpageGenerator", daemon=True
- )
- self._thread.start()
- mode_note = " [local-only: account mismatch]" if local_only else ""
- self._log(VERBOSITY_NORMAL,
- f"WebpageGenerator started{mode_note} owner={self._owner_id} "
- f"output={self._output_dir} interval={self._cycle_interval}s"
- + (f" upload={self._uploader._url}"
- if (self._uploader and not local_only) else " upload=disabled")
- + f"\n ➜ Status page:{local_url_msg}")
-
- def stop(self) -> None:
- """Stop the generator thread, local HTTP server, and release the singleton lock."""
- self._stop_event.set()
- if self._thread is not None:
- self._thread.join(timeout=self._cycle_interval + 5)
- # Always clear the reference so start() gets a clean slate,
- # even if the thread did not exit within the join timeout.
- self._thread = None
- if self._local_server is not None:
- self._local_server.stop()
- self._local_server = None
- self._release_lock()
- self._local_only = False
- self._log(VERBOSITY_NORMAL, "WebpageGenerator stopped.")
-
- @property
- def verbosity(self) -> int:
- return self._verbosity
-
- @verbosity.setter
- def verbosity(self, val: int) -> None:
- self._verbosity = val
- self._log(VERBOSITY_NORMAL, f"WebpageGenerator verbosity -> {val}")
-
- def status(self) -> None:
- """Print a human-readable status summary to the console."""
- lock = self._read_lock()
- running = self._thread is not None and self._thread.is_alive()
- local = self._local_server.url if (self._local_server and self._local_server.is_alive()) else "stopped"
- remote_user = self._get_remote_session_user() if self._session_query_url else "n/a"
- try:
- active_account = self._bec.active_account
- except Exception:
- active_account = "unknown"
- print(
- f"WebpageGenerator\n"
- f" This session running : {running}"
- + (" [local-only: account mismatch]" if self._local_only else "") + "\n"
- f" Lock owner : {lock.get('owner_id', 'none')}\n"
- f" Lock heartbeat : {lock.get('heartbeat', 'never')}\n"
- f" Output dir : {self._output_dir}\n"
- f" Local URL : {local}\n"
- f" Cycle interval : {self._cycle_interval}s\n"
- f" Upload URL : {self._uploader._url if self._uploader else 'disabled'}\n"
- f" Session query URL : {self._session_query_url or 'disabled'}\n"
- f" BEC active account : {active_account}\n"
- f" Server session user : {remote_user}\n"
- f" Verbosity : {self._verbosity}\n"
- )
-
- # ------------------------------------------------------------------
- # Logo
- # ------------------------------------------------------------------
-
- def _logo_path(self):
- """
- Return a Path to the logo PNG for this setup, or None for text fallback.
- Override in subclasses.
- """
- return None
-
- def _copy_logo(self) -> None:
- """Copy the setup logo to output_dir/logo.png if available."""
- src = self._logo_path()
- if src is None:
- return
- src = Path(src)
- if not src.exists():
- self._log(VERBOSITY_VERBOSE,
- f"Logo not found at {src}, using text fallback.", level="warning")
- return
- try:
- shutil.copy2(src, self._output_dir / "logo.png")
- self._log(VERBOSITY_VERBOSE, f"Logo copied from {src}")
- except Exception as exc:
- self._log(VERBOSITY_NORMAL,
- f"Failed to copy logo: {exc}", level="warning")
-
- # ------------------------------------------------------------------
- # Session authentication
- # ------------------------------------------------------------------
-
- def _get_remote_session_user(self) -> "str | None":
- """GET session_query.php and return the username in session.htpasswd, or None."""
- if self._session_query_url is None:
- return None
- try:
- import requests as _requests
- import urllib3
- urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
- r = _requests.get(self._session_query_url, timeout=10, verify=False)
- if r.status_code == 200:
- return r.json().get("user") # str or None
- self._log(VERBOSITY_VERBOSE,
- f"session_query: HTTP {r.status_code}", level="warning")
- except Exception as exc:
- self._log(VERBOSITY_VERBOSE,
- f"session_query failed: {exc}", level="warning")
- return None
-
- def _check_session_user(self) -> None:
- """Compare BEC active account with the server session user.
-
- - Match: silent, nothing to do.
- - No remote user: current user has no web auth yet — print reminder.
- - Mismatch: clear session.htpasswd immediately (old user loses access),
- then print reminder that current user has no web auth yet.
- """
- if self._session_query_url is None:
- return # no remote server configured
-
- try:
- current_account = self._bec.active_account # e.g. 'p23092'
- except Exception as exc:
- self._log(VERBOSITY_VERBOSE,
- f"session check: cannot read active_account: {exc}", level="warning")
- return
-
- remote_user = self._get_remote_session_user()
-
- if remote_user == current_account:
- return # all good, stay silent
-
- if remote_user is not None:
- # Different user still has web access — remove it immediately.
- print(f"WebpageGenerator: account changed "
- f"'{remote_user}' -> '{current_account}' "
- f"— removing web access for '{remote_user}'")
- self._clear_session_htpasswd()
-
- # Either no previous user or a different user — either way the
- # current account has no web authentication yet.
- _BOLD_RED = "\033[1;31m"
- _RESET = "\033[0m"
- print(f"{_BOLD_RED}***\nWebpageGenerator: '{current_account}' has no web authentication yet.{_RESET}")
- print(f"{_BOLD_RED} To set a password run: flomni.set_web_password(\"your_password\")\n***{_RESET}")
-
- def _clear_session_htpasswd(self) -> None:
- """Write an empty session.htpasswd locally and upload it."""
- session_path = self._output_dir / "session.htpasswd"
- session_path.write_text("")
- if self._uploader is not None:
- self._uploader.upload_file_async(session_path)
-
- def set_web_password(self, password: str) -> None:
- """Set the web password for the current BEC account.
-
- Sends the plaintext password to set_password.php on the server
- (IP-restricted, HTTPS). PHP generates the bcrypt hash and writes
- session.htpasswd directly — no Python bcrypt package needed.
-
- The username is taken from bec.active_account (e.g. 'p23092').
-
- Example::
- flomni.webpage_gen.set_web_password("my_secret_password")
- """
- if self._set_password_url is None:
- print("set_web_password: no upload URL configured — cannot reach server")
- return
-
- try:
- account = self._bec.active_account
- except Exception as exc:
- print(f"set_web_password: cannot determine active account: {exc}")
- return
-
- try:
- import requests as _requests
- import urllib3
- urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
- r = _requests.post(
- self._set_password_url,
- data={"username": account, "password": password},
- timeout=15,
- verify=False,
- )
- if r.status_code == 200:
- resp = r.json()
- if resp.get("ok"):
- print(f"set_web_password: password set for '{account}' on server")
- else:
- print(f"set_web_password: server error: {resp.get('error', '?')}")
- else:
- print(f"set_web_password: HTTP {r.status_code}: {r.text[:120]}")
- except Exception as exc:
- print(f"set_web_password: request failed: {exc}")
-
- # ------------------------------------------------------------------
- # Singleton lock
- # ------------------------------------------------------------------
-
- def _acquire_lock(self) -> bool:
- """Try to acquire the singleton lock.
-
- Returns True immediately if the lock is free or stale.
- Returns False and launches a background watcher thread if the lock is
- held by another session — the watcher will call _launch() automatically
- once the lock expires (no user action needed).
- """
- _POLL = 3 # seconds between retries in the watcher thread
-
- lock = self._read_lock()
- owner = lock.get("owner_id", "unknown") if lock else "none"
- age = _heartbeat_age_s(lock.get("heartbeat")) if lock else float("inf")
-
- if age >= _LOCK_STALE_AFTER_S:
- if lock:
- self._log(VERBOSITY_NORMAL,
- f"Stale lock (owner: '{owner}', {age:.0f}s ago). Taking over.")
- self._write_lock()
- return True
-
- # Lock is fresh — spin up a background watcher and return immediately.
- self._log(VERBOSITY_NORMAL,
- f"WebpageGenerator: lock held by '{owner}' ({age:.0f}s ago). "
- f"Will take over automatically when it expires.")
-
- def _watcher():
- while True:
- time.sleep(_POLL)
- lk = self._read_lock()
- age = _heartbeat_age_s(lk.get("heartbeat")) if lk else float("inf")
- self._log(VERBOSITY_VERBOSE,
- f" …waiting for lock, age {age:.0f}s / {_LOCK_STALE_AFTER_S}s")
- if age >= _LOCK_STALE_AFTER_S:
- break
- owner2 = lk.get("owner_id", "unknown") if lk else "none"
- self._log(VERBOSITY_VERBOSE,
- f"Lock expired (was '{owner2}'). Taking over.")
- self._write_lock()
- self._launch()
-
- t = threading.Thread(target=_watcher, name="WebpageGeneratorWatcher", daemon=True)
- t.start()
- return False
-
- def _write_lock(self) -> None:
- self._bec.set_global_var(_LOCK_VAR_KEY, {
- "owner_id": self._owner_id,
- "heartbeat": _now_iso(),
- "pid": os.getpid(),
- "hostname": socket.gethostname(),
- })
-
- def _read_lock(self) -> dict:
- val = self._bec.get_global_var(_LOCK_VAR_KEY)
- return val if isinstance(val, dict) else {}
-
- def _release_lock(self) -> None:
- lock = self._read_lock()
- if lock.get("owner_id") == self._owner_id:
- self._bec.delete_global_var(_LOCK_VAR_KEY)
-
- # ------------------------------------------------------------------
- # Main loop
- # ------------------------------------------------------------------
-
- def _run(self) -> None:
- while not self._stop_event.is_set():
- t0 = _epoch()
- try:
- self._cycle()
- except Exception as exc:
- self._log(VERBOSITY_NORMAL,
- f"WebpageGenerator cycle error: {exc}", level="warning")
- if not self._local_only:
- # The singleton lock coordinates real sessions for this
- # account; a mismatched-account fallback session must never
- # claim it (see _launch()'s local_only docstring).
- try:
- self._write_lock()
- except Exception:
- pass
- self._stop_event.wait(max(0.0, self._cycle_interval - (_epoch() - t0)))
-
- def _cycle(self) -> None:
- """One generator cycle: read state -> derive status -> write files -> upload."""
-
- # ── Progress ─────────────────────────────────────────────────
- progress = self._bec.get_global_var("tomo_progress") or {}
-
- # ── Queue ────────────────────────────────────────────────────
- # queue_status reflects BEC's ScanQueueStatus: 'RUNNING', 'PAUSED',
- # or 'LOCKED' (the latter set whenever any lock — e.g. from
- # ScanInterlockActor — is applied to the queue; the prior status is
- # restored once the lock is released).
- # A scan is actually executing only when info is non-empty AND
- # active_request_block is set on the first entry.
- try:
- qi = self._bec.queue.queue_storage.current_scan_queue
- primary = qi.get("primary")
- queue_status = primary.status if primary is not None else "unknown"
- queue_has_active_scan = (
- primary is not None
- and len(primary.info) > 0
- and primary.info[0].active_request_block is not None
- )
- queue_locks = _read_queue_locks(primary)
- except Exception:
- queue_status = "unknown"
- queue_has_active_scan = False
- queue_locks = []
-
- # ── Beamline states ──────────────────────────────────────────
- # Diagnostic display AND (via beamline_blocking below) one of the two
- # signals that drive the 'blocked' experiment_status; the other is
- # queue_locks. A watched state that is out of its accepted range
- # (mismatched) means the scan interlock is / would be holding the
- # queue, even if the running scan has already been interrupted and
- # active_request_block cleared.
- try:
- beamline_states_info = _read_beamline_states_info(self._bec)
- except Exception as exc:
- self._log(VERBOSITY_VERBOSE,
- f"beamline_states read error: {exc}", level="warning")
- beamline_states_info = {"enabled": None, "states": []}
-
- # True if the scan interlock is enabled AND at least one watched state
- # is out of its accepted range. Gated on `enabled`: a mismatched state
- # while the interlock is disabled does not actually block a scan, so it
- # must not raise the 'blocked' status. Kept in lock-step with the
- # beamline-states card summary (renderBeamlineStates), which likewise
- # only calls mismatched states "blocking" when the interlock is enabled,
- # so the top status pill and that card can never disagree.
- beamline_blocking = bool(
- beamline_states_info.get("enabled") is True
- and any(s.get("mismatched") for s in beamline_states_info.get("states", []))
- )
-
- # Current scan number (the BEC scan in progress right now, e.g. for
- # comparison against ptycho reconstruction filenames like S06770).
- # Only meaningful while a scan is actually active; None otherwise.
- current_scan_number = None
- if queue_has_active_scan:
- try:
- current_scan_number = primary.info[0].active_request_block.scan_number
- except Exception:
- current_scan_number = None
-
- # ── Idle tracking ────────────────────────────────────────────
- hb_age = _heartbeat_age_s(progress.get("heartbeat"))
- tomo_active = hb_age < _TOMO_HEARTBEAT_STALE_S
-
- # Detect scans that start and finish entirely between two polls
- # by watching for a new queue_id at the top of history.
- try:
- history = self._bec.queue.queue_storage.queue_history
- latest_queue_id = history[0].info.queue_id if history else None
- except Exception:
- latest_queue_id = None
-
- history_changed = (
- latest_queue_id is not None
- and latest_queue_id != self._last_queue_id
- )
- self._last_queue_id = latest_queue_id
-
- if (tomo_active or queue_has_active_scan or history_changed
- or beamline_blocking):
- self._last_active_time = _epoch()
- self._had_activity = True
-
- exp_status = _derive_status(
- progress, queue_has_active_scan,
- self._last_active_time, self._had_activity,
- queue_locks, beamline_blocking,
- )
- idle_for_s = (
- None if self._last_active_time is None
- else max(0.0, _epoch() - self._last_active_time)
- )
-
- # ── Tomo queue (setup-specific; see HAS_TOMO_QUEUE) ─────────────
- # Persisted list of parameter-snapshot jobs (global var "tomo_queue").
- # Each job: {"label": str, "params": {...},
- # "status": pending|running|incomplete|done, "added_at": ISO str}
- # Instruments without this feature (e.g. LamNI) never touch the
- # global var and always report an empty queue.
- if self.HAS_TOMO_QUEUE:
- tomo_queue = self._bec.get_global_var("tomo_queue") or []
- else:
- tomo_queue = []
-
- # ── Reconstruction queue ──────────────────────────────────────
- recon = self._collect_recon_data()
-
- # ── Ptychography images ───────────────────────────────────────
- # Snapshot scan id BEFORE collecting so we can detect changes
- prev_ptycho_scan = self._last_ptycho_scan
-
- ptycho = {}
- try:
- ptycho = self._collect_ptycho_images()
- except Exception as exc:
- self._log(VERBOSITY_VERBOSE, f"ptycho image error: {exc}", level="warning")
-
- # ── Setup-specific data (subclass hook) ───────────────────────
- setup = {}
- try:
- setup = self._collect_setup_data()
- except Exception as exc:
- self._log(VERBOSITY_VERBOSE, f"setup data error: {exc}", level="warning")
-
- # ── Payload ───────────────────────────────────────────────────
- payload = {
- "generated_at": _now_iso(),
- "generated_at_epoch": _epoch(),
- "experiment_status": exp_status,
- "queue_status": queue_status,
- "queue_has_active_scan": queue_has_active_scan,
- "idle_for_s": idle_for_s,
- "idle_for_human": _format_duration(idle_for_s),
- "queue_locks": queue_locks,
- "beamline_states": beamline_states_info,
- "tomo_queue": tomo_queue,
- "progress": {
- "tomo_type": progress.get("tomo_type", "N/A"),
- "projection": progress.get("projection", 0),
- "total_projections": progress.get("total_projections", 0),
- "subtomo": progress.get("subtomo", 0),
- "subtomo_projection": progress.get("subtomo_projection", 0),
- "subtomo_total_projections": progress.get("subtomo_total_projections", 1),
- "angle": progress.get("angle", 0),
- "tomo_start_time": progress.get("tomo_start_time"),
- "tomo_start_scan_number": progress.get("tomo_start_scan_number"),
- "current_scan_number": current_scan_number,
- "estimated_remaining_s": progress.get("estimated_remaining_time"),
- "estimated_remaining_human": _format_duration(
- progress.get("estimated_remaining_time")
- ),
- "estimated_finish_time": progress.get("estimated_finish_time"),
- "tomo_heartbeat": progress.get("heartbeat"),
- "tomo_heartbeat_age_s": round(hb_age, 1) if hb_age != float("inf") else None,
- },
- "recon": recon,
- "ptycho": ptycho,
- "setup": setup,
- "generator": {
- "owner_id": self._owner_id,
- "cycle_interval_s": self._cycle_interval,
- "active_account": getattr(self._bec, "active_account", None) or "",
- },
- }
-
- # ── Write status.json ─────────────────────────────────────────
- # (HTML is static, written once at _launch())
- (self._output_dir / "status.json").write_text(
- json.dumps(payload, indent=2, default=str)
- )
-
- # ── Upload ────────────────────────────────────────────────────
- # Uploads run in a background daemon thread — never blocks this cycle.
- # If the server is unreachable the cycle continues normally; failures
- # are logged as warnings. If an upload is still in progress from the
- # previous cycle the new request is dropped (logged at DEBUG level).
- # Never upload from the account-mismatch fallback (local_only) — the
- # active account isn't the one the remote page's auth is set up for.
- if self._uploader is not None and not self._local_only:
- new_scan = ptycho.get("scan_id")
- if new_scan and new_scan != prev_ptycho_scan:
- # Scan changed: ask server to delete old S*_*.png/jpg first,
- # then upload all current files (including new ptycho images).
- self._log(VERBOSITY_VERBOSE,
- f"Ptycho scan changed {prev_ptycho_scan} -> {new_scan}, "
- f"triggering remote cleanup + full upload")
- self._uploader.cleanup_ptycho_images_async()
- # Small delay so cleanup runs before the file uploads start.
- # Both are async, so we schedule the full upload on a short timer.
- def _delayed_full_upload():
- time.sleep(2)
- self._uploader.upload_dir_async(self._output_dir)
- threading.Thread(target=_delayed_full_upload,
- name="HttpUploaderDelayed", daemon=True).start()
- else:
- # Normal cycle: only upload files that have changed since last cycle.
- self._uploader.upload_changed_async(self._output_dir)
-
- # ── Console feedback ──────────────────────────────────────────
- blocked_summary = (
- f" locked_by={','.join(l['identifier'] for l in queue_locks)}"
- if queue_locks else ""
- )
- self._log(VERBOSITY_VERBOSE,
- f"[{_now_iso()}] {exp_status:<12} active={queue_has_active_scan} "
- f"proj={payload['progress']['projection']}/"
- f"{payload['progress']['total_projections']} "
- f"hb={payload['progress']['tomo_heartbeat_age_s']}s "
- f"idle={_format_duration(idle_for_s)}" + blocked_summary)
- self._log(VERBOSITY_DEBUG,
- f" payload:\n{json.dumps(payload, indent=4, default=str)}")
-
- # ------------------------------------------------------------------
- # Reconstruction queue
- # ------------------------------------------------------------------
-
- def _collect_recon_data(self) -> dict:
- foldername = (
- self._bec.get_global_var("ptycho_reconstruct_foldername") or "ptycho_reconstruct"
- )
- base = Path("~/data/raw/logs/reconstruction_queue").expanduser()
- queue_dir = base / foldername
- failed_dir = queue_dir / "failed"
-
- waiting = failed = 0
- try:
- if queue_dir.exists():
- waiting = len(list(queue_dir.glob("*.dat")))
- except Exception:
- pass
- try:
- if failed_dir.exists():
- failed = len(list(failed_dir.glob("*.dat")))
- except Exception:
- pass
-
- return {
- "foldername": foldername,
- "folder_path": str(queue_dir),
- "waiting": waiting,
- "failed": failed,
- }
-
- # ------------------------------------------------------------------
- # Ptychography images
- # ------------------------------------------------------------------
-
- def _collect_ptycho_images(self) -> dict:
- """Find the latest ptycho reconstruction, copy/thumbnail changed images.
-
- Scans _PTYCHO_ONLINE_DIR for the most-recently-modified subfolder whose
- name contains an underscore (e.g. dset_00005). Within that folder looks
- for PNG files matching S*_phase.png, S*_probe.png, S*_amplitude.png,
- S*_err.png, S*_spectrum.png.
-
- When the scan changes (new S##### prefix detected):
- - Thumbnails are generated with Pillow and saved to output_dir
- - Full-size PNGs are also copied to output_dir (for click-to-full)
- - Old ptycho files from previous scans are removed from output_dir
-
- Returns a dict consumed by the JS ptycho renderer:
- scan_id str e.g. "S06666"
- images list [{"role": "phase", "thumb": "..._thumb.jpg",
- "full": "..._phase.png"}, ...]
- folder str source folder path (for diagnostics)
- """
- base = Path(_PTYCHO_ONLINE_DIR).expanduser()
- if not base.exists():
- return {}
-
- # Find latest subfolder by modification time.
- candidates = sorted(
- [d for d in base.iterdir() if d.is_dir()],
- key=lambda d: d.stat().st_mtime,
- )
- if not candidates:
- return {}
- latest_dir = candidates[-1]
-
- # Roles we care about — phase and probe are primary (always shown),
- # amplitude/err/spectrum are secondary (shown on expand)
- _ROLES = [
- ("phase", "S*phase.png", True),
- ("probe", "S*probe.png", True),
- ("amplitude", "S*amplitude.png", False),
- ("error", "S*err.png", False),
- ("spectrum", "S*spectrum.png", False),
- ]
-
- # Collect available source files
- found = {}
- scan_id = None
- for role, pattern, _primary in _ROLES:
- matches = sorted(latest_dir.glob(pattern))
- if matches:
- src = matches[-1]
- stem = src.stem # e.g. S06666_400x400_b0_run_1_phase
- sid = stem.split("_")[0]
- if scan_id is None:
- scan_id = sid
- found[role] = src
-
- if not scan_id:
- return {}
-
- # If scan unchanged, thumbs exist, and source hasn't changed, skip all work
- thumb_key = f"{scan_id}_phase_thumb.jpg"
- thumb_path = self._output_dir / thumb_key
- src_phase = found.get("phase")
- thumb_fresh = (
- thumb_path.exists()
- and src_phase is not None
- and thumb_path.stat().st_mtime >= src_phase.stat().st_mtime
- )
- if (scan_id == self._last_ptycho_scan and thumb_fresh):
- # Rebuild result dict from existing files without doing I/O
- images = []
- for role, _pat, primary in _ROLES:
- thumb = self._output_dir / f"{scan_id}_{role}_thumb.jpg"
- full = self._output_dir / f"{scan_id}_{role}.png"
- if thumb.exists() and full.exists():
- images.append({
- "role": role,
- "thumb": thumb.name,
- "full": full.name,
- "primary": primary,
- })
- return {"scan_id": scan_id, "images": images, "folder": str(latest_dir)}
-
- # Scan changed (or first run) — clean up old ptycho files in output_dir
- self._log(VERBOSITY_VERBOSE,
- f"New ptycho scan detected: {scan_id} in {latest_dir.name}")
- for old_file in self._output_dir.glob("S*_*_thumb.jpg"):
- if not old_file.name.startswith(scan_id + "_"):
- old_file.unlink(missing_ok=True)
- for old_file in self._output_dir.glob("S*_*.png"):
- if not old_file.name.startswith(scan_id + "_"):
- for role, _, _ in _ROLES:
- if f"_{role}." in old_file.name or f"_{role}_" in old_file.name:
- old_file.unlink(missing_ok=True)
- break
-
- # Generate thumbnails and copy full-size for each found role
- images = []
- for role, _pat, primary in _ROLES:
- src = found.get(role)
- if src is None:
- continue
- thumb_name = f"{scan_id}_{role}_thumb.jpg"
- full_name = f"{scan_id}_{role}.png"
- thumb_path = self._output_dir / thumb_name
- full_path = self._output_dir / full_name
- try:
- if _PIL_AVAILABLE:
- img = _PILImage.open(src)
- img.thumbnail((_PTYCHO_THUMB_PX, _PTYCHO_THUMB_PX))
- img.save(thumb_path, "JPEG", quality=85)
- else:
- shutil.copy2(src, thumb_path)
- shutil.copy2(src, full_path)
- images.append({
- "role": role,
- "thumb": thumb_name,
- "full": full_name,
- "primary": primary,
- })
- self._log(VERBOSITY_VERBOSE, f" ptycho {role}: thumb={thumb_name}")
- except Exception as exc:
- self._log(VERBOSITY_VERBOSE,
- f" ptycho {role} failed: {exc}", level="warning")
-
- self._last_ptycho_scan = scan_id
- return {"scan_id": scan_id, "images": images, "folder": str(latest_dir)}
-
- # ------------------------------------------------------------------
- # Subclass hooks
- # ------------------------------------------------------------------
-
- def _collect_setup_data(self) -> dict:
- """
- Override in subclasses to return a dict of setup-specific data.
- The dict is embedded in the JSON payload as 'setup' and rendered
- by the HTML page as the 'Instrument details' section.
-
- Expected keys (all optional):
- type (str) -- setup identifier, e.g. "flomni"
- sample_name (str) -- current sample name
- temperatures (dict) -- label -> float (degrees C) or None
- current_params (dict) -- raw tomo parameters read live (keyed by
- the same global-var names the tomo queue
- stores in job.params, e.g. tomo_countingtime,
- fovx, tomo_angle_stepsize, ...). Rendered by
- the page through the same param-table path
- as a queue job (buildParamRows), so the
- 'Current measurement' section always matches
- the queue's level of detail -- and shows the
- details even for a scan started outside the
- queue. Values are left raw; the page formats
- them (fmtTqParam) and computes the projection
- count (calcProjections).
- """
- return {}
-
- # ------------------------------------------------------------------
- # Logging
- # ------------------------------------------------------------------
-
- def _log(self, min_verbosity: int, msg: str, level: str = "info") -> None:
- if self._verbosity < min_verbosity:
- return
- if level == "warning":
- logger.warning(msg)
- elif level == "error":
- logger.error(msg)
- else:
- print(msg)
-
-
-# ---------------------------------------------------------------------------
-# flOMNI subclass
-# ---------------------------------------------------------------------------
class FlomniWebpageGenerator(WebpageGeneratorBase):
"""
@@ -1502,6 +82,11 @@ class FlomniWebpageGenerator(WebpageGeneratorBase):
"Mirror": "flomni_temphum.temperature_mirror",
}
+ # Explicitly set (not just inherited) so flomni's page keeps rendering
+ # its own param names/labels even if WebpageGeneratorBase's default ever
+ # changes for another setup.
+ TQ_PARAM_DISPLAY = WebpageGeneratorBase.TQ_PARAM_DISPLAY
+
def _logo_path(self):
return Path(__file__).parent / "flOMNI.png"
@@ -1540,1504 +125,3 @@ class FlomniWebpageGenerator(WebpageGeneratorBase):
"temperatures": temperatures,
"current_params": current_params,
}
-
-
-# ---------------------------------------------------------------------------
-# Factory
-# ---------------------------------------------------------------------------
-
-def make_webpage_generator(bec_client, **kwargs):
- """
- Select the appropriate WebpageGenerator subclass based on the BEC session
- name (bec._ip.prompts.session_name) and return a ready-to-start instance.
-
- Usage (generic startup script):
- from csaxs_bec.bec_ipython_client.plugins.flomni.webpage_generator import (
- make_webpage_generator,
- )
- gen = make_webpage_generator(bec, output_dir="~/data/raw/webpage/",
- upload_url="http://omny.online/upload.php")
- gen.start()
- """
- try:
- session = bec_client._ip.prompts.session_name
- except Exception:
- session = "unknown"
-
- if session == "flomni":
- return FlomniWebpageGenerator(bec_client, **kwargs)
-
- elif session == "lamni":
- from csaxs_bec.bec_ipython_client.plugins.LamNI.webpage_generator import (
- LamniWebpageGenerator,
- )
- return LamniWebpageGenerator(bec_client, **kwargs)
-
- elif session == "omny":
- from csaxs_bec.bec_ipython_client.plugins.omny.webpage_generator import (
- OmnyWebpageGenerator,
- )
- return OmnyWebpageGenerator(bec_client, **kwargs)
-
- else:
- return WebpageGeneratorBase(bec_client, **kwargs)
-
-
-# ---------------------------------------------------------------------------
-# HTML template (written once at start(), not every cycle)
-# Three themes: dark (default), light, auto (follows OS preference).
-# Theme is stored in localStorage and applied via data-theme on .
-# ---------------------------------------------------------------------------
-
-def _render_html(phone_numbers: list, has_tomo_queue: bool = True, tomo_types: dict = None) -> str:
- phones_html = "\n".join(
- f'
'
- f'{label}'
- f'{num}
'
- for label, num in phone_numbers
- )
-
- # Tomo queue card + its slot in the default card order are only emitted
- # for setups that actually have the feature (see HAS_TOMO_QUEUE).
- if has_tomo_queue:
- tomo_queue_card_html = """
-
-
-
⋮⋮
-
Tomo queue
-
-
-"""
- else:
- tomo_queue_card_html = ""
-
- default_order = ['audio', 'recon-queue', 'ptycho', 'instrument', 'blstates', 'contacts']
- if has_tomo_queue:
- default_order.insert(default_order.index('contacts'), 'tomo-queue')
- default_order_json = json.dumps(default_order)
-
- # Declarative tomo-type -> projection-count formula, consumed by the
- # generic calcProjections() in JS instead of a hardcoded per-instrument
- # branch. See WebpageGeneratorBase.TOMO_TYPES for the schema.
- tomo_types_json = json.dumps(tomo_types or {})
-
- return f"""
-
-
-
-
- Experiment Status
-
-
-
-
-
-
- ⚠ Status page data is outdated — generator may have stopped
-