Source code for labfrog.common
# SPDX-FileCopyrightText: 2023-2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
import ast
import logging
import os
import threading
import time
from dataclasses import dataclass, field, replace
from pathlib import Path
from queue import Empty, Queue
from typing import Any
from dotenv import load_dotenv
from flask import current_app, g, has_app_context
from mwclient import Site
from labfrog.selectables.mediawiki import (
retrieveCampaigns,
retrievePCs,
retrieveTargets,
)
DEFAULT_COUNTER_MODES = (
"OFF",
"free running (not locked to laser)",
"triggered but not synchronized (matching by timestamps)",
"shot-counter (triggered and counter matches shot counter)",
)
DEFAULT_FILENAME_SCHEMAS = (
"OFF",
"_YYYY-MM-DD_HHh-MMm-SSs_COUNTER_original.png",
"_YYYY-MM-DD_HHh-MMm-SSs_COUNTER.png",
"_COUNTER.tif",
)
[docs]
@dataclass(frozen=True)
class Settings:
mediawiki_config_enabled: bool = False
mediawiki_timeout_seconds: float = 3.0
mediawiki_retry_available: bool = False
use_wiki: bool = False
custom_options: bool = False
custom_options_collection: str = ""
custom_options_campaigns: str = ""
custom_options_targets: str = ""
demo_mode: bool = False
demo_dropdowns: bool = False
demo_default_campaign: str = ""
demo_default_layout: str = ""
demo_default_header: str = ""
demo_lock_layout: bool = False
demo_lock_header: bool = False
demo_locked_layouts: tuple[str, ...] = ()
demo_locked_headers: tuple[str, ...] = ()
demo_campaign_dict: dict[str, Any] = field(default_factory=dict)
demo_target_dict: dict[str, Any] = field(default_factory=dict)
mediawiki_host: str = "athene.fz-rossendorf.de"
mediawiki_path: str = "/fwk/"
external_services_disabled: bool = False
mediawiki_env_disabled: bool = False
mediawiki_runtime_disabled: bool = False
[docs]
def external_services_disabled() -> bool:
return bool(
os.environ.get("LABFROG_TESTING")
or os.environ.get("PYTEST_CURRENT_TEST")
or os.environ.get("SKIP_CUSTOM_OPTIONS")
)
[docs]
def request_cache_bucket(name: str) -> dict:
cache = getattr(g, name, None)
if cache is None:
cache = {}
setattr(g, name, cache)
return cache
[docs]
def mediawiki_cache_bucket() -> dict:
return current_app.extensions.setdefault(
"labfrog_mediawiki_cache",
{
"target_dict": {},
"pc_dict": {"NONE": "()"},
"campaign_dict": {"NONE": "()"},
"loaded": False,
"last_error": None,
"last_refresh_ts": None,
},
)
[docs]
def sync_request_mediawiki_state_from_cache() -> None:
cache = mediawiki_cache_bucket()
g.target_dict = dict(cache.get("target_dict") or {})
g.pc_dict = dict(cache.get("pc_dict") or {"NONE": "()"})
g.campaign_dict = dict(cache.get("campaign_dict") or {"NONE": "()"})
[docs]
def refresh_mediawiki_cache(timeout=None) -> bool:
cache = mediawiki_cache_bucket()
targets, pcs, campaigns = fetch_mediawiki_selectables(timeout=timeout)
if not (targets or pcs or campaigns):
cache["last_error"] = "Refresh failed"
return False
cache["target_dict"] = targets or {}
cache["pc_dict"] = pcs or {"NONE": "()"}
cache["campaign_dict"] = campaigns or {"NONE": "()"}
cache["loaded"] = True
cache["last_error"] = None
cache["last_refresh_ts"] = time.time()
g.target_dict = dict(cache["target_dict"])
g.pc_dict = dict(cache["pc_dict"])
g.campaign_dict = dict(cache["campaign_dict"])
return True
[docs]
def parse_bool(value: Any) -> bool:
return str(value).strip().lower() in {"true", "1", "yes", "on"}
[docs]
def parse_name_list(value):
if not value:
return []
text = str(value).strip()
if not text:
return []
if len(text) >= 2 and text[0] == text[-1] and text[0] in {'"', "'"}:
text = text[1:-1].strip()
if text.startswith("[") or text.startswith("("):
try:
parsed = ast.literal_eval(text)
except Exception:
parsed = None
if isinstance(parsed, (list, tuple, set)):
cleaned = []
for item in parsed:
item_text = str(item).strip()
if (
len(item_text) >= 2
and item_text[0] == item_text[-1]
and item_text[0] in {'"', "'"}
):
item_text = item_text[1:-1].strip()
if item_text:
cleaned.append(item_text)
return cleaned
if isinstance(parsed, str):
text = parsed
cleaned = []
for item in text.split(","):
item_text = item.strip()
if (
len(item_text) >= 2
and item_text[0] == item_text[-1]
and item_text[0] in {'"', "'"}
):
item_text = item_text[1:-1].strip()
if item_text:
cleaned.append(item_text)
return cleaned
[docs]
def init_request_state():
if not hasattr(g, "target_dict"):
g.target_dict = {}
if not hasattr(g, "pc_dict"):
g.pc_dict = empty_pc_choices()
if not hasattr(g, "campaign_dict"):
g.campaign_dict = empty_campaign_choices()
if not hasattr(g, "custom_target_choices"):
g.custom_target_choices = {}
if not hasattr(g, "custom_campaign_choices"):
g.custom_campaign_choices = {}
[docs]
def read_config(file_path):
config = {}
in_block_comment = False
with Path(file_path).open(encoding="utf-8") as f:
for line in f:
stripped = line.strip()
if not stripped:
continue
if stripped.startswith("'''") or stripped.startswith('"""'):
in_block_comment = not in_block_comment
continue
if in_block_comment or stripped.startswith("#") or "=" not in stripped:
continue
key, value = stripped.split("=", 1)
config[key.strip()] = value.strip()
return config
[docs]
def resolve_config_name(config: dict | None = None, environ=None) -> str:
"""Return the active instance config file name from config/env fallbacks."""
source_config = config or {}
source_environ = environ or os.environ
return (
source_config.get("CONFIG_NAME")
or source_config.get("CONFIG_FILE")
or source_environ.get("CONFIG_NAME")
or source_environ.get("CONFIG_FILE")
or "default.cfg"
)
[docs]
def load_configuration_from_files():
current_directory = os.path.dirname(__file__)
instance_directory = os.path.join(os.path.dirname(current_directory), "instance")
load_dotenv(dotenv_path=Path(__file__).resolve().parents[1] / ".env", override=True)
default_config_file = (
"shotsheet_protected.cfg"
if os.environ.get("FLASK_APP_EXECUTOR") in {"waitress", "gunicorn"}
else "default.cfg"
)
config_file = (
os.environ.get("CONFIG_NAME")
or os.environ.get("CONFIG_FILE")
or default_config_file
)
full_config_file_path = os.path.join(instance_directory, config_file)
default_config_path = os.path.join(instance_directory, "default.cfg")
try:
config = read_config(full_config_file_path)
except FileNotFoundError:
logging.error("Configuration file not found: %s", full_config_file_path)
config = {}
try:
default_config = read_config(default_config_path)
except FileNotFoundError:
logging.error("Default configuration file not found: %s", default_config_path)
default_config = {}
return current_directory, config, default_config
[docs]
def build_settings(config: dict, default_config: dict) -> Settings:
skip_external = external_services_disabled()
skip_mediawiki = mediawiki_env_disabled()
mediawiki_enabled = parse_bool(config.get("USE_WIKI", False))
custom_options = (
False if skip_external else parse_bool(config.get("USE_CUSTOM_OPTIONS", False))
)
demo_mode = parse_bool(config.get("DEMO_MODE", False))
demo_dropdowns = parse_bool(config.get("DEMO_DROPDOWNS", False))
return Settings(
mediawiki_config_enabled=mediawiki_enabled,
mediawiki_timeout_seconds=float(config.get("MEDIAWIKI_TIMEOUT_SECONDS", 3.0)),
mediawiki_retry_available=False,
use_wiki=mediawiki_enabled and not skip_mediawiki and not skip_external,
custom_options=custom_options,
custom_options_collection=config.get("CUSTOM_OPTIONS_COLLECTION", "").replace(
'"', ""
),
custom_options_campaigns=config.get("CUSTOM_OPTIONS_CAMPAIGNS", "").replace(
'"', ""
),
custom_options_targets=config.get("CUSTOM_OPTIONS_TARGETS", "").replace(
'"', ""
),
demo_mode=demo_mode,
demo_dropdowns=demo_dropdowns,
demo_default_campaign=str(config.get("DEMO_DEFAULT_CAMPAIGN", "")).strip(),
demo_default_layout=str(config.get("DEMO_DEFAULT_LAYOUT", "")).strip(),
demo_default_header=str(config.get("DEMO_DEFAULT_HEADER", "")).strip(),
demo_lock_layout=parse_bool(config.get("DEMO_LOCK_LAYOUT", False)),
demo_lock_header=parse_bool(config.get("DEMO_LOCK_HEADER", False)),
demo_locked_layouts=tuple(parse_name_list(config.get("DEMO_LOCKED_LAYOUTS"))),
demo_locked_headers=tuple(parse_name_list(config.get("DEMO_LOCKED_HEADERS"))),
external_services_disabled=skip_external,
mediawiki_env_disabled=skip_mediawiki,
)
[docs]
def get_settings() -> Settings:
settings = current_app.extensions.get("labfrog_settings")
if settings is None:
settings = build_settings(current_app.config, {})
current_app.extensions["labfrog_settings"] = settings
return settings
[docs]
def mediawiki_disabled(settings: Settings | None = None) -> bool:
if settings is None:
if has_app_context():
active_settings = get_settings()
else:
return bool(external_services_disabled() or mediawiki_env_disabled())
else:
active_settings = settings
return bool(
active_settings.external_services_disabled
or active_settings.mediawiki_env_disabled
or active_settings.mediawiki_runtime_disabled
)
[docs]
def get_mediawiki_site():
site = current_app.extensions.get("mediawiki_site")
if site is None:
settings = get_settings()
site = Site(settings.mediawiki_host, path=settings.mediawiki_path)
current_app.extensions["mediawiki_site"] = site
return site
[docs]
def create_mediawiki_site():
try:
return get_mediawiki_site()
except Exception:
logging.exception("Failed to connect to MediaWiki")
return None
[docs]
def disable_mediawiki_runtime(reason):
settings = get_settings()
current_app.extensions["labfrog_settings"] = replace(
settings,
use_wiki=False,
mediawiki_retry_available=True,
mediawiki_runtime_disabled=True,
)
logging.warning("MediaWiki disabled for this run: %s", reason)
[docs]
def run_with_timeout(func, *args, timeout):
result_queue = Queue(maxsize=1)
app_obj = current_app._get_current_object() if has_app_context() else None
def runner():
try:
if app_obj is None:
result_queue.put((True, func(*args)))
return
with app_obj.app_context():
result_queue.put((True, func(*args)))
except Exception as exc:
result_queue.put((False, exc))
thread = threading.Thread(target=runner, daemon=True)
thread.start()
try:
ok, payload = result_queue.get(timeout=timeout)
except Empty as exc:
raise TimeoutError from exc
if ok:
return payload
raise payload
[docs]
def try_create_mediawiki_site(timeout=3):
if mediawiki_disabled():
return None
try:
return run_with_timeout(create_mediawiki_site, timeout=timeout)
except TimeoutError:
logging.error("Timeout reached while connecting to MediaWiki.")
disable_mediawiki_runtime("timeout while connecting to MediaWiki")
return None
except Exception as exc:
logging.error("MediaWiki connection failed: %s", exc)
disable_mediawiki_runtime("connection failure")
return None
[docs]
def run_mediawiki_call(func, default, operation_name, timeout=None, site=None):
settings = get_settings()
if mediawiki_disabled(settings):
return default
site = site or get_mediawiki_site()
try:
return run_with_timeout(
func, site, timeout=timeout or settings.mediawiki_timeout_seconds
)
except TimeoutError:
logging.error("Timeout reached while running MediaWiki %s.", operation_name)
disable_mediawiki_runtime(f"timeout during {operation_name}")
return default
except Exception as exc:
logging.error("MediaWiki %s failed: %s", operation_name, exc)
disable_mediawiki_runtime(f"failure during {operation_name}")
return default
[docs]
def fetch_mediawiki_selectables(timeout=None):
settings = get_settings()
if mediawiki_disabled(settings):
return {}, {}, {}
site_timeout = timeout or settings.mediawiki_timeout_seconds
site = try_create_mediawiki_site(timeout=site_timeout)
if site is None:
return {}, {}, {}
targets = run_mediawiki_call(
retrieveTargets, {}, "target retrieval", site_timeout, site=site
)
pcs = run_mediawiki_call(retrievePCs, {}, "PC retrieval", site_timeout, site=site)
campaigns = run_mediawiki_call(
retrieveCampaigns, {}, "campaign retrieval", site_timeout, site=site
)
return targets, pcs, campaigns
[docs]
def retry_mediawiki_connection(timeout=None):
settings = get_settings()
if settings.external_services_disabled or settings.mediawiki_env_disabled:
return False
current_app.extensions["labfrog_settings"] = replace(
settings,
use_wiki=False,
mediawiki_retry_available=False,
mediawiki_runtime_disabled=False,
)
success = refresh_mediawiki_cache(timeout=timeout)
current_app.extensions["labfrog_settings"] = replace(
get_settings(),
use_wiki=bool(success),
mediawiki_retry_available=not bool(success),
mediawiki_runtime_disabled=not bool(success),
)
if not success:
disable_mediawiki_runtime("retry failed")
return success
[docs]
def mediawiki_retry_available():
settings = get_settings()
return settings.mediawiki_config_enabled and settings.mediawiki_retry_available
[docs]
def get_cached_mediawiki_campaigns() -> dict:
init_request_state()
cache = mediawiki_cache_bucket()
if not (g.campaign_dict or {}) and cache.get("campaign_dict"):
sync_request_mediawiki_state_from_cache()
settings = get_settings()
campaigns = dict(g.campaign_dict or {})
if settings.demo_dropdowns and settings.demo_mode and settings.demo_campaign_dict:
for name, details in settings.demo_campaign_dict.items():
if name and name not in campaigns:
campaigns[name] = details
return campaigns
[docs]
def get_cached_mediawiki_targets() -> dict:
init_request_state()
cache = mediawiki_cache_bucket()
if not (g.target_dict or {}) and cache.get("target_dict"):
sync_request_mediawiki_state_from_cache()
settings = get_settings()
targets = dict(g.target_dict or {})
if settings.demo_dropdowns and settings.demo_mode and settings.demo_target_dict:
for name, details in settings.demo_target_dict.items():
if name and name not in targets:
targets[name] = details
return targets
[docs]
def mediawiki_status_state() -> dict:
settings = get_settings()
enabled = bool(settings.mediawiki_config_enabled)
connected = enabled and bool(settings.use_wiki)
return {
"enabled": enabled,
"connected": connected,
"message": "MediaWiki connected" if connected else "MediaWiki not connected",
"action_label": "Refresh" if connected else "Try to connect",
}
[docs]
def initialize_common(app):
settings = build_settings(app.config, {})
app.extensions["labfrog_settings"] = settings
app.extensions.setdefault(
"labfrog_mediawiki_cache",
{
"target_dict": {},
"pc_dict": {"NONE": "()"},
"campaign_dict": {"NONE": "()"},
"loaded": False,
"last_error": None,
"last_refresh_ts": None,
},
)
if settings.use_wiki:
with app.app_context():
site = try_create_mediawiki_site(timeout=settings.mediawiki_timeout_seconds)
if site is not None:
app.extensions["mediawiki_site"] = site
return app