Python API reference

This reference is generated from the docstrings in the LabFrog source code. It covers the application factory, Flask routes, database and form services, optional integrations, and the reusable helper modules. Select View source beside an object to inspect its implementation.

Objects without explanatory text do not yet have a source docstring. They are still listed so that gaps stay visible instead of silently disappearing from the developer documentation.

__init__

Utilities for creating and configuring a Flask application instance.

labfrog.normalize_config_string(value)[source]
labfrog.infer_username_from_user_id(user_id: str) str[source]

Infer a display username from persisted auth identifiers.

Flask-Login stores User.get_id() in the session. For LDAP users this is commonly a DN like uid=jdoe,ou=users,…. Prefer concise username-like values for UI display, and fall back to the raw identifier when no safe parse is available.

labfrog.infer_helmholtz_username_from_session() str[source]

Infer a Helmholtz display username from stored OIDC profile fields.

labfrog.normalize_auth_method(app)[source]

Force browser-session settings that work for local test clients.

labfrog.ensure_oidc_client_secrets(app)[source]
labfrog.load_oidc_credentials(app)[source]
labfrog.create_oidc_client()[source]
labfrog.create_app(test_config=None)[source]

Site Routes (site_routes)

Shared app-level routes for session state, navigation, and utility redirects.

labfrog.site_routes.imprint()[source]
labfrog.site_routes.damnit_workflow()[source]
labfrog.site_routes.privacy()[source]
labfrog.site_routes.documentation()[source]

Open hosted docs when reachable, otherwise fall back to local built docs.

labfrog.site_routes.docs_index()[source]

Serve the local docs index from the short public docs path.

labfrog.site_routes.docs(path: str)[source]

Serve local built docs from the short public docs path.

labfrog.site_routes.local_documentation(path: str)[source]

Serve locally built Sphinx docs when present.

labfrog.site_routes.retry_mediawiki()[source]

Try reconnecting to MediaWiki and return to the active data-entry tab.

labfrog.site_routes.add_selectable(kind)[source]

Redirect to MediaWiki when available, otherwise to parameter options.

labfrog.site_routes.set_campaign()[source]

Store the active campaign in session and refresh its layout mapping.

labfrog.site_routes.set_mode()[source]

Store the active shooting mode in the session.

labfrog.site_routes.get_shot_number()[source]

Return the latest shot number (or latest shot within the newest set).

labfrog.site_routes.goto_shot_SHOT()[source]

Navigate to a specific shot entry while in shot mode.

labfrog.site_routes.goto_set()[source]

Navigate to a specific set entry while in set mode.

labfrog.site_routes.goto_shot_SET()[source]

Navigate from a shot number to the set entry that contains it.

login

labfrog.login.prepare_request_state()[source]
labfrog.login.user_factory()[source]
labfrog.login.user_store()[source]
labfrog.login.logged_in_users_store()[source]
labfrog.login.startup_state()[source]
labfrog.login.clear_startup_state_once()[source]
labfrog.login.remember_authenticated_user(user)[source]
labfrog.login.forget_authenticated_user()[source]
labfrog.login.is_public_request()[source]
labfrog.login.persist_user(dn, username, data)[source]
labfrog.login.extract_username(user_info)[source]
labfrog.login.sync_oidc_session_user()[source]
labfrog.login.with_app_prefix(path)[source]
labfrog.login.optional_url_for(endpoint, **values)[source]
labfrog.login.safe_next_page(default='/add_entry', consume=False)[source]
labfrog.login.clear_oidc_session()[source]
labfrog.login.oidc_tokens(oidc_ext)[source]
labfrog.login.oidc_single_logout(oidc_ext)[source]
labfrog.login.apply_post_login_demo_defaults()[source]
labfrog.login.enable_oidc_login(app, oidc_client)[source]
labfrog.login.add_user(username)[source]
labfrog.login.remove_user(username)[source]
labfrog.login.show_operations_dashboard()[source]

Render the operations dashboard with active sessions and recent updates.

labfrog.login.home()[source]

Redirect user to login or main application page.

labfrog.login.login()[source]
labfrog.login.logout()[source]
labfrog.login.logout_full()[source]
labfrog.login.check_login()[source]

db

This module provides utilities for managing a MongoDB database connection and initializing the database with data from JSON files.

The module contains functions to: - Establish a connection to the MongoDB database. - Close the database connection. - Initialize the database with data from JSON files. - Add a command to the Flask CLI for initializing the database.

Imports:
  • json: Used for loading data from JSON files.

  • quote from urllib.parse: Used for URL-encoding MongoDB credentials.

  • click: Used for creating and managing Flask CLI commands.

  • current_app, g from flask: Used for accessing the current Flask app context and storing data.

  • MongoClient from pymongo: Used for connecting to the MongoDB database.

Functions:
  • get_db: Establishes and returns a connection to the MongoDB database.

  • close_db: Closes the MongoDB database connection.

  • init_db: Initializes the database with data from JSON files.

  • init_db_command: Flask CLI command to call the init_db function.

  • init_app: Registers the database functions with the Flask app context.

class labfrog.db.InitPlan(reset: str, rebuild: str, source: str)[source]

Bases: object

reset: str
rebuild: str
source: str
labfrog.db.navigation_sort_for_mode(mode: str | None) list[tuple[str, int]][source]

Return the stable sort used by campaign navigation.

labfrog.db.clear_navigation_caches() None[source]

Clear cached navigation query results after shot/set writes.

labfrog.db.clear_runtime_metadata_caches() None[source]

Clear app/request caches derived from metadata collections.

labfrog.db.fetch_shot_navigation_blocks(shots_collection, query, projection: dict | None = None) list[list[dict]][source]

Return shot-mode navigation blocks in the same order shown in the UI.

labfrog.db.fetch_sorted_navigation_docs(shots_collection, query, *, mode: str, projection: dict | None = None) list[dict][source]

Return navigation docs in the order shown to the user.

labfrog.db.ensure_default_form_documents(include_organize_form: bool = True) dict[source]

Ensure default Select Fields and Organize Form docs exist and follow expected schema.

labfrog.db.get_db()[source]

Establish and return a connection to the MongoDB database.

If a connection already exists in the Flask app context, it is reused. Otherwise, a new connection is established using the app’s configuration.

Returns:

The MongoDB database instance.

Return type:

pymongo.database.Database

labfrog.db.ensure_runtime_indexes() None[source]

Create the hot-path indexes used by request-time navigation queries.

labfrog.db.fetch_shot_navigation_state(shots_collection, query, *, current_index: int, mode: str) dict[source]

Fetch the documents needed for navigation on tab1/tab2.

labfrog.db.close_db(e=None)[source]

Close the MongoDB database connection.

If a connection exists in the Flask app context, it is closed.

Parameters:

e (Exception, optional) – An exception that triggered the function call. Defaults to None.

labfrog.db.close_app_mongo_client(app=None)[source]

Close the shared app-level real Mongo client, if one exists.

labfrog.db.init_db(reset=None, rebuild=None, source=None)[source]

Initialize the MongoDB database with a simple plan-based API.

Parameters:
  • reset (str | None) – One of RESET_CHOICES to control collection drops.

  • rebuild (str | None) – One of REBUILD_CHOICES to control metadata rebuild.

  • source (str | None) – One of SOURCE_CHOICES to choose metadata source.

labfrog.db.init_app(app)[source]

Register the database functions with the Flask app context.

Parameters:

app (Flask) – The Flask app instance.

Kafka Listener (kafka_listener)

Optional Kafka consumer that caches the most recent message from a topic.

Attach it to new shot records before they are saved to MongoDB. All failures are caught and logged β€” a broken Kafka connection never blocks a shot record.

labfrog.kafka_listener.get_latest_kafka_snapshot(app) dict | None[source]

Return the most recent cached Kafka message, or None if none received yet.

labfrog.kafka_listener.start_kafka_listener(app) None[source]

Start the background Kafka consumer if KAFKA_ENABLED is True. Never raises.

In Flask debug mode the Werkzeug reloader runs create_app() in both the reloader parent process and the worker child. We skip the parent so only one consumer per group_id exists, preventing constant partition rebalances that would cause the listener to miss every message.

Add Entry (add_entry)

Compatibility-focused add-entry route refactor.

This pass keeps the route close to the current LabFrog helper/template contract while still using the newer typed field/payload helpers where they fit cleanly.

class labfrog.add_entry.SubmitResult(handled: 'bool' = False, response: 'object | None' = None)[source]

Bases: object

handled: bool = False
response: object | None = None
labfrog.add_entry.add_entry()[source]
labfrog.add_entry.start_target_series()[source]

Start a new same-target series and store it in the session.

labfrog.add_entry.finish_target_series()[source]

Close the active same-target series.

labfrog.add_entry.clear_target_series()[source]

Remove the active series state from the session.

Edit Entry (edit_entry)

Compatibility-focused edit-entry route refactor.

This follows the add-entry route structure so both tab1/tab2 keep a shared mental model while preserving edit-specific behavior (versioning + diffs).

labfrog.edit_entry.identify_matching_documents(old_id)[source]

Follow an edit chain backwards so the UI can compare historical versions.

labfrog.edit_entry.edit_entry()[source]

Offline Queue (offline_queue)

The page that shows what is waiting to reach MongoDB, and sends it.

labfrog.offline_queue.offline_queue()[source]

List the entries saved on this computer while the database was away.

labfrog.offline_queue.sync_offline_queue()[source]

Send the queued entries to MongoDB, keeping anything that will not go.

