Source code for labfrog.forms
# SPDX-FileCopyrightText: 2023-2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
"""WTForms/Flask-WTF form classes for LabFrog data capture and search."""
# TODO: if fail to connect to mediawiki OR running via docker, instead yield free-field text box options
import logging
import os
from flask import g, session
from flask_wtf import FlaskForm
from wtforms import ( # ValidationError,
BooleanField,
DateField,
DateTimeField,
FieldList,
FloatField,
Form,
FormField,
HiddenField,
IntegerField,
SelectField,
SelectMultipleField,
StringField,
SubmitField,
TextAreaField,
validators,
)
from wtforms.fields.core import UnboundField
from wtforms.validators import DataRequired, InputRequired, Optional
from labfrog.common import get_cached_mediawiki_targets, get_settings
from labfrog.db import get_db
from labfrog.field_specs import (
build_field_label,
normalize_custom_field_type,
)
from labfrog.form_factory import (
apply_field_aliases,
apply_parameter_options,
bind_runtime_field,
build_campaign_field,
build_pc_field,
build_target_field,
canonicalize_select_field_data,
)
from labfrog.helpers.form_lookups import (
_find_selected_header_doc,
_form_runtime_metadata,
_get_header_choice,
_header_session_key,
_set_header_choice,
diagnostic_catalog_docs,
diagnostic_choice_docs,
)
from labfrog.helpers.layout import (
_ensure_shot_identity_fields,
available_campaign_names,
campaign_is_unset,
organize_layout_filter,
paired_mode,
)
# DEFINE FORMS ####################################
[docs]
class InputForm(FlaskForm):
"""Base data-entry form whose selectables are bound at request time.
Args:
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``.
"""
def __init__(
self,
collection="custom_selections",
mode="shot",
visible_fields=None,
*args,
**kwargs,
):
initial_data = kwargs.get("data")
self._initial_dynamic_data = (
dict(initial_data) if isinstance(initial_data, dict) else {}
)
super().__init__(*args, **kwargs)
if mode == "shot":
mode = "shot"
else:
mode = "set"
if visible_fields is None:
visible_fields = []
else:
visible_fields = list(visible_fields)
self.visible_fields = visible_fields
InputForm.visible_fields = visible_fields
if mode == "shot":
for extra in ("shot_day", "shot_group"):
if extra not in visible_fields:
visible_fields.append(extra)
self.mode_field = SelectField(
"Choose Mode",
choices=["shot", "set"],
default=mode,
render_kw={"data-toggle": "tooltip", "title": "Select Shot or Set"},
)
# Bind mode field to this instance (added after FlaskForm init).
bound_mode = self.mode_field.bind(form=self, name="mode")
if not hasattr(bound_mode, "data"):
bound_mode.data = mode
self._fields["mode"] = bound_mode
self.mode = bound_mode
bind_runtime_field(self, "Campaign", build_campaign_field())
bind_runtime_field(self, "target", build_target_field())
db = get_db()
runtime_metadata = _form_runtime_metadata(collection, mode)
diagnostic_docs = [
doc for doc in runtime_metadata["diagnostics"] if doc.get("DisplayName")
]
custom_field_docs_list = list(runtime_metadata["custom_fields"])
default_doc = runtime_metadata["default_doc"]
sections_doc = default_doc
sections_collection = db[collection]
headers_choice = _get_header_choice(mode)
custom_sections = None
column_count = 6
if headers_choice != "DEFAULT":
custom_doc, custom_collection = _find_selected_header_doc(
collection, headers_choice, mode
)
if custom_doc and custom_doc.get("field_sections_dict"):
custom_sections = custom_doc.get("field_sections_dict")
column_count = custom_doc.get("column_count", column_count)
sections_doc = custom_doc
if custom_collection is not None:
sections_collection = custom_collection
else:
_set_header_choice(mode, "DEFAULT")
headers_choice = "DEFAULT"
# default column count
if headers_choice == "DEFAULT":
column_count = 3
# Use custom sections if available, otherwise default
def _normalize_sections(sections_dict):
"""Ensure values are lists, drop any special NOT_USED bucket."""
if not sections_dict:
return {}, {}
normalized = {}
colors = {}
# If dict values are dicts with fields/order, sort by order
items = []
for sec, val in sections_dict.items():
if isinstance(val, dict):
items.append((
sec,
val.get("order", 0),
val.get("fields", []),
val.get("color"),
))
else:
items.append((sec, 0, val, None))
items.sort(key=lambda x: x[1])
for sec, _order, fields, color in items:
if sec == "NOT_USED":
continue
normalized[sec] = fields
if color:
colors[sec] = color
return normalized, colors
sections_source = (
custom_sections
if custom_sections is not None
else (default_doc.get("field_sections_dict") if default_doc else {})
)
not_used_fields = []
if sections_source and "NOT_USED" in sections_source:
not_used_val = sections_source.get("NOT_USED")
if isinstance(not_used_val, dict):
not_used_fields = list(not_used_val.get("fields", []) or [])
else:
not_used_fields = list(not_used_val or [])
normalized_sections, section_colors = _normalize_sections(sections_source)
always_include = default_doc.get("always_include") if default_doc else []
if "Campaign" not in (always_include or []):
always_include = list(always_include or [])
always_include.append("Campaign")
# Determine all field names this form can safely render.
known_renderable_fields = set()
for klass in type(self).mro():
for attr_name, value in vars(klass).items():
if isinstance(value, UnboundField):
known_renderable_fields.add(attr_name)
known_renderable_fields.update({"mode", "shot_day", "shot_group"})
# Pull diagnostics from DB so we do not depend on DIAG_NAMES being preloaded.
for diag_doc in diagnostic_docs:
diag_name = diag_doc.get("DisplayName")
if isinstance(diag_name, str) and diag_name:
known_renderable_fields.add(diag_name)
known_renderable_fields.update(
name for name in g.get("DIAG_NAMES", {}).keys() if name
)
for field_doc in custom_field_docs_list:
name = field_doc.get("field_name")
if isinstance(name, str) and name:
known_renderable_fields.add(name)
known_renderable_fields.update(
name for name in visible_fields if isinstance(name, str) and name
)
# Only keep fields that are selected or always included.
allowed_fields = set(visible_fields) | set(always_include or [])
allowed_fields.update({"shot_day", "shot_group"})
filtered_sections = {}
removed_unknown_fields = set()
for section_name, fields in normalized_sections.items():
kept = []
for field_name in fields or []:
if (
field_name in allowed_fields
and field_name in known_renderable_fields
):
kept.append(field_name)
elif field_name not in known_renderable_fields:
removed_unknown_fields.add(field_name)
if kept:
filtered_sections[section_name] = kept
# Campaign is controlled from the toolbar above the form, not inside sections.
for section_name, fields in list(filtered_sections.items()):
kept = [field_name for field_name in fields if field_name != "Campaign"]
if kept:
filtered_sections[section_name] = kept
else:
filtered_sections.pop(section_name, None)
# Diagnostics are rendered in their own card beneath the main sections.
diagnostic_field_names = {
doc.get("DisplayName")
for doc in diagnostic_docs
if isinstance(doc.get("DisplayName"), str) and doc.get("DisplayName")
}
for section_name, fields in list(filtered_sections.items()):
kept = [
field_name
for field_name in fields
if field_name not in diagnostic_field_names
]
if kept:
filtered_sections[section_name] = kept
else:
filtered_sections.pop(section_name, None)
default_section_map = {}
custom_section_map = {}
calculated_sources: dict[str, str] = {}
default_sections = (default_doc or {}).get("field_sections_dict") or {}
for section_name, value in default_sections.items():
fields = (
value.get("fields", []) if isinstance(value, dict) else (value or [])
)
for field_name in fields:
if (
isinstance(field_name, str)
and field_name
and field_name not in default_section_map
):
default_section_map[field_name] = section_name
for field_doc in custom_field_docs_list:
field_name = field_doc.get("field_name")
section_name = field_doc.get("section")
if (
isinstance(field_name, str)
and field_name
and isinstance(section_name, str)
and section_name
):
custom_section_map.setdefault(field_name, section_name)
default_section_map.setdefault(field_name, section_name)
normalized_type = normalize_custom_field_type(
field_doc.get("custom_field_type")
)
if normalized_type in {"calculated", "calculated_prev_shot"}:
source_field = str(field_doc.get("calculated_from") or "").strip()
if source_field and source_field != field_name:
calculated_sources[field_name] = source_field
source_section = custom_section_map.get(
source_field
) or default_section_map.get(source_field)
if source_section:
custom_section_map[field_name] = source_section
default_section_map.setdefault(field_name, source_section)
placed_fields = {
field_name
for fields in filtered_sections.values()
for field_name in (fields or [])
}
hidden_not_used = set(not_used_fields or [])
shot_identity_fields = set(
_ensure_shot_identity_fields(
[field_name for field_name in visible_fields if field_name],
known_fields=known_renderable_fields,
)
)
for field_name in visible_fields:
if (
field_name in placed_fields
or field_name == "Campaign"
or field_name in diagnostic_field_names
or (
field_name in hidden_not_used
and field_name not in shot_identity_fields
)
or field_name not in known_renderable_fields
):
continue
target_section = (
custom_section_map.get(field_name)
or default_section_map.get(field_name)
or "Shot Details"
)
filtered_sections.setdefault(target_section, [])
filtered_sections[target_section].append(field_name)
placed_fields.add(field_name)
for section_name, fields in list(filtered_sections.items()):
filtered_sections[section_name] = _ensure_shot_identity_fields(
fields,
known_fields=known_renderable_fields,
)
# Keep calculated fields attached to their source field in the same section.
for calculated_field, source_field in calculated_sources.items():
source_section = None
source_index = None
for section_name, fields in filtered_sections.items():
if source_field in (fields or []):
source_section = section_name
source_index = fields.index(source_field)
break
if source_section is None:
continue
current_section = None
for section_name, fields in filtered_sections.items():
if calculated_field in (fields or []):
current_section = section_name
break
if current_section is not None:
filtered_sections[current_section] = [
name
for name in (filtered_sections.get(current_section) or [])
if name != calculated_field
]
if not filtered_sections[current_section]:
filtered_sections.pop(current_section, None)
section_fields = filtered_sections.setdefault(source_section, [])
insert_at = min(source_index + 1, len(section_fields))
section_fields.insert(insert_at, calculated_field)
if (
removed_unknown_fields
and sections_doc
and sections_doc.get("_id")
and not os.environ.get("LABFROG_READ_ONLY_DB")
):
repaired_sections = {}
for idx, (section_name, fields) in enumerate(filtered_sections.items()):
repaired_sections[section_name] = {
"fields": fields,
"order": idx,
"color": section_colors.get(section_name),
}
selected_fields = [
f
for f in (sections_doc.get("selected_fields") or [])
if isinstance(f, str) and f in known_renderable_fields
]
sections_collection.update_one(
{"_id": sections_doc["_id"]},
{
"$set": {
"field_sections_dict": repaired_sections,
"selected_fields": selected_fields,
}
},
)
logging.warning(
"Removed invalid fields from %s document %s: %s",
sections_collection.name,
sections_doc.get("_id"),
", ".join(sorted(removed_unknown_fields)),
)
self.FIELD_SECTIONS = filtered_sections
self.SECTION_COLORS = section_colors
self.COLUMN_COUNT = column_count
self.NOT_USED_FIELDS = [f for f in not_used_fields if f]
shot_anchor_fields = (
"shot_number",
"shot_number_list",
"set_number",
"set_length",
"update_date_time",
"date_time",
)
def _normalize_field_name(value):
return "_".join(
str(value or "").replace("\u00a0", " ").strip().lower().split()
)
shot_anchor_keys = {_normalize_field_name(f) for f in shot_anchor_fields}
shot_section_name = None
for section_name, fields in self.FIELD_SECTIONS.items():
normalized_fields = {_normalize_field_name(f) for f in (fields or [])}
if shot_anchor_keys.intersection(normalized_fields):
shot_section_name = section_name
break
if not shot_section_name:
shot_section_name = "Shot Details"
self.FIELD_SECTIONS.setdefault(shot_section_name, [])
shot_fields = list(self.FIELD_SECTIONS.get(shot_section_name, []))
if mode == "shot":
for extra in ("shot_day", "shot_group"):
if extra not in shot_fields:
shot_fields.append(extra)
if shot_fields:
self.FIELD_SECTIONS[shot_section_name] = shot_fields
self.SHOT_DETAILS_SECTION = shot_section_name
def _bind_and_set_field(name, unbound_field):
bind_runtime_field(self, name, unbound_field)
if visible_fields:
core_fields = {
"Campaign",
"target",
"mode",
"material",
"thickness",
"notes",
}
for _section, fields in self.FIELD_SECTIONS.items():
for field_name in fields:
# If the field is not in visible_fields or always_include, remove it
if (
field_name not in self.visible_fields
and field_name not in always_include
):
if field_name in core_fields:
continue
if hasattr(self, field_name):
delattr(
self, field_name
) # Remove the attribute if it exists
else:
# If the field is not present as an attribute, create a new string field
if not hasattr(self, field_name):
_bind_and_set_field(
field_name,
StringField(
field_name, id=field_name, validators=[Optional()]
),
)
# Create a dictionary to store custom field types
custom_field_types = {}
custom_field_docs = {}
validator_values = {}
get_all_details = {}
custom_options_text = {}
for field in custom_field_docs_list:
field_name = field.get("field_name")
custom_field_type = field.get("custom_field_type")
details = field.get("details")
options_text = field.get("options_text", "")
if field_name and custom_field_type:
custom_field_types[field_name] = custom_field_type
if field_name:
custom_field_docs[field_name] = field
validator_value = field.get("validators")
if field_name and validator_value:
validator_values[field_name] = validator_value
if field_name and details:
get_all_details[field_name] = details
if field_name:
custom_options_text[field_name] = options_text or details or ""
for field_name, field_type in custom_field_types.items():
this_validator = [Optional()]
details = ""
if field_name in visible_fields:
validators = validator_values.get(field_name, "Optional")
# print(field_name, field_type, validators)
if "InputRequired" in validators:
this_validator = [InputRequired()]
# print(field_name, field_type, this_validator)
details = str(get_all_details.get(field_name, ""))
normalized_type = normalize_custom_field_type(field_type)
label_text = build_field_label(
custom_field_docs.get(field_name), field_name
)
if "string" in normalized_type:
# print("field type string")
_bind_and_set_field(
field_name,
StringField(
label_text,
validators=this_validator,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
),
)
elif "float" in normalized_type:
# print("field type float")
_bind_and_set_field(
field_name,
FloatField(
label_text,
validators=this_validator,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
),
)
elif normalized_type in {"calculated", "calculated_prev_shot"}:
calc_source = str(
custom_field_docs.get(field_name, {}).get("calculated_from")
or ""
).strip()
calc_formula = str(
custom_field_docs.get(field_name, {}).get("calculated_formula")
or ""
).strip()
calc_decimal_places = custom_field_docs.get(field_name, {}).get(
"calculated_decimal_places"
)
_bind_and_set_field(
field_name,
FloatField(
label_text,
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": details,
"readonly": True,
"data-calculated-source": calc_source,
"data-calculated-formula": calc_formula,
"data-calculated-type": normalized_type,
"data-calculated-decimal-places": (
""
if calc_decimal_places is None
else calc_decimal_places
),
},
),
)
elif "integer" in normalized_type:
_bind_and_set_field(
field_name,
IntegerField(
label_text,
validators=this_validator,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
),
)
elif normalized_type == "date":
_bind_and_set_field(
field_name,
DateField(
label_text,
validators=this_validator,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
),
)
elif normalized_type == "datetime":
_bind_and_set_field(
field_name,
DateTimeField(
label_text,
validators=this_validator,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
),
)
elif normalized_type == "boolean":
_bind_and_set_field(
field_name,
BooleanField(
label_text,
validators=this_validator,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
),
)
elif "quality_select" in normalized_type:
_bind_and_set_field(
field_name,
SelectField(
label_text,
choices=[
"Very Good",
"Good",
"OK",
"Bad",
"Very Bad",
"Not specified",
],
render_kw={
"data-toggle": "tooltip",
"title": details,
},
default="Not specified",
),
)
elif "select_custom" in normalized_type:
options_raw = custom_options_text.get(field_name, details or "")
choices_raw = [
opt.strip() for opt in options_raw.split(",") if opt.strip()
] or ["Option 1", "Option 2"]
choices_list = [(opt, opt) for opt in choices_raw]
_bind_and_set_field(
field_name,
SelectField(
label_text,
choices=choices_list,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
validate_choice=False,
),
)
elif "select_multi_custom" in normalized_type:
options_raw = custom_options_text.get(field_name, details or "")
choices_raw = [
opt.strip() for opt in options_raw.split(",") if opt.strip()
] or ["Option 1", "Option 2"]
choices_list = [(opt, opt) for opt in choices_raw]
_bind_and_set_field(
field_name,
SelectMultipleField(
label_text,
choices=choices_list,
render_kw={
"data-toggle": "tooltip",
"title": details,
},
validate_choice=False,
),
)
"""
for field_name, field_type in custom_field_types.items():
if field_name in visible_fields and field_type == "float":
validators = validator_values.get(field_name, "Optional")
if "Optional" not in validators:
if not (isinstance(field_name.data, "float")):
self.field_name.errors += (ValidationError("Not a float"),)
"""
apply_field_aliases(self)
# MAIN DEFAULT NON-CUSTOM FIELDS INIT
# mode done via init
# mode = SelectField('Choose Mode', choices = ['shot', 'set'], default = InputForm.mode,
# .. render_kw={'data-toggle': 'tooltip','title': 'Select Shot or Set'})
date_time = DateTimeField(
"Date and Time",
id="date_time",
validators=[DataRequired()],
render_kw={
"data-toggle": "tooltip",
"title": "Modify here to update manually",
},
)
update_date_time = 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 = IntegerField(
"This Shot",
validators=[Optional()],
default=1,
render_kw={
"data-toggle": "tooltip",
"title": "Enter the shot number of the next shot to be recorded",
},
)
set_number = IntegerField(
"Set Number",
validators=[Optional()],
default=1,
render_kw={
"data-toggle": "tooltip",
"title": "Enter the set number of the next set to be recorded",
},
)
set_length = IntegerField(
"Set Length",
validators=[Optional()],
default=10,
render_kw={
"data-toggle": "tooltip",
"title": "Enter the number of shots in the upcoming set",
},
)
shot_number_list = FieldList(
IntegerField(
"Shot Numbers",
render_kw={"readonly": True},
validators=[Optional()],
),
min_entries=0,
)
comments = TextAreaField(
"Comments:",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Add any comments",
"style": "height: 100px;",
},
)
Campaign = SelectField(
"Campaign",
choices=[("NONE", "NONE")],
validate_choice=False,
render_kw={
"data-toggle": "tooltip",
"title": "Select a saved campaign.",
},
)
# Laser parameters
gvd = IntegerField(
"GVD (fs^2)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter the GVD value in fs^2",
},
)
laser_energy = FloatField(
"Laser Energy (J)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter the laser energy in J",
},
)
plasma_mirror = BooleanField(
"Plasma Mirror?",
render_kw={
"data-toggle": "tooltip",
"title": "Did you use a plasma mirror?",
},
)
tw_intensity = FloatField(
"TW Intensity",
validators=[Optional()],
render_kw={"data-toggle": "tooltip", "title": "Enter the TW intensity"},
)
tod = IntegerField(
"TOD (fs^3)",
validators=[Optional()],
render_kw={"data-toggle": "tooltip", "title": "Enter the TOD in fs^3"},
)
tw_delay = FloatField(
"TW Delay",
validators=[Optional()],
render_kw={"data-toggle": "tooltip", "title": "Enter the TW delay"},
)
# Interaction parameters
tro = FloatField(
"TRO (deg)",
validators=[Optional()],
render_kw={"data-toggle": "tooltip", "title": "Enter the TRO"},
)
tfo = IntegerField(
"TFO (um)",
validators=[Optional()],
render_kw={"data-toggle": "tooltip", "title": "Enter the TFO"},
)
target = StringField("Target", id="target", validators=[Optional()])
target_type = 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=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Canonical target class for a manually entered target",
},
)
material = StringField(
"Material",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Target material (for OTHER target)",
},
)
thickness = StringField(
"Thickness",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Target thickness (for OTHER target)",
},
)
notes = StringField(
"Notes",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Additional target notes (for OTHER target)",
},
)
gas_species = StringField(
"Gas Species",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Gas-jet or cluster species, for example Ar, N2, or He",
},
)
gas_pressure = FloatField(
"Gas Pressure (bar)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Gas backing pressure in bar",
"step": "any",
},
)
# Post-shot readout
ramlon = FloatField(
"Ramlon (µSv)",
validators=[Optional()],
render_kw={"data-toggle": "tooltip", "title": "Enter Ramlon in µSv"},
)
proton_energy_tps45 = FloatField(
"Proton Energy TPS 45 (MeV)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter Proton Energy TPS 45 in MeV",
},
)
proton_energy_tps15 = FloatField(
"Proton Energy TPS 15 (MeV)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter Proton Energy TPS 15 in MeV",
},
)
time_of_flight = FloatField(
"Time of flight (ns)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter time of flight in ns",
},
)
proton_energy_from_tof = FloatField(
"Proton Energy from TOF (MeV)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter Proton Energy from TOF in MeV",
},
)
measured_gvd = FloatField(
"Measured GVD (fs^2)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter measured GVD in fs^2",
},
)
measured_tod = FloatField(
"Measured TOD (fs^3)",
validators=[Optional()],
render_kw={
"data-toggle": "tooltip",
"title": "Enter measured TOD in fs^3",
},
)
submit = SubmitField("Save and goto next shot")
[docs]
class DynamicForm(InputForm):
"""
Represents a dynamic form that extends the ShotForm.
It initializes and updates fields dynamically based on DIAG_NAMES.
Attributes:
fields: A dictionary to store dynamically created fields.
Methods:
initialize_fields: Initializes fields based on DIAG_NAMES.
update_fields: Updates the form fields.
"""
def __init__(self, options_collection=None, *args, **kwargs):
super().__init__(*args, **kwargs)
self.fields = {}
self._diagnostic_fields = set()
self.update_fields()
self.refresh_campaign_choices(options_collection)
self.refresh_target_choices(options_collection)
apply_parameter_options(self)
apply_field_aliases(self)
canonicalize_select_field_data(self)
[docs]
def initialize_fields(self):
"""Compatibility wrapper that delegates diagnostic field binding."""
self.update_fields()
[docs]
def update_fields(self):
visible_field_names = set(getattr(self, "visible_fields", []) or [])
previous_diagnostic_fields = set(
getattr(self, "_diagnostic_fields", set()) or []
)
if visible_field_names:
diag_cache = g.get("DIAG_NAMES", {}) or {}
tooltip_cache = g.get("TOOLTIPS", {}) or {}
# Only real diagnostics can be missing here. Without this filter every
# ordinary field (shot_number, Campaign, ...) was sent to the catalog
# query below, which then matched nothing -- one wasted round trip per
# request. The catalog itself is request/app cached.
known_diagnostic_names = {
doc.get("DisplayName"): doc.get("Tooltip", "")
for doc in diagnostic_catalog_docs()
if doc.get("DisplayName")
}
missing_visible_diagnostics = [
name
for name in visible_field_names
if name in known_diagnostic_names and name not in diag_cache
]
if missing_visible_diagnostics:
for diagnostic_name in missing_visible_diagnostics:
diag_cache[diagnostic_name] = [
doc.get("name")
for doc in diagnostic_choice_docs(diagnostic_name)
if doc.get("name")
]
tooltip_cache[diagnostic_name] = known_diagnostic_names[
diagnostic_name
]
g.DIAG_NAMES = diag_cache
g.TOOLTIPS = tooltip_cache
if visible_field_names:
diagnostic_field_names = {
name
for name in g.get("DIAG_NAMES", {}).keys()
if name in visible_field_names
}
else:
diagnostic_field_names = {
name for name in g.get("DIAG_NAMES", {}).keys() if name
}
for name in diagnostic_field_names:
choices = g.get("DIAG_NAMES", {}).get(name) or []
field_choices = [(n, n) for n in reversed(choices)]
tooltip = g.get("TOOLTIPS", {}).get(name, "")
field = SelectField(
name,
choices=field_choices,
render_kw={"data-toggle": "tooltip", "title": tooltip},
validators=[Optional()],
validate_choice=False,
)
bind_runtime_field(self, name, field)
for field_name in previous_diagnostic_fields - diagnostic_field_names:
self._fields.pop(field_name, None)
if hasattr(self, field_name):
delattr(self, field_name)
self._diagnostic_fields = diagnostic_field_names
[docs]
def refresh_campaign_choices(self, options_collection):
if not hasattr(self, "Campaign"):
return
settings = get_settings()
if settings.use_wiki or not settings.custom_options:
prioritized = [
getattr(self.Campaign, "data", None),
session.get("selected_campaign"),
]
self.Campaign.choices = [("NONE", "NONE")] + [
(name, name)
for name in available_campaign_names(prioritized)
if name != "NONE"
]
elif options_collection is not None:
db = get_db()
campaigns_list = list(
db[str(options_collection)].find(
{"option": "campaigns"}, {"campaigns_dict": 1, "_id": 0}
)
)
merged_campaign_dict = {}
for campaign in campaigns_list:
merged = campaign.get("campaigns_dict", {})
if merged:
merged_campaign_dict.update(merged)
g.campaign_dict = merged_campaign_dict
[docs]
def refresh_target_choices(self, options_collection):
if not hasattr(self, "target"):
return
settings = get_settings()
if settings.use_wiki:
targets = get_cached_mediawiki_targets()
self.target.choices = [(key, key) for key in targets.keys()]
elif settings.custom_options and options_collection is not None:
db = get_db()
targets_list = list(
db[str(options_collection)].find(
{"option": "targets"}, {"targets_dict": 1, "_id": 0}
)
)
g.target_dict = {}
for target in targets_list:
merged = target.get("targets_dict", {})
if merged:
g.target_dict.update(merged)
selected_campaign = session.get("selected_campaign")
target_tags = {}
for name, details in g.target_dict.items():
tags = str(details.get("Campaign", "")).strip()
tag_list = [
t.strip() for t in tags.replace(";", ",").split(",") if t.strip()
]
target_tags[name] = tag_list
has_selected_tags = False
if not campaign_is_unset(selected_campaign):
has_selected_tags = any(
selected_campaign in tags for tags in target_tags.values()
)
filtered_choices = []
for name, tags in target_tags.items():
if campaign_is_unset(selected_campaign):
filtered_choices.append((name, name))
continue
if has_selected_tags:
if selected_campaign in tags:
filtered_choices.append((name, name))
else:
if not tags:
filtered_choices.append((name, name))
if filtered_choices:
self.target.choices = filtered_choices
else:
self.target.choices = [(key, key) for key in g.target_dict.keys()]
# USED FOR SELECTION FORM ONLY, don't show always include?
[docs]
@classmethod
def get_all_fields(cls, collection, mode):
"""Return all known field names (default layout + custom + diagnostics)."""
all_fields = []
db = get_db()
custom_fields = list(
db["custom_fields_app"].find(
{"field_name": {"$exists": True}},
{
"field_name": 1,
"custom_field_type": 1,
"options_text": 1,
"details": 1,
"_id": 0,
},
)
)
default_fields = db[str(collection)].find_one({
"layout_name": "DEFAULT",
"mode": mode,
})
if default_fields and default_fields.get("field_sections_dict"):
for section_value in default_fields.get("field_sections_dict").values():
if isinstance(section_value, dict):
section_fields = section_value.get("fields", []) or []
else:
section_fields = section_value or []
for field_name in section_fields:
if field_name and field_name not in all_fields:
all_fields.append(field_name)
else:
logging.warning(
"No default_fields found for collection=%s mode=%s", collection, mode
)
# print("custom_fields", custom_fields)
if custom_fields:
custom_fields_names = [field["field_name"] for field in custom_fields]
for field in custom_fields_names:
all_fields.append(field)
diagnostics_collections_names = list(
db["diagnostics"].find({}, {"DisplayName": 1, "_id": 0})
)
choices_within_diagnostic = list()
DIAG_NAMES = g.get("DIAG_NAMES", {}) or {}
for diagnostic_doc in diagnostics_collections_names:
diagnostic_name = diagnostic_doc["DisplayName"]
# print("diagnostic_name", diagnostic_name)
choices_within_diagnostic = list(
db["diagnostics"][diagnostic_name].find({}, {"name": 1, "_id": 0})
)
DIAG_NAMES[diagnostic_name] = [
choice["name"] for choice in choices_within_diagnostic
]
all_fields.append(diagnostic_name)
return all_fields
# NEW FIELD SELECTION FORM
[docs]
class FieldSelectionForm(FlaskForm):
layout_name = StringField("Layout name", validators=[InputRequired()])
description = StringField("Description", validators=[InputRequired()])
responsible_person = StringField("Responsible person", validators=[Optional()])
non_selected_fields = SelectMultipleField(
"Non-Selected Fields", validate_choice=False
)
selected_fields = SelectMultipleField("Selected Fields", validate_choice=False)
def __init__(self, *args, **kwargs):
all_fields = kwargs.pop(
"all_fields", []
) # all_fields includes custom fields + all diagnostics + default fields
selected_fields = kwargs.pop("selected_fields", [])
super().__init__(*args, **kwargs)
self.non_selected_fields.choices = [
(field, field) for field in all_fields if field not in selected_fields
]
self.selected_fields.choices = [(field, field) for field in selected_fields]
[docs]
class DetailsForm(Form):
"""
Represents a form to capture key-value pair details.
Attributes:
key: A field to capture the key.
value: A field to capture the value.
Methods:
validate: Validates the form data.
"""
key = StringField("Key", validators=[validators.Optional()])
value = StringField("Value", validators=[validators.Optional()])
[docs]
def validate(self, extra_validators=None):
if not super().validate(extra_validators=extra_validators):
return False
key = self.key.data.strip() if self.key.data else ""
value = self.value.data.strip() if self.value.data else ""
if not key and not value:
self.key.data = ""
self.value.data = ""
return True
[docs]
def validate_details(form, field):
keys_filled = any(
entry.key.data.strip() if entry.key.data else False for entry in field.entries
)
values_filled = any(
entry.value.data.strip() if entry.value.data else False
for entry in field.entries
)
if keys_filled or values_filled:
for entry in field.entries:
key = entry.key.data.strip() if entry.key.data else ""
value = entry.value.data.strip() if entry.value.data else ""
if not key and not value:
entry.key.data = ""
entry.value.data = ""
if key and not value:
entry.value.data = ""
if value and not key:
entry.key.data = ""
[docs]
class DiagnosticsForm(FlaskForm):
"""
Capture diagnostic metadata and runtime-bound selectable options.
Attributes:
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 = StringField("Name", validators=[validators.DataRequired()])
description = StringField("Description", validators=[validators.DataRequired()])
set_up_date = DateField("Set-up_date", validators=[Optional()])
details = FieldList(
FormField(DetailsForm), min_entries=5, validators=[validate_details]
)
responsible_person = StringField(
"Responsible Person", validators=[validators.Optional()]
)
diag_type = SelectField(
"Diagnostics Type",
choices=[], # This is filled during initialization
coerce=str,
validators=[validators.DataRequired()],
validate_choice=False,
)
file_path = StringField("File path", validators=[validators.Optional()])
filename_schema = SelectField(
"Filename_schema",
choices=["ADD LATER"],
validators=[validators.Optional()],
)
counter_mode = SelectField(
"Counter_mode",
choices=["ADD LATER"],
validators=[validators.Optional()],
)
pc = StringField("PC Name", validators=[Optional()])
def __init__(self, choices, *args, **kwargs):
"""Initialize diagnostic choices and bind runtime PC selectables.
Args:
choices: Available diagnostic type choices.
*args: Positional arguments forwarded to ``FlaskForm``.
**kwargs: Keyword arguments forwarded to ``FlaskForm``.
"""
super().__init__(*args, **kwargs)
bind_runtime_field(self, "pc", build_pc_field())
# old version used all diagnostics
# new version only visible fields, passed in to form upn init
# choices = get_db()["diagnostics"].find({}, {"DisplayName": 1})
# diagonstics = [choice["DisplayName"] for choice in choices]
choices = sorted(choices, key=lambda x: x[0])
self.diag_type.choices = choices
[docs]
class ChoicesForm(FlaskForm):
"""
Represents a form for adding choices to a field.
Attributes:
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 = SelectField("Field to Add Choices", choices=[], coerce=str)
choices = StringField("Add a Choice")
choice_submit = SubmitField("Add Choices")
[docs]
class AddtlFields(FlaskForm):
"""
Represents a form for capturing additional fields.
"""
display_name = StringField("Display Name", validators=[DataRequired()])
tooltip = TextAreaField("Tooltip", validators=[Optional()])
details = TextAreaField("Details", validators=[DataRequired()])
class_field = StringField("Class", validators=[Optional()])
wiki_links = StringField("Wiki-Link(s)", validators=[Optional()])
team = StringField("Team", validators=[Optional()])
responsible = StringField("Responsible", validators=[Optional()])
valid_since = DateField("Valid since", validators=[Optional()])
valid_until = DateField("Valid until", validators=[Optional()])
counter_mode_options = HiddenField("Counter mode options", validators=[Optional()])
filename_schema_options = HiddenField(
"Filename schema options", validators=[Optional()]
)
# add new custom fields: String, Float, or Quality (multichoice)
[docs]
class AddNewField(FlaskForm):
"""
Represents a form for adding new custom fields to LabFrog application.
Inherits:
FlaskForm: Base class provided by Flask-WTF for creating web forms.
Attributes:
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 = StringField("New field name", validators=[InputRequired()])
display_name = StringField("Display label", validators=[Optional()])
unit = StringField("Unit", validators=[Optional()])
details = StringField("Details", validators=[Optional()])
options_text = TextAreaField(
"Options (comma-separated, for select types)",
validators=[Optional()],
render_kw={"rows": 2},
)
calculated_from = SelectField(
"Calculated source field",
choices=[],
validators=[Optional()],
validate_choice=False,
)
calculated_formula = StringField(
"Calculated equation (use x)",
validators=[Optional()],
)
calculated_decimal_places = IntegerField(
"Decimal places",
validators=[Optional()],
)
section = SelectField("Section", choices=[]) # done via init
custom_field_type = 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 = SelectField(
"Validation",
choices=[("Optional", "Optional"), ("InputRequired", "InputRequired")],
default="Optional",
)
copy_forward = BooleanField("Copy field value to next shot", default=True)
responsible_person = StringField("Responsible person", validators=[Optional()])
def __init__(self, choices, *args, **kwargs):
calculated_source_choices = kwargs.pop("calculated_source_choices", [])
super().__init__(*args, **kwargs)
choices = sorted(choices, key=lambda x: x[0])
self.section.choices = choices
self.calculated_from.choices = [("", "Select source field")] + sorted(
calculated_source_choices, key=lambda x: x[1]
)
# Define the form for user input
[docs]
class SearchForm(FlaskForm):
"""
Represents a form for searching shot-related data.
"""
# Query
Campaign = SelectField(
"Campaign", choices=["rendered via what is found in mongodb"]
)
# "shot" is the default. "set" narrows to shot-set records; "both" drops the
# mode filter entirely. Anything else (blank, a stale "not set" from an old
# bookmark) is normalised to "shot" in the search route. A document with no
# `mode` field is treated as a shot.
mode = SelectField(
"Select shooting mode",
choices=["shot", "set", "both"],
default="shot",
validate_choice=False,
)
start_date = DateField("Start Date", format="%Y-%m-%d", validators=[Optional()])
end_date = DateField("End Date", format="%Y-%m-%d", validators=[Optional()])
user_query = TextAreaField("Query", validators=[Optional()])
# Projection
latest_versions_only = BooleanField("Only Latest Version", default=True)
return_all_fields = BooleanField(
"Return All Fields", default=True
) # ON unless they turn it off and pick fields
what_to_return = SelectMultipleField(
"Select Fields to Return",
choices=[], # rendered in search route
validators=[Optional()],
)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Safety: ensure projection field exists even if class import was stale
if not hasattr(self, "what_to_return"):
self.__class__.what_to_return = SelectMultipleField(
"Select Fields to Return",
choices=[],
validators=[Optional()],
)
self.what_to_return = self.__class__.what_to_return.bind(
form=self, name="what_to_return"
)
if not hasattr(self.what_to_return, "data"):
self.what_to_return.data = []
self._fields["what_to_return"] = self.what_to_return
apply_parameter_options(self)
apply_field_aliases(self)