labfrog.offline_queue.offline_queue_banner_state() dict[source]

Return the small banner state every data-entry page shows.

Kept cheap: it counts files, it does not touch MongoDB, because this runs on pages that must render even when the database is the thing that is broken.

labfrog.offline_queue.inject_offline_queue_state()[source]

Expose the queue state to templates without each route passing it.

Offline Spool (offline_spool)

Keep entries safe when MongoDB cannot be reached.

The shotsheet competes with a spreadsheet, and a spreadsheet never tells an operator that their work went nowhere. So a database that is unreachable must not lose the entry and must not stall the page: the record is written to a file on this machine, the operator is told exactly where it is, and it is replayed into MongoDB when the connection comes back.

Two decisions matter here. The spool is plain JSON, one file per entry, in a directory operators can open – if every other part of this app failed, the shot is still readable and could be retyped from what is on disk. And nothing is ever deleted on replay: a synced file is moved into synced/, so an operator who wants to check what happened can still see it.

labfrog.offline_spool.safe_reason(reason: str | None) str[source]

Return a reason safe to write to disk and show on screen.

labfrog.offline_spool.spool_dir() Path[source]

Return the directory holding entries that could not be saved.

labfrog.offline_spool.record_unsaved_entry(document: dict, *, mode: str, reason: str, operation: str = 'insert', doc_id: str | None = None) Path[source]

Write one unsaved entry to the spool and return the file it landed in.

Input: the document the route was about to write, plus why it could not. Output: the path to show the operator. Raising here would defeat the purpose, so a spool failure is logged and re-raised only if even the directory cannot be created – at that point there is nowhere left to put it and the caller must say so plainly.

labfrog.offline_spool.pending_files() list[Path][source]

Return the spooled entries still waiting to reach MongoDB, oldest first.

labfrog.offline_spool.pending_count() int[source]
labfrog.offline_spool.load_pending() list[dict][source]

Return the pending entries as dictionaries, unreadable files included.

A file that will not parse is reported rather than skipped: silently ignoring it would be the same failure this module exists to prevent.

labfrog.offline_spool.replay_pending(shots_collection) dict[source]

Insert spooled entries into MongoDB, keeping whatever will not go in.

Output: counts plus per-file detail. Entries that fail stay in the spool with the reason written beside them, so a partial recovery never quietly drops the remainder.

labfrog.offline_spool.database_is_reachable(database) tuple[bool, str | None][source]

Ping the database. Returns (reachable, reason) and never raises.

Set Diagnostic Options (diagnostic_options)

Diagnostic preset management routes.

labfrog.diagnostic_options.diagnostic_options()[source]

Render the diagnostic options page and persist add/update operations.

labfrog.diagnostic_options.get_collection_names(collection_name)[source]

Return the currently selected diagnostics for the dropdown menu.

labfrog.diagnostic_options.get_option_lists(collection_name)[source]

Return stored filename/counter options for a diagnostic DisplayName.

labfrog.diagnostic_options.add_option(collection_name)[source]

Append a new option value to a diagnostic’s stored option lists.

labfrog.diagnostic_options.get_entries(collection_name)[source]

Return the entries stored for one diagnostic preset collection.

labfrog.diagnostic_options.update_active_status(collection_name, entry_id)[source]

Toggle the active status of one preset entry.

labfrog.diagnostic_options.get_entry_details(collection_name, entry_id)[source]

Return one preset entry as JSON for the tab 3 detail panel.

labfrog.diagnostic_options.clone_entry(collection_name, entry_id)[source]

Clone a preset entry into the same diagnostic collection.

labfrog.diagnostic_options.modify_entry(collection_name, entry_id)[source]

Modify a preset entry for the selected diagnostic collection.

labfrog.diagnostic_options.delete_entry(collection_name, entry_id)[source]

Delete one preset entry from a diagnostic collection.

Adjust Diagnostics (diagnostic_definitions)

Diagnostics Configuration Module for LabFrog - Tab 4 Adjusted Diagnostics.

This module provides backend functionalities for the adjusted diagnostics configuration tab (tab4) of LabFrog application. It includes routes and handlers for: - Displaying the main diagnostics configuration page. - Handling form submissions for adding, modifying, and cloning diagnostic entries. - Fetching specific diagnostic entries based on their IDs.

labfrog.diagnostic_definitions.diagnostic_definitions()[source]

Route for the main diagnostics configuration page. Handles both GET and POST requests.

Returns:

Renders the β€˜diagnostic_definitions.html’ template with the form, documents, and other relevant data.

Return type:

render_template

labfrog.diagnostic_definitions.get_document(document_id=None)[source]

Route to fetch a specific diagnostic entry based on its ID.

Returns:

Returns the diagnostic entry data as a JSON object.

Return type:

jsonify

labfrog.diagnostic_definitions.add_new()[source]
labfrog.diagnostic_definitions.toggle_diagnostic_in_current_templates(document_id)[source]

Toggle one diagnostic on/off in active Select Fields + Organized Form templates.

labfrog.diagnostic_definitions.save_diagnostic()[source]

Persist add/clone/modify actions for diagnostics from the tab 4 form.

Select Fields (field_selection)

labfrog.field_selection.campaign_layout_setup()[source]

Guide users when a campaign has no mapped form layout yet.

labfrog.field_selection.field_selection()[source]

Render the layout picker and persist layout field/campaign selections.

labfrog.field_selection.fetch_layout_fields()[source]

Fetch the field data for a selected layout from the database.

This route retrieves the custom field data associated with a particular layout selected by the user.

Returns:

Returns a JSON representation of the field data for the selected layout.

Return type:

jsonify

labfrog.field_selection.modify_layout(layout_id)[source]

Render and persist edits to an existing saved layout.

labfrog.field_selection.delete_layout(layout_id)[source]

Delete a specific layout identified by its ID.

The route allows users to delete a specific layout from the database based on its ID.

Returns:

A response message indicating the success or failure of the deletion operation.

Return type:

str

Search Records (search_records)

labfrog.search_records.search_records()[source]

Render and process the search form for querying shot data.

This route supports two modes:
  • POST: Submits the SearchForm, validates inputs, and retrieves data from the database according to user criteria (campaign, date range, quality, scan type, etc.). If successful, renders the template with results, plots, and a generated Python snippet for reproducibility.

  • GET: Allows search and projection queries to be passed directly as URL parameters (query, projection), parsed and sanitized for safety. Renders results if available.

Returns:

  • Renders the search form with retrieved results if valid data is found.

  • Renders the search form with no results if the query returns nothing.

  • Renders the search form with error messages if validation fails or query/projection parameters are invalid.

Return type:

flask.Response

Side Effects:
  • Uses session variables to remember selected fields, layout, and mode.

  • Flashes validation and query errors to the user.

  • Logs form validation errors.

labfrog.search_records.search_results_status()[source]

Return a compact fingerprint for the current search result set.

labfrog.search_records.export_search_records_sqlite()[source]

Download a SQLite snapshot for the last successful search result set.

labfrog.search_records.export_search_records_to_explorer()[source]

Redirect the browser to the Explorer for the current search campaign.

labfrog.search_records.update_data()[source]

Update a single field of a shot document by creating a new version.

Accepts JSON data containing a document ID, field name, new value, and field type. The function: 1. Archives the old document (status set to β€œarchived”). 2. Creates a new version of the document with the updated field value. 3. If the shot is in β€œset” mode and certain fields are updated, recalculates the shot_number_list.

Args (JSON body):

id (str): The MongoDB _id of the document to update. field (str): The field name to update. value (Any): The new value for the field. fieldType (str): The type of the field (int, float, bool, date, string, etc.).

Returns:

  • {β€œsuccess”: True} if the update succeeds.

  • {β€œsuccess”: False, β€œerror”: …} if the update fails.

Return type:

flask.Response (JSON)

Side Effects:
  • Inserts a new version of the document with incremented version number.

  • Flashes messages to the user on success/failure.

  • Logs errors when exceptions occur.

Raises:
  • Exception – Any error during database update or insertion is caught,

  • flashed to the user, and returned in the JSON response. –

labfrog.search_records.get_field_type(field_name)[source]

Return the inferred field type of a given form field.

Parameters:

field_name (str) – The name of the form field to check.

Returns:

A JSON object containing the field type (e.g., {β€œfield_type”: β€œstring”}).

Return type:

flask.Response (JSON)

labfrog.search_records.determineFieldType(fieldName, form)[source]

Determine the type of a form field by inspecting its WTForms class.

Parameters:
  • fieldName (str) – The name of the field to check.

  • form (FlaskForm) – The form class containing field definitions. Typically DynamicForm.

Returns:

The determined field type:
  • ”string”: StringField, TextAreaField (or default fallback)

  • ”int”: IntegerField

  • ”float”: FloatField

  • ”date”: DateField, DateTimeField

  • ”boolean”: BooleanField

  • ”enum”: SelectField, SelectMultipleField

str: The determined field type:
  • ”string”: StringField, TextAreaField (or default fallback)

  • ”int”: IntegerField

  • ”float”: FloatField

  • ”date”: DateField, DateTimeField

  • ”boolean”: BooleanField

  • ”enum”: SelectField, SelectMultipleField

Return type:

str

Organize Sections (organize_sections)

Section layout management (Organize Form page).

labfrog.organize_sections.organize_sections()[source]
labfrog.organize_sections.list_headers()[source]

List available layouts for organize form.

labfrog.organize_sections.header_data()[source]

Return layout section data by name (or default).

labfrog.organize_sections.save_header()[source]

Save organize-form section ordering back into the layout document.

labfrog.organize_sections.select_header()[source]

Select a layout for this session.

labfrog.organize_sections.template_health()[source]

Report organized forms that are broken, dead, or wrongly mapped.

labfrog.organize_sections.repair_templates()[source]

Repair broken organized forms in place, by name or across the mode.

labfrog.organize_sections.delete_header()[source]

Delete one organized form, or sweep away dead ones and stale mappings.

Sending a name removes that organized form and the campaign mappings that pointed at it. Sending prune_dead instead removes every organized form whose source layout is gone plus every campaign mapping whose header no longer resolves – the two cases that can only ever be dead weight. Callers confirm in the UI; this route does not prompt.

Set Parameter Options (parameter_options)

Parameter options management.

labfrog.parameter_options.parameter_options()[source]

Render the parameter options manager.

labfrog.parameter_options.list_parameter_options()[source]

Return options for a field (shared across layouts/modes).

labfrog.parameter_options.save_parameter_options()[source]

Save option set for a field (shared across layouts/modes).

Adjust Parameters (new_parameter)

labfrog.new_parameter.sanitize_field_name(value: str | None) str[source]

Normalize user-entered field names before persistence.

labfrog.new_parameter.verify_calculated_field()[source]

Validate calculated field configuration and return a safe preview value.

labfrog.new_parameter.new_parameter()[source]

Add a new custom field or modify an existing one.

This route provides an interface for users to add new custom fields or modify the details of existing ones. The function handles both the creation and modification operations based on the user’s input.

Returns:

Renders the β€˜new_parameter.html’ template with the appropriate context.

Return type:

render_template

labfrog.new_parameter.modify_field(field_name)[source]

Modify a specific custom field identified by its name.

This route provides an interface for users to modify the details of a custom field. It fetches the current details of the field and updates them based on the form data submitted by the user.

Returns:

Renders the β€˜new_parameter.html’ template with the updated context.

Return type:

render_template

labfrog.new_parameter.delete_field(field_name)[source]

Delete a specific custom field identified by its name.

The route allows users to delete a custom field from the database based on its name.

Returns:

Redirects the user back to the field management page.

Return type:

redirect

labfrog.new_parameter.add_field_to_current_templates(field_name)[source]

Add one field into the active campaign’s Selected Fields and Organized Form.

labfrog.new_parameter.toggle_field_in_current_templates(field_name)[source]

Toggle one field on/off in the active campaign’s Selected Fields and Organized Form.

Campaign Layout Setup (campaign_layout_setup)

Shared helpers for campaign layout setup workflows.

labfrog.campaign_layout_setup.mapped_layout_sources(parameters_collection, mode: str) list[dict][source]

Return campaign-mapped layouts with light metadata for setup helpers.

labfrog.campaign_layout_setup.build_campaign_layout_setup_context(parameters_collection, mode: str, campaign: str, next_url: str) dict[source]

Build the template context for the missing-campaign-layout chooser.

labfrog.campaign_layout_setup.handle_campaign_layout_setup_submission(parameters_collection, mode: str, campaign: str, next_url: str, form_data)[source]

Handle setup chooser submissions and return the next response.

Request State (request_state)

Compatibility-focused request/session state helpers for dynamic forms.

This version intentionally stays close to the current LabFrog helper contract and session semantics while still centralizing repeated route logic.

labfrog.request_state.ensure_campaign_layout_session_state(mode: str, campaign: str | None, *, parameters_collection=None, force: bool = False, hydrate_diagnostics: bool = False) bool[source]

Synchronize campaign layout/header session state only when needed.

labfrog.request_state.invalidate_diagnostic_choice_caches(*, collection_name: str | None = None) None[source]

Drop cached diagnostic preset choices so add/edit forms refresh immediately.

labfrog.request_state.get_custom_options_collection_name() str | None[source]
labfrog.request_state.choose_default_campaign_name() str[source]
labfrog.request_state.get_default_layout_doc(parameters_collection, mode: str)[source]
labfrog.request_state.get_active_layout_doc(parameters_collection, selected_layout: str, mode: str, *, campaign: str | None = None) dict | None[source]

Return the selected layout doc, falling back to the mode-specific default.

labfrog.request_state.sync_campaign_header_choice(mode: str, campaign: str | None, *, parameters_collection=None) None[source]
labfrog.request_state.get_active_header_doc(mode: str)[source]
labfrog.request_state.flatten_layout_fields(layout_doc: dict | None) list[str][source]
labfrog.request_state.refresh_diagnostic_choice_cache(visible_fields: list[str]) None[source]
labfrog.request_state.load_custom_field_documents(fields_collection=None) tuple[list[dict], dict[str, dict]][source]

Return the request-cached custom-field docs and their field_name lookup.

labfrog.request_state.invalidate_custom_field_doc_cache(*, collection_name: str | None = None) None[source]

Invalidate cached custom-field metadata for the current request.

labfrog.request_state.bootstrap_form_session_state(*, startup_session_key: str, ensure_defaults: bool = False, initialize_default_campaign: bool = False, apply_demo_defaults_on_startup: bool = False) bool[source]
labfrog.request_state.bootstrap_request_form_session_state(endpoint: str | None = None) bool[source]
labfrog.request_state.build_route_form_state(mode: str, parameters_collection, *, apply_demo_defaults_now: bool = False) dict[source]
labfrog.request_state.merge_layout_session_selection_state(parameters_collection, selected_layout: str, mode: str, *, ensure_diagnostics_match_selected: bool = False, default_selected_to_diagnostics: bool = False) tuple[list[str], list[str]][source]

Merge layout-backed and session-backed selection state and persist it.

labfrog.request_state.refresh_dynamic_form_bindings(*, mode: str | None = None, visible_fields: Iterable[str] | None = None)[source]
labfrog.request_state.collect_visible_custom_fields(form, visible_fields: Iterable[str], fields_cursor: Iterable[dict] | None) tuple[dict[str, str], dict[str, list[str]]][source]

Form Factory (form_factory)

Factories for forms that depend on runtime-bound selectable choices.

This module centralizes construction of fields and forms whose selectable choices depend on runtime state such as MediaWiki availability, custom option files, or session-derived visibility rules.

labfrog.form_factory.canonicalize_select_field_data(form)[source]

Normalize loaded select data to the exact case used by field choices.

labfrog.form_factory.apply_parameter_options(form)[source]

Replace select choices with the selected campaign-layout options.

labfrog.form_factory.apply_field_aliases(form)[source]

Replace visible labels with saved aliases while keeping field keys intact.

labfrog.form_factory.bind_runtime_field(form, field_name: str, field) None[source]

Bind a field to a form instance and preserve any existing value.

Parameters:
  • form – The bound WTForms form instance.

  • field_name – The name of the field on the form.

  • field – The WTForms unbound field to bind.

labfrog.form_factory.build_campaign_field()[source]

Build the campaign field from the current selectable source state.

labfrog.form_factory.build_target_field()[source]

Build the target field from the current selectable source state.

labfrog.form_factory.build_pc_field()[source]

Build the diagnostic PC field from the current selectable source state.

labfrog.form_factory.create_input_form(*args, **kwargs)[source]

Create an InputForm with runtime-bound selectable fields.

labfrog.form_factory.create_dynamic_form(*args, **kwargs)[source]

Create a DynamicForm with runtime-bound selectable fields.

labfrog.form_factory.create_diagnostics_form(*, choices: list[tuple[str, str]], **kwargs: Any)[source]

Create a DiagnosticsForm with runtime-bound selectable fields.

Parameters:
  • choices – Diagnostic type choices in (value, label) form.

  • **kwargs – Additional arguments forwarded to DiagnosticsForm.

Returns:

A fully initialized diagnostics form instance.

Return type:

DiagnosticsForm

shot_details

Primarily stuff for figuring out SETS Including calculating shot_number_list, etc.

labfrog.shot_details.update_indiv_count_NEW_SHOT(current_set_number, starting_shot_number, set_length)[source]

Update the individual count for a new shot.

This route calculates and updates the shot numbers for a new shot based on the current set number and set length. It works for new shots and doesn’t cater to modifications of existing shots.

Returns:

Total number of shots, number of shots at the current set number, and a list of shot numbers.

Return type:

tuple

labfrog.shot_details.get_total_number_of_shots()[source]

Retrieve the total number of shots across all sets.

Returns:

Total number of shots.

Return type:

int

labfrog.shot_details.string_get_total_number_of_shots()[source]

Fetch the total number of shots as a string.

This is useful for updating the frontend via AJAX requests.

Returns:

Total number of shots as a JSON string.

Return type:

jsonify

labfrog.shot_details.get_current_number_of_shots_at_set(set_number)[source]

Fetch the current number of shots at a specific set number.

Parameters:

set_number (int) – The set number for which to retrieve the shot count.

Returns:

Number of shots at the specified set number.

Return type:

int

labfrog.shot_details.get_total_number_of_shots_at_set(set_number)[source]

Fetch the total number of shots up to and including a specific set number.

Parameters:

set_number (int) – The set number up to which to retrieve the total shot count.

Returns:

Total number of shots up to and including the specified set number.

Return type:

int

labfrog.shot_details.get_REAL_highest_set_number(campaign=None)[source]

Get the highest set number in the database.

Parameters:

campaign (str, optional) – The campaign name. Defaults to None.

Returns:

The highest set number.

Return type:

int

labfrog.shot_details.string_get_highest_set_number()[source]

Retrieve the last set number.

Returns:

Last set number as a JSON string.

Return type:

jsonify

labfrog.shot_details.get_last_shot_number()[source]
labfrog.shot_details.generate_shot_numbers(starting_shot_number, set_length)[source]
labfrog.shot_details.get_set_length(set_number)[source]

Fetch the length (number of shots) of a specified set.

Parameters:

set_number (int) – The set number for which to retrieve the length.

Returns:

Length of the specified set.

Return type:

int

labfrog.shot_details.update_indiv_count_MODIFY_SHOT(current_set_number, set_length, campaign=None)[source]

Update the individual count when modifying a shot.

This route calculates and updates the shot numbers when modifying an existing shot based on the current set number and set length. It also backfeeds updates for subsequent sets.

Returns:

Total number of shots, number of shots at the current set number, and a list of shot numbers.

Return type:

tuple

labfrog.shot_details.get_shot_number_list_for_set(set_number)[source]

Retrieves a list of shot numbers for a specified set number.

Parameters:

set_number (int) – The set number for which to retrieve shot numbers.

Returns:

A list of shot numbers for the specified set number.

Return type:

List[int]

labfrog.shot_details.int_current_number_of_sets()[source]

Returns the current number of shot sets as an integer. Necessary for navigating database.

Returns:

The current number of shot sets.

Return type:

int

labfrog.shot_details.string_current_number_of_sets()[source]

Returns the current number of shot sets as a string. Necessary for updating the page via ajax request.

Returns:

The current number of shot sets.

Return type:

str

helper_functions

labfrog.helper_functions.normalize_choice_value(value: str | None) str | None[source]

Normalize request/session values used for campaign and layout selections.

labfrog.helper_functions.normalize_shot_day_scope(value) tuple[str | None, str | None, int | None][source]

Normalize a shot-day value into a stable UI/navigation token.

labfrog.helper_functions.shot_day_scope_for_doc(doc: dict | None) str | None[source]

Return the shot-day scope token for one document.

labfrog.helper_functions.annotate_search_rows_with_shot_day_helper(rows: list[dict] | None) list[dict][source]

Attach stable shot-day helper metadata to search-result rows.

labfrog.helper_functions.build_campaign_day_summary(shots_collection, query: dict, *, current_doc: dict | None = None) tuple[list[dict], dict | None][source]

Return shot-day rows plus the current entry’s resolved shot-day summary.

labfrog.helper_functions.shot_group_start_iso(block: Iterable[dict] | None) str | None[source]

Return the exact datetime token used to identify one shot-number restart block.

labfrog.helper_functions.build_shot_group_summary(blocks: Iterable[Iterable[dict]] | None, *, current_doc: dict | None = None) tuple[list[dict], str | None, str | None][source]

Return UI-friendly shot-group summaries keyed by numbering restarts.

labfrog.helper_functions.retrieve_data(user_input=None, query=None, projection=None)[source]

Retrieve data from MongoDB.

Either:
  • Pass user_inputs dict (form data) β†’ builds query and projection internally.

  • Or pass query and projection directly.

Returns:

For table display (NaN/None β†’ β€œβ€) df_columns (list[str]): Column order df_values_chart (list[dict]): For charts (NaN β†’ None) numeric_columns (list[str]): Columns with numeric values python_snippet (str) search_link (str)

Return type:

df_table (pd.DataFrame)

labfrog.helper_functions.create_new_version(doc_id, updated_fields, *, form_owned_fields=None)[source]

Create a new version of a document in the database.

Parameters:
  • doc_id (str) – The ID of the document to be versioned.

  • updated_fields (dict) – The fields to be updated in the new version.

  • form_owned_fields (set[str] | None) – Record keys the submitting form is authoritative for. When given, every other key on the archived document is carried into the new version, so values the form never showed – Kafka provenance, target-series grouping, fields outside the active layout – survive an edit instead of being dropped. Leave as None for a straight replacement of the previous version.

Returns:

True if the new version is successfully created, False otherwise.

Return type:

bool

labfrog.helper_functions.compare_field_diffs(documents, multi_value_fields=None)[source]

Return changed fields and changed multivalue options across a version chain.

labfrog.helper_functions.compare_fields(documents, multi_value_fields=None)[source]

Compare fields across multiple documents to identify differing fields.

Parameters:
  • documents (list) – List of documents to compare.

  • multi_value_fields (Iterable[str] | None) – Field names that should be compared as multi-value selections even if stored as scalars.

Returns:

List of field names that have differing values across the documents.

Return type:

list

labfrog.helper_functions.update_names_collection(collection_name)[source]

Update the names collection to reflect changes made in tab3 in tab1 and tab2.

Parameters:

collection_name (str) – The name of the collection to be updated.

entry_route_common

Shared helpers for add/edit entry routes.

labfrog.entry_route_common.set_session_mode_and_campaign(*, mode: str, campaign: str) None[source]
labfrog.entry_route_common.resolve_mode_and_campaign(*, default_campaign: str | None, posted_values: Mapping[str, str] | None = None) tuple[str, str][source]

Resolve mode/campaign from session and current POST payload.

labfrog.entry_route_common.coerce_datetime_field(raw_value)[source]
labfrog.entry_route_common.local_datetime_timezone_name() str[source]

Return the configured operator-facing timezone name.

labfrog.entry_route_common.add_utc_datetime_metadata(data: dict[str, Any], *, timezone_name: str | None = None) None[source]

Add UTC metadata while keeping date_time as local operator time.

LabFrog’s UI works in local time. For integration consumers, each saved record also carries the local timezone name and a UTC ISO timestamp.

labfrog.entry_route_common.prefill_form_from_current_doc(form, current_doc, *, parameters_db, options_collection, mode, visible_fields, custom_field_lookup, respect_copy_forward: bool = False)[source]
labfrog.entry_route_common.build_context_counters(*, mode: str, latest_doc: dict | None, current_doc: dict | None, has_entries: bool) dict[source]
labfrog.entry_route_common.submission_allowed_fields(*, visible_fields: list[str], always_include: list[str], diagnostic_names: set[str]) set[str][source]

Return every field name a submission is permitted to carry.

labfrog.entry_route_common.form_owned_field_names(form, *, visible_fields: list[str], always_include: list[str], diagnostic_names: set[str]) set[str][source]

Return the record keys this submission is authoritative for.

A field is owned when the bound form actually carries it, which is what makes an empty value meaningful: the operator cleared it. Keys outside this set were never on screen, so a save must leave them alone instead of dropping them from the record.

labfrog.entry_route_common.prepare_submission_payload_for_entry(form, *, mode: str, visible_fields: list[str], always_include: list[str], custom_field_lookup: dict[str, dict], diagnostic_names: set[str], shots_collection=None, campaign_choices: Mapping[str, Any] | None = None) tuple[dict[str, Any], dict[str, Any]][source]
labfrog.entry_route_common.build_entry_page_context(*, shots_collection, all_docs_query: dict, current_doc: dict | None, campaign: str) dict[str, Any][source]

Form Feedback (form_feedback)

Shared UI feedback helpers for form-based routes.

labfrog.form_feedback.flash_demo_limit_reached() None[source]

Show the shared demo-limit warning and discard stale form flashes.

labfrog.form_feedback.current_username(*, default: str = 'anonymous') str[source]

Return the active username for persistence/audit fields.

labfrog.form_feedback.flash_form_errors(form, *, logger: Logger | None = None) None[source]

Log and flash WTForms validation errors consistently.

Target Series (target_series)

Target-series (same-target multipart shot series) session helpers.

A target series groups multiple shots on the same physical target/sample. While a series is active, each newly saved shot automatically receives the shared series metadata plus an incrementing per-shot index.

Session state lives under the key TARGET_SERIES_SESSION_KEY and is a dict with the shape shown in _empty_series(), or None when no series is active.

All fields that are persisted into the shot document are prefixed target_series_ to avoid clashing with existing fields.

labfrog.target_series.get_active_series(session: dict) dict | None[source]

Return the active series state dict, or None if no series is open.

labfrog.target_series.start_series(session: dict, *, label: str = '', sample: str = '', planned_count: int | None = None, notes: str = '') dict[source]

Open a new target series and store it in the session.

Returns the new series state dict.

labfrog.target_series.finish_series(session: dict) None[source]

Close the active series so future shots are not assigned to it.

labfrog.target_series.clear_series(session: dict) None[source]

Remove the series state entirely from the session.

labfrog.target_series.build_series_fields_for_doc(session: dict) dict[str, Any][source]

Return the target_series_* fields to embed in a new shot document.

Returns an empty dict when no series is open or the series is closed/finished. Increments next_index in the session for the next shot.

labfrog.target_series.count_series_shots(shots_collection, series_id: str) int[source]

Return the number of active shots assigned to series_id.

labfrog.target_series.series_context_for_template(session: dict) dict[str, Any][source]

Return template context variables for the active series banner.

field_specs

Helpers for typed field definitions and compatibility-safe value storage.

labfrog.field_specs.normalize_custom_field_type(raw_type: str | None) str[source]

Map UI/form aliases to the canonical field-type names used internally.

labfrog.field_specs.custom_field_uses_options(raw_type: str | None) bool[source]

Return True when the field type stores a selectable options list.

labfrog.field_specs.validate_calculation_expression(expression: str | None, *, allowed_names: set[str] | None = None) tuple[bool, str | None][source]

Validate a user-provided equation for calculated custom fields.

labfrog.field_specs.evaluate_calculation_expression(expression: str | None, *, x: float, prev_x: float | None = None, allowed_names: set[str] | None = None) float[source]

Evaluate a validated equation with numeric source values.

labfrog.field_specs.normalize_calculated_decimal_places(value: Any) tuple[int | None, str | None][source]

Validate the optional decimal-place limit for calculated fields.

labfrog.field_specs.apply_calculated_decimal_places(value: float, decimal_places: int | None) float[source]

Round a calculated value when a decimal-place limit is configured.

labfrog.field_specs.build_field_label(field_doc: Mapping[str, Any] | None, field_name: str) str[source]

Return the label shown in forms, keeping units separate from the key.

labfrog.field_specs.field_value_type(*, custom_type: str | None = None, form_field=None, default: str = 'string') str[source]

Resolve the stored value type from field metadata or the WTForms field.

labfrog.field_specs.is_empty_value(value: Any) bool[source]

Return True when the value should be treated as empty for persistence.

labfrog.field_specs.serialize_value_for_storage(value: Any, value_type: str) Any[source]

Convert WTForms values into stable Mongo-friendly representations.

labfrog.field_specs.coerce_loaded_value(value: Any, value_type: str | None) Any[source]

Convert stored values back into the Python types expected by WTForms.

labfrog.field_specs.build_value_metadata(field_name: str, value: Any, *, field_doc: Mapping[str, Any] | None = None, form_field=None, specs: Mapping[str, Any] | None = None, allow_label_inference: bool = False) dict[str, Any] | None[source]

Build the structured metadata entry stored beside the legacy flat value.

labfrog.field_specs.infer_label_metadata(*, form_field=None, field_name: str = '') tuple[str, str][source]

Infer (display_name, unit) from a bound field label when possible.

labfrog.field_specs.get_record_value_metadata(record: Mapping[str, Any] | None) dict[str, Any][source]

Return the per-record metadata map if present, otherwise an empty dict.

labfrog.field_specs.value_for_form(record: Mapping[str, Any] | None, field_name: str, *, fallback_type: str | None = None) Any[source]

Return the best available field value for binding into a form.

labfrog.field_specs.flatten_record_for_form(record: Mapping[str, Any] | None, field_names: list[str], *, field_types: Mapping[str, str] | None = None) dict[str, Any][source]

Build the flat data dictionary expected by DynamicForm(data=…).

labfrog.field_specs.normalize_detail_key(value: str | None) str[source]

Convert free-text detail keys such as β€˜Beamline Name’ to β€˜beamline_name’.

labfrog.field_specs.clean_specs(specs: Mapping[str, Any] | None) dict[str, Any][source]

Drop empty values from nested specs while preserving stable key order.

labfrog.field_specs.build_diagnostic_specs(diag_name: str | None, preset_name: str | None) dict[str, Any][source]

Return a normalized snapshot of diagnostic metadata for one saved choice.

labfrog.field_specs.build_submission_payload(form, *, allowed_fields: set[str], custom_field_docs: Mapping[str, Mapping[str, Any]] | None = None, diagnostic_names: set[str] | None = None, always_include: set[str] | None = None, previous_record: Mapping[str, Any] | None = None) tuple[dict[str, Any], dict[str, Any]][source]

Extract typed values from a bound WTForms form for shot persistence.

helpers.choices

Selection and normalization helpers shared across route modules.

labfrog.helpers.choices.normalize_choice_value(value: str | None) str | None[source]

Normalize request/session values used for campaign and layout selections.

labfrog.helpers.choices.campaign_name_to_experiment_id(campaign: str | None) str | None[source]

Derive the canonical experiment_id from a MediaWiki campaign name.

The MediaWiki FWKTBeamtimeName (e.g. "Solenoid Beamline Tests 01.2025") stays the human-readable label everywhere - this only adds a machine-readable counterpart for cross-repository matching (see DAMNIT-web-hzdr’s architecture.md, β€œPilot Identity”). The transform is intentionally narrow (whitespace -> underscore) rather than a general slugify, so case and punctuation such as the campaign’s . separator are preserved exactly: "Solenoid Beamline Tests 01.2025" becomes "Solenoid_Beamline_Tests_01.2025".

Returns None for an unset/placeholder campaign (None, empty, or "NONE") so callers can omit the field rather than store a misleading derived id for β€œno campaign selected”.

labfrog.helpers.choices.campaign_identity_fields(campaign: str | None, campaign_choices: Mapping[str, Any] | None = None) dict[str, Any][source]

Return stable LabFrog export identity fields for a campaign selection.

labfrog.helpers.choices.ordered_unique_strings(values: Iterable[str | None]) list[str][source]

Return non-empty strings in first-seen order.

helpers.aliases

Field alias lookup with request- and app-level caching.

labfrog.helpers.aliases.invalidate_field_alias_cache() None[source]

Clear cached field aliases after alias rows are changed.

labfrog.helpers.aliases.get_field_alias_map(campaign=None) dict[str, str][source]

Return a dict of field_name -> alias for display use.

helpers.demo

Demo-mode helpers: user filtering, document counting, and limit enforcement.

labfrog.helpers.demo.demo_user_filter(query)[source]

Add a demo user filter to a Mongo query when DEMO_MODE is enabled.

labfrog.helpers.demo.demo_user_limit() int | None[source]
labfrog.helpers.demo.demo_user_doc_count(shots, username: str) int[source]
labfrog.helpers.demo.demo_limit_reached(shots, username: str) bool[source]

Return True if demo mode is enabled and the user is at/over the cap.

labfrog.helpers.demo.apply_demo_defaults(mode, parameters_collection=None)[source]

Apply demo-specific defaults for campaign/layout/header.

helpers.form_lookups

Session and DB lookups used while constructing runtime forms.

labfrog.helpers.form_lookups.diagnostic_catalog_docs() list[dict][source]

Return the cached diagnostic catalog (DisplayName + Tooltip).

Single reader for the diagnostics catalog so every path in one request sees the same snapshot. Previously _form_runtime_metadata() re-queried it while request_state served an app-cached copy, so two parts of the same form could disagree about the catalog for up to the cache TTL. Cache names match the ones db.clear_runtime_metadata_caches already drops.

labfrog.helpers.form_lookups.diagnostic_choice_docs(collection_name: str) list[dict][source]

Return one diagnostic’s preset choices, active first then OFF.

Cached per request and per app, so a form showing many diagnostics does not reissue two queries each. request_state and forms both read through here; cache names match what invalidate_diagnostic_choice_caches drops.

labfrog.helpers.form_lookups.custom_field_documents(collection=None) tuple[list[dict], dict[str, dict]][source]

Return the request-cached custom-field docs plus their field_name lookup.

Single reader for custom_fields_app, so one request never issues this same projection twice. request_state.load_custom_field_documents and _form_runtime_metadata() both route through here; the cache bucket is the one invalidate_custom_field_doc_cache already clears.

helpers.layout

Layout document resolution, CRUD, campaign mapping, and header-form helpers.

labfrog.helpers.layout.resolve_layout_selections(doc: dict | None, *, known_fields: set[str] | None = None) tuple[list[str], list[str], list[str]][source]

Return (parameters, diagnostics, combined) from a layout doc with fallback support.

labfrog.helpers.layout.build_layout_selection_update(selected_parameters: list[str] | None, selected_diagnostics: list[str] | None) dict[source]

Build a compatibility payload for layout selection fields.

labfrog.helpers.layout.paired_mode(mode: str) str[source]

Return the sibling mode used for paired layout saves.

labfrog.helpers.layout.campaign_is_unset(campaign: str | None) bool[source]

Return True only when no campaign is selected.

labfrog.helpers.layout.required_identity_fields_for_mode(mode: str, *, known_fields: set[str] | None = None, parameters_collection=None, include_set_shot_number_list: bool = True) list[str][source]

Return required identity fields for a mode from a single shared source.

labfrog.helpers.layout.build_mode_fallback_layout_doc(parameters_collection, *, source_doc: dict | None, target_mode: str) dict | None[source]

Build a synthetic mode-specific layout doc from a paired-mode source.

labfrog.helpers.layout.get_effective_campaign_layout_doc(parameters_collection, mode: str, campaign: str | None)[source]

Return the campaign layout doc, reusing a paired-mode layout when needed.

labfrog.helpers.layout.get_effective_layout_doc(parameters_collection, layout_name: str, mode: str, *, campaign: str | None = None)[source]

Return the requested mode layout doc, with paired-mode fallback support.

labfrog.helpers.layout.normalize_layout_selection_for_mode(parameters_collection, mode: str, selected_parameters: list[str] | None, selected_diagnostics: list[str] | None) tuple[list[str], list[str]][source]

Filter layout selections down to fields valid for the requested mode.

labfrog.helpers.layout.build_field_layout_doc(parameters_collection, *, layout_name: str, mode: str, description: str | None, responsible_person: str | None, campaigns: list[str] | None, selected_parameters: list[str] | None, selected_diagnostics: list[str] | None) dict[source]

Build a normalized field-layout document for the requested mode.

labfrog.helpers.layout.upsert_field_layout_doc(parameters_collection, *, layout_name: str, mode: str, description: str | None, responsible_person: str | None, campaigns: list[str] | None, selected_parameters: list[str] | None, selected_diagnostics: list[str] | None, merge_existing_campaigns: bool = False)[source]

Insert or update a mode-specific field layout document.

labfrog.helpers.layout.campaign_sort_key(value: str | None) tuple[int, int, int, int, str][source]

Sort dated campaign names by recency first, then fall back to text.

labfrog.helpers.layout.available_campaign_names(prioritized: Iterable[str | None] | None = None) list[str][source]

Return campaign names from session-facing sources, with optional priority.

labfrog.helpers.layout.split_layout_picker_fields(all_fields: list[str], selected_fields: list[str], diagnostic_names: Iterable[str]) dict[str, list[str]][source]

Split layout-picker fields into selected/non-selected parameter buckets.

labfrog.helpers.layout.field_layout_filter() dict[source]

Return a query fragment that matches Select Fields layouts only.

labfrog.helpers.layout.organize_layout_filter() dict[source]

Return a query fragment that matches Organize Form layouts only.

labfrog.helpers.layout.find_mode_layout_doc(parameters_collection, layout_name: str, mode: str)[source]

Return the mode-specific layout doc, falling back only to legacy mode-less docs.

labfrog.helpers.layout.build_layout_inventory(parameters_collection, mode: str | None = None) tuple[list[str], list[dict]][source]

Return layout names and non-default layout docs prepared for the picker UI.

labfrog.helpers.layout.get_campaign_layout_doc(parameters_collection, mode: str, campaign: str | None)[source]

Return the most recent field-layout document mapped to a campaign.

labfrog.helpers.layout.get_campaign_header_mapping_doc(mode: str, campaign: str | None)[source]

Return the explicit campaign-to-header mapping document, if configured.

labfrog.helpers.layout.upsert_campaign_header_mapping(mode: str, campaign: str | None, header_name: str | None, *, source_layout_name: str | None = None, updated_by: str | None = None) None[source]

Persist an explicit campaign-to-organized-form mapping.

labfrog.helpers.layout.get_campaign_header_doc(mode: str, campaign: str | None, *, source_layout_name: str | None = None)[source]

Return the most recent campaign override organize-form document.

labfrog.helpers.layout.get_source_layout_header_doc(mode: str, source_layout_name: str | None)[source]

Return the most recent default organize-form document for one source layout.

labfrog.helpers.layout.autofix_header_doc_source(doc: dict | None, source_layout_name: str | None, *, mode: str) dict | None[source]

Backfill source_layout_name onto legacy organize-form docs when inferred.

labfrog.helpers.layout.campaign_layout_missing(mode: str, campaign: str | None, parameters_collection) bool[source]

Return True when a non-empty campaign has no mapped field layout.

labfrog.helpers.layout.allow_default_for_campaign(mode: str, campaign: str) None[source]
labfrog.helpers.layout.default_allowed_for_campaign(mode: str, campaign: str | None) bool[source]
labfrog.helpers.layout.apply_campaign_layout(mode: str, campaign: str | None, parameters_collection, *, hydrate_diagnostics: bool = True)[source]

Select field layout based on campaign (falls back to DEFAULT).

helpers.navigation

Navigation and grouping helpers for shot-day and shot-block UIs.

labfrog.helpers.navigation.normalize_shot_day_scope(value) tuple[str | None, str | None, int | None][source]

Normalize a shot-day value into a stable UI/navigation token.

labfrog.helpers.navigation.shot_day_scope_for_doc(doc: dict | None) str | None[source]

Return the shot-day scope token for one document in isolation.

Uses the explicit shot_day field when present; otherwise falls back to the calendar date. This single-document view cannot know about neighboring shots, so it does NOT apply the gap-based session rule and should not be used to decide whether two shots belong together.

For multi-document grouping β€” including deciding whether a shot-day selector token matches a given document β€” use assign_session_scopes() (or the docs_for_day_scope() wrapper below), which is the single source of truth for the β€œshort gap should not start a new day” rule (see DEFAULT_SHOT_DAY_GAP_HOURS).

labfrog.helpers.navigation.assign_session_scopes(docs: list[dict], *, gap_hours: float = 8) list[str | None][source]

Return a per-doc scope token list using gap-based session detection.

Docs with an explicit shot_day field keep their explicit scope. For the remaining docs, consecutive shots separated by less than gap_hours are assigned to the same session scope. This prevents a short break around midnight from creating an accidental new shot day.

The returned list has the same length as docs and is in the same order.

labfrog.helpers.navigation.docs_for_day_scope(docs: list[dict], day_scope: str | None, *, gap_hours: float = 8) list[dict][source]

Return the subset of docs that belong to the given day-scope token.

This is the single place that resolves a shot-day selector token back to the documents that belong to it. Callers such as the shot-day selector, goto-shot/goto-set navigation, and search should all go through this helper instead of re-deriving membership so that β€œday 2” always means the same set of shots everywhere.

Three token shapes are supported:

  • shot_day:<token> / session:<token> β€” resolved via assign_session_scopes(), the gap-based rule (a short break β€” e.g. 30 minutes β€” never starts a new session on its own).

  • date:<ISO date> β€” a legacy plain-calendar-date scope (still emitted by some older links/bookmarks); matched directly against each doc’s calendar date regardless of gap, for backward compatibility.

labfrog.helpers.navigation.annotate_search_rows_with_shot_day_helper(rows: list[dict] | None, *, gap_hours: float = 8) list[dict][source]

Attach stable shot-day helper metadata to search-result rows.

labfrog.helpers.navigation.build_campaign_day_summary(shots_collection, query: dict, *, current_doc: dict | None = None, gap_hours: float = 8) tuple[list[dict], dict | None][source]

Return shot-day rows plus the current entry’s resolved shot-day summary.

labfrog.helpers.navigation.shot_group_start_iso(block: Iterable[dict] | None) str | None[source]

Return the datetime token used to identify one shot-number restart block.

labfrog.helpers.navigation.build_shot_group_summary(blocks: Iterable[Iterable[dict]] | None, *, current_doc: dict | None = None) tuple[list[dict], str | None, str | None][source]

Return UI-friendly shot-group summaries keyed by numbering restarts.

helpers.query

MongoDB query parsing, sanitization, and serialization helpers.

labfrog.helpers.query.parse_query(input_str)[source]

Parse and normalize a raw user query string.

Attempts to interpret the input as JSON first. If that fails, falls back to Python’s ast.literal_eval and then converts the result into JSON-compatible form.

Parameters:

input_str (str) – The raw query string provided by the user.

Returns:

A JSON-compatible dictionary parsed from the input.

Return type:

dict

Raises:

ValueError – If the input cannot be parsed into a valid query format.

labfrog.helpers.query.sanitize_query(query_dict)[source]

Sanitize a parsed query dictionary to block unsafe operators.

Recursively checks the query dictionary for forbidden MongoDB-like operators and raises an error if any are found.

Parameters:

query_dict (dict) – The parsed query dictionary to sanitize.

Returns:

The sanitized query dictionary (same object as input).

Return type:

dict

Raises:

ValueError – If the query contains forbidden operators such as $where, $accumulator, $function, $merge, or $out.

labfrog.helpers.query.normalize_search_datetime_query(query)[source]

Revive serialized date_time range bounds before MongoDB execution.

labfrog.helpers.query.serialize_query(query)[source]

Recursively convert a MongoDB query into a JSON-serializable form.

Converts complex Python objects (e.g., regex, datetime) into safe JSON representations so the query can be printed, copied, or embedded in URLs.

Parameters:

query (Any) – The MongoDB query or subquery to serialize. May be a dictionary, list, regex pattern, datetime, or basic type.

Returns:

A JSON-serializable object representing the input query.
  • dict or list are traversed recursively

  • regex patterns are converted into MongoDB $regex objects

  • datetime/date objects are converted to ISO-8601 strings

  • primitive types are returned unchanged

Return type:

Any

Examples

>>> import re, datetime
>>> serialize_query({"name": re.compile("^a", re.IGNORECASE)})
{'name': {'$regex': '^a', '$options': 'i'}}
>>> serialize_query({"created": datetime.date(2023, 1, 1)})
{'created': '2023-01-01'}
labfrog.helpers.query.query_to_json_string(query, indent=2)[source]

Convert a query to a JSON string suitable for embedding or sharing.

helpers.target

Target field normalization and JSON-safe serialization helpers.

labfrog.helpers.target.normalize_target_for_row(row: dict) dict[source]

Flatten nested target values into the display text used by tables and plots.

labfrog.helpers.target.to_json_safe(v)[source]

Recursively convert Python and MongoDB-specific objects into JSON-safe types.

helpers.urls

Safe return-url utilities for management workflows.

labfrog.helpers.urls.get_safe_return_url(candidate: str | None, default: str | None = None) str | None[source]

Return a same-app relative path suitable for β€œback” links.

labfrog.helpers.urls.get_management_return_url(candidate: str | None, default: str | None = None) str | None[source]

Resolve a safe workflow return URL for setup/design pages only.

forms

WTForms/Flask-WTF form classes for LabFrog data capture and search.

class labfrog.forms.InputForm(*args, **kwargs)[source]

Bases: FlaskForm

Base data-entry form whose selectables are bound at request time.

Parameters:
  • collection – Name of the collection containing layout configuration.

  • mode – Active shooting mode, either "shot" or "set".

  • visible_fields – Field names that should be available on the form.

  • *args – Positional arguments forwarded to FlaskForm.

  • **kwargs – Keyword arguments forwarded to FlaskForm.

date_time = <UnboundField(DateTimeField, ('Date and Time',), {'id': 'date_time', 'validators': [<wtforms.validators.DataRequired object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Modify here to update manually'}})>
update_date_time = <UnboundField(SubmitField, ('Update Date/Time',), {'id': 'update-date-time', 'render_kw': {'class': 'btn btn-primary', 'data-toggle': 'tooltip', 'title': 'Click when taking a shot'}})>
shot_number = <UnboundField(IntegerField, ('This Shot',), {'validators': [<wtforms.validators.Optional object>], 'default': 1, 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the shot number of the next shot to be recorded'}})>
set_number = <UnboundField(IntegerField, ('Set Number',), {'validators': [<wtforms.validators.Optional object>], 'default': 1, 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the set number of the next set to be recorded'}})>
set_length = <UnboundField(IntegerField, ('Set Length',), {'validators': [<wtforms.validators.Optional object>], 'default': 10, 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the number of shots in the upcoming set'}})>
shot_number_list = <UnboundField(FieldList, (<UnboundField(IntegerField, ('Shot Numbers',), {'render_kw': {'readonly': True}, 'validators': [<wtforms.validators.Optional object>]})>,), {'min_entries': 0})>
comments = <UnboundField(TextAreaField, ('Comments:',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Add any comments', 'style': 'height: 100px;'}})>
Campaign = <UnboundField(SelectField, ('Campaign',), {'choices': [('NONE', 'NONE')], 'validate_choice': False, 'render_kw': {'data-toggle': 'tooltip', 'title': 'Select a saved campaign.'}})>
gvd = <UnboundField(IntegerField, ('GVD (fs^2)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the GVD value in fs^2'}})>
laser_energy = <UnboundField(FloatField, ('Laser Energy (J)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the laser energy in J'}})>
plasma_mirror = <UnboundField(BooleanField, ('Plasma Mirror?',), {'render_kw': {'data-toggle': 'tooltip', 'title': 'Did you use a plasma mirror?'}})>
tw_intensity = <UnboundField(FloatField, ('TW Intensity',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the TW intensity'}})>
tod = <UnboundField(IntegerField, ('TOD (fs^3)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the TOD in fs^3'}})>
tw_delay = <UnboundField(FloatField, ('TW Delay',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the TW delay'}})>
tro = <UnboundField(FloatField, ('TRO (deg)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the TRO'}})>
tfo = <UnboundField(IntegerField, ('TFO (um)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter the TFO'}})>
target = <UnboundField(StringField, ('Target',), {'id': 'target', 'validators': [<wtforms.validators.Optional object>]})>
target_type = <UnboundField(SelectField, ('Target Type',), {'choices': [('', 'Select target type'), ('foil', 'Foil'), ('gas_jet', 'Gas jet'), ('cluster', 'Cluster source'), ('liquid', 'Liquid jet or sheet'), ('structured', 'Structured target'), ('other', 'Other')], 'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Canonical target class for a manually entered target'}})>
material = <UnboundField(StringField, ('Material',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Target material (for OTHER target)'}})>
thickness = <UnboundField(StringField, ('Thickness',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Target thickness (for OTHER target)'}})>
notes = <UnboundField(StringField, ('Notes',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Additional target notes (for OTHER target)'}})>
gas_species = <UnboundField(StringField, ('Gas Species',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Gas-jet or cluster species, for example Ar, N2, or He'}})>
gas_pressure = <UnboundField(FloatField, ('Gas Pressure (bar)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Gas backing pressure in bar', 'step': 'any'}})>
ramlon = <UnboundField(FloatField, ('Ramlon (Β΅Sv)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter Ramlon in Β΅Sv'}})>
proton_energy_tps45 = <UnboundField(FloatField, ('Proton Energy TPS 45 (MeV)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter Proton Energy TPS 45 in MeV'}})>
proton_energy_tps15 = <UnboundField(FloatField, ('Proton Energy TPS 15 (MeV)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter Proton Energy TPS 15 in MeV'}})>
time_of_flight = <UnboundField(FloatField, ('Time of flight (ns)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter time of flight in ns'}})>
proton_energy_from_tof = <UnboundField(FloatField, ('Proton Energy from TOF (MeV)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter Proton Energy from TOF in MeV'}})>
measured_gvd = <UnboundField(FloatField, ('Measured GVD (fs^2)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter measured GVD in fs^2'}})>
measured_tod = <UnboundField(FloatField, ('Measured TOD (fs^3)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'data-toggle': 'tooltip', 'title': 'Enter measured TOD in fs^3'}})>
submit = <UnboundField(SubmitField, ('Save and goto next shot',), {})>
class labfrog.forms.DynamicForm(*args, **kwargs)[source]

Bases: InputForm

Represents a dynamic form that extends the ShotForm. It initializes and updates fields dynamically based on DIAG_NAMES.

fields

A dictionary to store dynamically created fields.

initialize_fields()[source]

Initializes fields based on DIAG_NAMES.

update_fields()[source]

Updates the form fields.

initialize_fields()[source]

Compatibility wrapper that delegates diagnostic field binding.

update_fields()[source]
refresh_campaign_choices(options_collection)[source]
refresh_target_choices(options_collection)[source]
classmethod get_all_fields(collection, mode)[source]

Return all known field names (default layout + custom + diagnostics).

class labfrog.forms.FieldSelectionForm(*args, **kwargs)[source]

Bases: FlaskForm

layout_name = <UnboundField(StringField, ('Layout name',), {'validators': [<wtforms.validators.InputRequired object>]})>
description = <UnboundField(StringField, ('Description',), {'validators': [<wtforms.validators.InputRequired object>]})>
responsible_person = <UnboundField(StringField, ('Responsible person',), {'validators': [<wtforms.validators.Optional object>]})>
non_selected_fields = <UnboundField(SelectMultipleField, ('Non-Selected Fields',), {'validate_choice': False})>
selected_fields = <UnboundField(SelectMultipleField, ('Selected Fields',), {'validate_choice': False})>
class labfrog.forms.DetailsForm(*args, **kwargs)[source]

Bases: Form

Represents a form to capture key-value pair details.

key

A field to capture the key.

value

A field to capture the value.

validate()[source]

Validates the form data.

key = <UnboundField(StringField, ('Key',), {'validators': [<wtforms.validators.Optional object>]})>
value = <UnboundField(StringField, ('Value',), {'validators': [<wtforms.validators.Optional object>]})>
validate(extra_validators=None)[source]

Validate the form by calling validate on each field. Returns True if validation passes.

If the form defines a validate_<fieldname> method, it is appended as an extra validator for the field’s validate.

Parameters:

extra_validators – A dict mapping field names to lists of extra validator methods to run. Extra validators run after validators passed when creating the field. If the form has validate_<fieldname>, it is the last extra validator.

labfrog.forms.validate_details(form, field)[source]
class labfrog.forms.DiagnosticsForm(*args, **kwargs)[source]

Bases: FlaskForm

Capture diagnostic metadata and runtime-bound selectable options.

name

Human-readable diagnostic name.

description

Free-text description of the diagnostic.

diag_type

Selectable diagnostic type choices.

pc

Runtime-bound PC field sourced from MediaWiki when available.

name = <UnboundField(StringField, ('Name',), {'validators': [<wtforms.validators.DataRequired object>]})>
description = <UnboundField(StringField, ('Description',), {'validators': [<wtforms.validators.DataRequired object>]})>
set_up_date = <UnboundField(DateField, ('Set-up_date',), {'validators': [<wtforms.validators.Optional object>]})>
details = <UnboundField(FieldList, (<UnboundField(FormField, (<class 'labfrog.forms.DetailsForm'>,), {})>,), {'min_entries': 5, 'validators': [<function validate_details>]})>
responsible_person = <UnboundField(StringField, ('Responsible Person',), {'validators': [<wtforms.validators.Optional object>]})>
diag_type = <UnboundField(SelectField, ('Diagnostics Type',), {'choices': [], 'coerce': <class 'str'>, 'validators': [<wtforms.validators.DataRequired object>], 'validate_choice': False})>
file_path = <UnboundField(StringField, ('File path',), {'validators': [<wtforms.validators.Optional object>]})>
filename_schema = <UnboundField(SelectField, ('Filename_schema',), {'choices': ['ADD LATER'], 'validators': [<wtforms.validators.Optional object>]})>
counter_mode = <UnboundField(SelectField, ('Counter_mode',), {'choices': ['ADD LATER'], 'validators': [<wtforms.validators.Optional object>]})>
pc = <UnboundField(StringField, ('PC Name',), {'validators': [<wtforms.validators.Optional object>]})>
class labfrog.forms.ChoicesForm(*args, **kwargs)[source]

Bases: FlaskForm

Represents a form for adding choices to a field.

field_choice

A dropdown to select the field to which choices will be added.

choices

A field to input the new choice.

choice_submit

A button to submit the new choice.

field_choice = <UnboundField(SelectField, ('Field to Add Choices',), {'choices': [], 'coerce': <class 'str'>})>
choices = <UnboundField(StringField, ('Add a Choice',), {})>
choice_submit = <UnboundField(SubmitField, ('Add Choices',), {})>
class labfrog.forms.AddtlFields(*args, **kwargs)[source]

Bases: FlaskForm

Represents a form for capturing additional fields.

display_name = <UnboundField(StringField, ('Display Name',), {'validators': [<wtforms.validators.DataRequired object>]})>
tooltip = <UnboundField(TextAreaField, ('Tooltip',), {'validators': [<wtforms.validators.Optional object>]})>
details = <UnboundField(TextAreaField, ('Details',), {'validators': [<wtforms.validators.DataRequired object>]})>
class_field = <UnboundField(StringField, ('Class',), {'validators': [<wtforms.validators.Optional object>]})>
wiki_links = <UnboundField(StringField, ('Wiki-Link(s)',), {'validators': [<wtforms.validators.Optional object>]})>
team = <UnboundField(StringField, ('Team',), {'validators': [<wtforms.validators.Optional object>]})>
responsible = <UnboundField(StringField, ('Responsible',), {'validators': [<wtforms.validators.Optional object>]})>
valid_since = <UnboundField(DateField, ('Valid since',), {'validators': [<wtforms.validators.Optional object>]})>
valid_until = <UnboundField(DateField, ('Valid until',), {'validators': [<wtforms.validators.Optional object>]})>
counter_mode_options = <UnboundField(HiddenField, ('Counter mode options',), {'validators': [<wtforms.validators.Optional object>]})>
filename_schema_options = <UnboundField(HiddenField, ('Filename schema options',), {'validators': [<wtforms.validators.Optional object>]})>
class labfrog.forms.AddNewField(*args, **kwargs)[source]

Bases: FlaskForm

Represents a form for adding new custom fields to LabFrog application.

Inherits:

FlaskForm: Base class provided by Flask-WTF for creating web forms.

newFieldName

The name of the new field to be added.

details

Additional details or description about the new field.

section

The section of the application where the new field should be displayed.

custom_field_type

The data type of the new field, which can be β€˜string’, β€˜float’, or β€˜quality’.

newFieldName = <UnboundField(StringField, ('New field name',), {'validators': [<wtforms.validators.InputRequired object>]})>
display_name = <UnboundField(StringField, ('Display label',), {'validators': [<wtforms.validators.Optional object>]})>
unit = <UnboundField(StringField, ('Unit',), {'validators': [<wtforms.validators.Optional object>]})>
details = <UnboundField(StringField, ('Details',), {'validators': [<wtforms.validators.Optional object>]})>
options_text = <UnboundField(TextAreaField, ('Options (comma-separated, for select types)',), {'validators': [<wtforms.validators.Optional object>], 'render_kw': {'rows': 2}})>
calculated_from = <UnboundField(SelectField, ('Calculated source field',), {'choices': [], 'validators': [<wtforms.validators.Optional object>], 'validate_choice': False})>
calculated_formula = <UnboundField(StringField, ('Calculated equation (use x)',), {'validators': [<wtforms.validators.Optional object>]})>
calculated_decimal_places = <UnboundField(IntegerField, ('Decimal places',), {'validators': [<wtforms.validators.Optional object>]})>
section = <UnboundField(SelectField, ('Section',), {'choices': []})>
custom_field_type = <UnboundField(SelectField, ('Custom Field Type',), {'choices': [('string', 'String'), ('integer', 'Integer'), ('float', 'Float'), ('calculated', 'Calculated (formula)'), ('calculated_prev_shot', 'Calculated vs previous shot'), ('date', 'Date'), ('datetime', 'Date and Time'), ('boolean', 'Boolean'), ('SelectField', 'SelectField (dropdown)'), ('SelectMultipleField', 'SelectMultipleField (multi-select)')], 'default': 'string'})>
validators = <UnboundField(SelectField, ('Validation',), {'choices': [('Optional', 'Optional'), ('InputRequired', 'InputRequired')], 'default': 'Optional'})>
copy_forward = <UnboundField(BooleanField, ('Copy field value to next shot',), {'default': True})>
responsible_person = <UnboundField(StringField, ('Responsible person',), {'validators': [<wtforms.validators.Optional object>]})>
class labfrog.forms.SearchForm(*args, **kwargs)[source]

Bases: FlaskForm

Represents a form for searching shot-related data.

Campaign = <UnboundField(SelectField, ('Campaign',), {'choices': ['rendered via what is found in mongodb']})>
mode = <UnboundField(SelectField, ('Select shooting mode',), {'choices': ['shot', 'set', 'both'], 'default': 'shot', 'validate_choice': False})>
start_date = <UnboundField(DateField, ('Start Date',), {'format': '%Y-%m-%d', 'validators': [<wtforms.validators.Optional object>]})>
end_date = <UnboundField(DateField, ('End Date',), {'format': '%Y-%m-%d', 'validators': [<wtforms.validators.Optional object>]})>
user_query = <UnboundField(TextAreaField, ('Query',), {'validators': [<wtforms.validators.Optional object>]})>
latest_versions_only = <UnboundField(BooleanField, ('Only Latest Version',), {'default': True})>
return_all_fields = <UnboundField(BooleanField, ('Return All Fields',), {'default': True})>
what_to_return = <UnboundField(SelectMultipleField, ('Select Fields to Return',), {'choices': [], 'validators': [<wtforms.validators.Optional object>]})>

common

class labfrog.common.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, typing.Any] = <factory>, demo_target_dict: dict[str, typing.Any] = <factory>, 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)[source]

Bases: object

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]
demo_target_dict: dict[str, Any]
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
labfrog.common.external_services_disabled() bool[source]
labfrog.common.mediawiki_env_disabled() bool[source]
labfrog.common.request_cache_bucket(name: str) dict[source]
labfrog.common.mediawiki_cache_bucket() dict[source]
labfrog.common.sync_request_mediawiki_state_from_cache() None[source]
labfrog.common.refresh_mediawiki_cache(timeout=None) bool[source]
labfrog.common.parse_bool(value: Any) bool[source]
labfrog.common.parse_name_list(value)[source]
labfrog.common.empty_pc_choices()[source]
labfrog.common.empty_campaign_choices()[source]
labfrog.common.init_request_state()[source]
labfrog.common.read_config(file_path)[source]
labfrog.common.resolve_config_name(config: dict | None = None, environ=None) str[source]

Return the active instance config file name from config/env fallbacks.

labfrog.common.load_configuration_from_files()[source]
labfrog.common.build_settings(config: dict, default_config: dict) Settings[source]
labfrog.common.get_settings() Settings[source]
labfrog.common.set_settings(settings: Settings)[source]
labfrog.common.mediawiki_disabled(settings: Settings | None = None) bool[source]
labfrog.common.get_mediawiki_site()[source]
labfrog.common.create_mediawiki_site()[source]
labfrog.common.disable_mediawiki_runtime(reason)[source]
labfrog.common.run_with_timeout(func, *args, timeout)[source]
labfrog.common.try_create_mediawiki_site(timeout=3)[source]
labfrog.common.run_mediawiki_call(func, default, operation_name, timeout=None, site=None)[source]
labfrog.common.fetch_mediawiki_selectables(timeout=None)[source]
labfrog.common.retry_mediawiki_connection(timeout=None)[source]
labfrog.common.mediawiki_retry_available()[source]
labfrog.common.get_cached_mediawiki_campaigns() dict[source]
labfrog.common.get_cached_mediawiki_targets() dict[source]
labfrog.common.mediawiki_status_state() dict[source]
labfrog.common.initialize_common(app)[source]

mediawiki

labfrog.selectables.mediawiki.standardize_date(date_str)[source]

Convert known MediaWiki date formats to YYYY-MM-DD.

labfrog.selectables.mediawiki.retrieveTargets(site=None)[source]

Retrieves targets from IonenTarget cargo table

Returns:

A dictionary containing target information with names as keys

labfrog.selectables.mediawiki.retrievePCs(site=None)[source]

Retrieves PC’s from Computer cargo table

Returns:

A dictionary containing PC information with hostnames as keys

labfrog.selectables.mediawiki.retrieveCampaigns(site=None)[source]

Retrieves Campaigns from FWKTBeamtime cargo table

Returns:

A dictionary containing campaign information with names as keys

custom

labfrog.selectables.custom.read_custom_campaigns(file_path, CUSTOM_OPTIONS_COLLECTION, *, connection_settings=None)[source]
labfrog.selectables.custom.read_custom_campaigns_local(file_path)[source]

Read campaigns from file without writing to MongoDB.

labfrog.selectables.custom.read_custom_targets(file_path, CUSTOM_OPTIONS_COLLECTION, *, connection_settings=None)[source]
labfrog.selectables.custom.read_custom_targets_local(file_path)[source]

Read targets from file without writing to MongoDB.

Customization Builder (customize.diagnostics_maker)

labfrog.customize.diagnostics_maker.get_data_and_docs(config_name)[source]
labfrog.customize.diagnostics_maker.make_shot_and_set_defaults(collection, shot_document, set_document)[source]
labfrog.customize.diagnostics_maker.create_new_diagnostics(db, collection, data)[source]

Developer Tasks (dev_tasks)

Platform-aware helpers for documented uv run poe … workflows.

class labfrog.dev_tasks.DoctorCheck(name: 'str', status: 'str', detail: 'str', hint: 'str | None' = None)[source]

Bases: object

name: str
status: str
detail: str
hint: str | None = None
labfrog.dev_tasks.resolve_repo_config_name(repo_env: dict[str, str] | None = None, environ: dict[str, str] | None = None) str[source]

Match the app’s config-file selection rules for contributor checks.

labfrog.dev_tasks.collect_doctor_checks() list[DoctorCheck][source]
labfrog.dev_tasks.run_doctor() int[source]

Print a concise contributor-environment report.

labfrog.dev_tasks.resolve_server_backend(preferred: str = 'auto', *, system_name: str | None = None, gunicorn_available: bool | None = None) str[source]

Choose a stable production-style server backend for the current platform.

labfrog.dev_tasks.run_serve(*, host: str, port: int, workers: int, backend: str, preload: bool) int[source]

Run the production-style app server with a platform-aware backend.

labfrog.dev_tasks.restart_docker_service(*, system_name: str | None = None) int[source]

Restart Docker explicitly when the host needs a clean daemon reset.

labfrog.dev_tasks.ensure_docker_running(*, system_name: str | None = None) int[source]

Make Docker-backed tasks fail clearly instead of assuming one platform.

labfrog.dev_tasks.main(argv: list[str] | None = None) int[source]