# SPDX-FileCopyrightText: 2024-2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
import logging
import re
from datetime import datetime, timezone
from flask import (
Blueprint,
abort,
current_app,
flash,
jsonify,
redirect,
render_template,
request,
session,
url_for,
)
from flask_login import login_required
from wtforms import (
BooleanField,
DateField,
DateTimeField,
FloatField,
IntegerField,
SelectField,
SelectMultipleField,
StringField,
TextAreaField,
)
from wtforms.fields.core import UnboundField
from wtforms.validators import DataRequired, InputRequired
from labfrog.db import get_db
from labfrog.field_specs import (
CURRENT_CALCULATION_NAMES,
PREVIOUS_CALCULATION_NAMES,
custom_field_uses_options,
evaluate_calculation_expression,
infer_label_metadata,
normalize_calculated_decimal_places,
normalize_custom_field_type,
validate_calculation_expression,
)
from labfrog.form_factory import create_dynamic_form
from labfrog.form_feedback import current_username, flash_form_errors
from labfrog.forms import AddNewField, DynamicForm, _get_header_choice
from labfrog.helpers.aliases import invalidate_field_alias_cache
from labfrog.helpers.choices import normalize_choice_value, ordered_unique_strings
from labfrog.helpers.layout import (
build_layout_selection_update,
campaign_is_unset,
find_mode_layout_doc,
get_campaign_header_doc,
get_effective_campaign_layout_doc,
get_source_layout_header_doc,
organize_layout_filter,
paired_mode,
required_identity_fields_for_mode,
resolve_layout_selections,
upsert_campaign_header_mapping,
)
from labfrog.helpers.urls import get_management_return_url
from labfrog.request_state import (
invalidate_custom_field_doc_cache,
refresh_dynamic_form_bindings,
)
def _resolve_section_names(parameters_collection, mode):
headers_choice = _get_header_choice(mode)
if headers_choice != "DEFAULT":
doc = parameters_collection.find_one({
"layout_name": headers_choice,
"$or": [{"mode": mode}, {"mode": {"$exists": False}}],
})
if doc and doc.get("field_sections_dict"):
sections = doc.get("field_sections_dict", {})
return [name for name in sections.keys() if name and name != "NOT_USED"]
layout = parameters_collection.find_one(
{"layout_name": "DEFAULT", "mode": mode},
{"field_sections_dict": 1},
)
if layout and "field_sections_dict" in layout:
return list(layout["field_sections_dict"].keys())
return []
def _resolve_calculated_source_choices(mode: str, parameters_db: str):
"""Return selectable source-field choices for calculated fields."""
all_fields = DynamicForm.get_all_fields(parameters_db, mode)
fields_collection = get_db()["custom_fields_app"]
custom_field_types = {
str(doc.get("field_name") or "").strip(): normalize_custom_field_type(
doc.get("custom_field_type")
)
for doc in fields_collection.find(
{"field_name": {"$exists": True}},
{"field_name": 1, "custom_field_type": 1, "_id": 0},
)
if str(doc.get("field_name") or "").strip()
}
hidden_fields = {
"csrf_token",
"submit",
"update_date_time",
"include_diag_details",
"_field_value_meta",
}
class_level_numeric_fields: set[str] = set()
for klass in DynamicForm.mro():
for attr_name, value in vars(klass).items():
if not isinstance(value, UnboundField):
continue
field_class = getattr(value, "field_class", None)
if field_class in (FloatField, IntegerField):
class_level_numeric_fields.add(attr_name)
def _is_numeric_source(field_name: str) -> bool:
custom_type = custom_field_types.get(field_name)
if custom_type in {"float", "integer"}:
return True
return field_name in class_level_numeric_fields
selected_fields = {
str(name).strip()
for name in session.get("SELECTED_FIELDS_LIST", []) or []
if str(name).strip()
}
choices = []
seen = set()
for field_name in all_fields:
if not isinstance(field_name, str):
continue
name = field_name.strip()
if not name or name in hidden_fields:
continue
if not _is_numeric_source(name):
continue
if selected_fields and name not in selected_fields:
continue
if name in seen:
continue
seen.add(name)
choices.append((name, name))
return choices
blueprint = Blueprint("new_parameter", __name__)
def _display_custom_field_type(raw_type: str) -> str:
"""Map stored values back to the form select options."""
mapping = {
"selectfield": "SelectField",
"selectmultiplefield": "SelectMultipleField",
"select_custom": "SelectField",
"select_multi_custom": "SelectMultipleField",
}
if not raw_type:
return ""
normalized = raw_type.strip()
return mapping.get(normalized.lower(), normalized)
def _normalized_calculated_config(
*,
custom_field_type: str,
field_name: str,
source_field_name: str | None,
equation: str | None,
mode: str = "shot",
) -> tuple[str, str, str | None]:
"""Return validated (source, equation, error)."""
calculated_types = {"calculated", "calculated_prev_shot"}
if custom_field_type not in calculated_types:
return "", "", None
if custom_field_type == "calculated_prev_shot" and mode != "shot":
return (
"",
"",
"Previous-shot calculated fields are only available in shot mode.",
)
source = str(source_field_name or "").strip()
formula = str(equation or "").strip()
if custom_field_type == "calculated_prev_shot" and not formula:
formula = "x - prev_x"
if not source:
return "", "", "Calculated fields require a source field name."
if source == field_name:
return "", "", "Calculated fields cannot reference themselves."
allowed_names = CURRENT_CALCULATION_NAMES
if custom_field_type == "calculated_prev_shot":
allowed_names = CURRENT_CALCULATION_NAMES | PREVIOUS_CALCULATION_NAMES
is_valid, reason = validate_calculation_expression(
formula, allowed_names=allowed_names
)
if not is_valid:
return "", "", f"Calculated equation is invalid: {reason}."
return source, formula, None
def _normalized_calculated_decimal_places(value) -> tuple[int | None, str | None]:
"""Return a validated decimal-place limit for calculated fields."""
places, reason = normalize_calculated_decimal_places(value)
if reason:
return None, f"Calculated decimal places invalid: {reason}."
return places, None
_UNIT_SUFFIX_RE = re.compile(r"\(([^()]+)\)\s*$")
_INTERNAL_BASE_FIELDS = {
"csrf_token",
"submit",
"update_date_time",
"include_diag_details",
"shot_day",
"shot_group",
"_field_value_meta",
}
_NON_EDITABLE_MANAGEMENT_FIELDS = {
"Campaign",
"target",
"mode",
"update_date_time",
"shot_day",
"shot_group",
}
[docs]
def sanitize_field_name(value: str | None) -> str:
"""Normalize user-entered field names before persistence."""
# Normalize non-breaking spaces and collapse whitespace runs.
return " ".join(str(value or "").replace("\u00a0", " ").split())
def _locked_management_fields(mode: str, parameters_collection=None) -> set[str]:
"""Return field names that should not be editable/movable in Adjust Parameters."""
locked = set(_INTERNAL_BASE_FIELDS)
locked.update(_NON_EDITABLE_MANAGEMENT_FIELDS)
locked.update(
required_identity_fields_for_mode(
mode, parameters_collection=parameters_collection
)
)
return locked
def _set_session_field_inclusion(field_name: str, *, include_in_form: bool) -> None:
"""Apply include/exclude changes to current session field selections."""
selected_fields = ordered_unique_strings(
session.get("SELECTED_FIELDS_LIST", []) or []
)
diagnostics_only = ordered_unique_strings(session.get("diagnostics_only", []) or [])
if include_in_form:
if field_name not in selected_fields:
selected_fields.append(field_name)
else:
selected_fields = [name for name in selected_fields if name != field_name]
diagnostics_only = [name for name in diagnostics_only if name != field_name]
session["SELECTED_FIELDS_LIST"] = ordered_unique_strings(selected_fields)
session["diagnostics_only"] = ordered_unique_strings(diagnostics_only)
def _replace_field_name_in_list(
values: list | None,
*,
old_name: str,
new_name: str,
) -> tuple[list, bool]:
current = list(values or [])
replaced = []
changed = False
for value in current:
if value == old_name:
replaced.append(new_name)
changed = True
else:
replaced.append(value)
deduped = ordered_unique_strings(replaced)
if deduped != current:
changed = True
return deduped, changed
def _replace_field_name_in_sections_dict(
sections: dict | None,
*,
old_name: str,
new_name: str,
) -> tuple[dict, bool]:
if not isinstance(sections, dict):
return {}, False
changed = False
updated = {}
for section_name, value in sections.items():
if isinstance(value, dict):
fields, fields_changed = _replace_field_name_in_list(
value.get("fields"),
old_name=old_name,
new_name=new_name,
)
updated_value = dict(value)
updated_value["fields"] = fields
if fields_changed:
changed = True
updated[section_name] = updated_value
else:
fields, fields_changed = _replace_field_name_in_list(
value,
old_name=old_name,
new_name=new_name,
)
if fields_changed:
changed = True
updated[section_name] = fields
return updated, changed
def _section_fields(value) -> list[str]:
"""Return a plain list of field names from one section payload value."""
if isinstance(value, dict):
return list(value.get("fields") or [])
return list(value or [])
def _with_section_fields(
original_value,
fields: list[str],
*,
fallback_order: int,
prefer_dict_shape: bool,
):
"""Write updated fields back while preserving section metadata shape."""
if isinstance(original_value, dict):
updated = dict(original_value)
updated["fields"] = fields
return updated
if prefer_dict_shape:
return {"fields": fields, "order": fallback_order, "color": None}
return fields
def _move_field_to_section(
sections: dict | None, *, field_name: str, section_name: str | None
) -> tuple[dict, bool]:
"""Move one field into the requested section and remove duplicates elsewhere."""
if not isinstance(sections, dict):
sections = {}
changed = False
updated_sections = {}
prefer_dict_shape = any(isinstance(value, dict) for value in sections.values())
for index, (name, value) in enumerate(sections.items()):
fields = _section_fields(value)
filtered_fields = [existing for existing in fields if existing != field_name]
if filtered_fields != fields:
changed = True
updated_sections[name] = _with_section_fields(
value,
filtered_fields,
fallback_order=index,
prefer_dict_shape=prefer_dict_shape,
)
target_section = str(section_name or "").strip()
if target_section:
existing_target = updated_sections.get(target_section)
if existing_target is None:
existing_target = _with_section_fields(
None,
[],
fallback_order=len(updated_sections),
prefer_dict_shape=prefer_dict_shape,
)
target_fields = _section_fields(existing_target)
if field_name not in target_fields:
target_fields.append(field_name)
changed = True
updated_sections[target_section] = _with_section_fields(
existing_target,
target_fields,
fallback_order=len(updated_sections),
prefer_dict_shape=prefer_dict_shape,
)
return updated_sections, changed
def _remove_field_from_hidden_sections(
hidden_sections: dict | None, *, field_name: str
) -> tuple[dict, bool]:
"""Drop one field from hidden-section caches and prune empty section entries."""
if not isinstance(hidden_sections, dict):
return {}, False
changed = False
updated = {}
for section_name, values in hidden_sections.items():
current_values = list(values or [])
filtered_values = [value for value in current_values if value != field_name]
if filtered_values != current_values:
changed = True
if filtered_values:
updated[section_name] = filtered_values
return updated, changed
def _ensure_campaign_membership(
campaigns: list | None, *, campaign: str | None
) -> tuple[list[str], bool]:
"""Ensure the normalized active campaign is present in a layout campaigns list."""
current = ordered_unique_strings(campaigns or [])
normalized_campaign = normalize_choice_value(campaign or "")
if campaign_is_unset(normalized_campaign):
return current, False
if normalized_campaign in current:
return current, False
return [*current, normalized_campaign], True
def _update_layout_doc_with_field(
*,
parameters_collection,
layout_doc: dict,
field_name: str,
section_name: str | None,
username: str,
add_to_selected: bool,
include_in_form: bool | None,
campaign: str | None,
) -> bool:
"""Persist one field/section sync into a layout or organize-form document."""
if not layout_doc or not layout_doc.get("_id"):
return False
target_section_name = section_name if include_in_form is not False else None
updated_sections, sections_changed = _move_field_to_section(
layout_doc.get("field_sections_dict"),
field_name=field_name,
section_name=target_section_name,
)
hidden_sections, hidden_changed = _remove_field_from_hidden_sections(
layout_doc.get("section_hidden_fields"),
field_name=field_name,
)
selected_parameters, selected_diagnostics, _selected_fields = (
resolve_layout_selections(layout_doc)
)
if include_in_form is False:
selected_parameters = [
name for name in selected_parameters if name != field_name
]
selected_diagnostics = [
name for name in selected_diagnostics if name != field_name
]
elif (
(include_in_form is True or add_to_selected)
and field_name not in selected_parameters
and field_name not in selected_diagnostics
):
selected_parameters.append(field_name)
selection_update = build_layout_selection_update(
selected_parameters,
selected_diagnostics,
)
selection_changed = any(
list(layout_doc.get(key) or []) != list(selection_update.get(key) or [])
for key in (
"selected_parameters",
"selected_diagnostics",
"diagnostics_list",
"selected_fields",
)
)
updated_campaigns, campaigns_changed = _ensure_campaign_membership(
layout_doc.get("campaigns"),
campaign=campaign,
)
if not any((
sections_changed,
hidden_changed,
selection_changed,
campaigns_changed,
)):
return False
updates = {
"field_sections_dict": updated_sections,
**selection_update,
"updated_by": username,
"updated_at": datetime.now(timezone.utc),
}
if hidden_changed:
updates["section_hidden_fields"] = hidden_sections
if campaigns_changed:
updates["campaigns"] = updated_campaigns
parameters_collection.update_one({"_id": layout_doc["_id"]}, {"$set": updates})
return True
def _active_source_layout_doc(parameters_collection, mode: str) -> dict | None:
"""Resolve the active source field-layout doc for the current session context."""
selected_layout = (
session.get("selected_layout")
or session.get(f"selected_source_layout_{mode}")
or session.get(f"selected_source_layout_{paired_mode(mode)}")
or "DEFAULT"
)
selected_layout = str(selected_layout or "").strip() or "DEFAULT"
selected_layout_doc = find_mode_layout_doc(
parameters_collection, selected_layout, mode
)
if selected_layout_doc:
return selected_layout_doc
current_campaign = normalize_choice_value(session.get("selected_campaign") or "")
if not campaign_is_unset(current_campaign):
campaign_doc = get_effective_campaign_layout_doc(
parameters_collection,
mode,
current_campaign,
)
if campaign_doc and campaign_doc.get("_id"):
return campaign_doc
return None
def _active_organized_doc_for_source_layout(
parameters_collection,
mode: str,
*,
source_layout_name: str | None,
) -> dict | None:
"""Resolve the active organize-form doc linked to one source layout."""
source_layout = (source_layout_name or "").strip()
if not source_layout:
return None
current_campaign = normalize_choice_value(session.get("selected_campaign") or "")
if not campaign_is_unset(current_campaign):
campaign_header_doc = get_campaign_header_doc(
mode,
current_campaign,
source_layout_name=source_layout,
)
if campaign_header_doc:
return campaign_header_doc
selected_header = (
session.get("selected_headers_layout")
or session.get(f"selected_headers_layout_{mode}")
or session.get(f"selected_headers_layout_{paired_mode(mode)}")
or ""
)
selected_header = str(selected_header or "").strip()
if selected_header and selected_header != "DEFAULT":
mode_filters = (
{"mode": mode},
{"mode": {"$exists": False}},
{"mode": paired_mode(mode)},
)
for require_source_layout in (True, False):
for mode_filter in mode_filters:
query = {
"$and": [
{"layout_name": selected_header, **mode_filter},
organize_layout_filter(),
]
}
if require_source_layout:
query["source_layout_name"] = source_layout
header_doc = parameters_collection.find_one(query)
if header_doc:
return header_doc
return get_source_layout_header_doc(mode, source_layout)
def _sync_current_field_templates(
*,
parameters_collection,
mode: str,
field_name: str,
section_name: str | None,
username: str,
add_to_selected: bool,
include_in_form: bool | None = None,
) -> None:
"""Sync field placement/selection into active Select Fields + Organize Form docs."""
source_layout_doc = _active_source_layout_doc(parameters_collection, mode)
source_layout_name = (
(source_layout_doc or {}).get("layout_name")
or session.get("selected_layout")
or "DEFAULT"
)
current_campaign = normalize_choice_value(session.get("selected_campaign") or "")
if source_layout_doc:
_update_layout_doc_with_field(
parameters_collection=parameters_collection,
layout_doc=source_layout_doc,
field_name=field_name,
section_name=section_name,
username=username,
add_to_selected=add_to_selected,
include_in_form=include_in_form,
campaign=current_campaign,
)
session["selected_layout"] = source_layout_name
session[f"selected_source_layout_{mode}"] = source_layout_name
session[f"selected_source_layout_{paired_mode(mode)}"] = source_layout_name
organized_doc = _active_organized_doc_for_source_layout(
parameters_collection,
mode,
source_layout_name=source_layout_name,
)
if organized_doc:
_update_layout_doc_with_field(
parameters_collection=parameters_collection,
layout_doc=organized_doc,
field_name=field_name,
section_name=section_name,
username=username,
add_to_selected=add_to_selected,
include_in_form=include_in_form,
campaign=current_campaign,
)
if not campaign_is_unset(current_campaign):
header_name = (
organized_doc.get("layout_name") or organized_doc.get("name") or ""
).strip()
if header_name:
upsert_campaign_header_mapping(
mode,
current_campaign,
header_name,
source_layout_name=source_layout_name,
updated_by=username,
)
def _rename_field_references(
*,
old_name: str,
new_name: str,
parameters_collection,
) -> None:
if old_name == new_name:
return
db = get_db()
db["custom_fields_app"].update_many(
{"calculated_from": old_name},
{"$set": {"calculated_from": new_name}},
)
db["campaign_layouts"].update_many(
{"field_name": old_name},
{"$set": {"field_name": new_name}},
)
db["field_aliases"].update_many(
{"field_name": old_name},
{"$set": {"field_name": new_name}},
)
invalidate_field_alias_cache()
for doc in parameters_collection.find({}):
updates = {}
any_changed = False
for key in (
"selected_fields",
"selected_parameters",
"selected_diagnostics",
"diagnostics_list",
"always_include",
):
if key not in doc:
continue
replaced, changed = _replace_field_name_in_list(
doc.get(key),
old_name=old_name,
new_name=new_name,
)
if changed:
updates[key] = replaced
any_changed = True
sections, sections_changed = _replace_field_name_in_sections_dict(
doc.get("field_sections_dict"),
old_name=old_name,
new_name=new_name,
)
if sections_changed:
updates["field_sections_dict"] = sections
any_changed = True
hidden_sections, hidden_changed = _replace_field_name_in_sections_dict(
doc.get("section_hidden_fields"),
old_name=old_name,
new_name=new_name,
)
if hidden_changed:
updates["section_hidden_fields"] = hidden_sections
any_changed = True
if (
any_changed
and doc.get("template_type") == "field_layout"
and ("selected_parameters" in updates or "selected_diagnostics" in updates)
):
selected_parameters, selected_diagnostics, _ = resolve_layout_selections({
**doc,
**updates,
})
updates.update(
build_layout_selection_update(selected_parameters, selected_diagnostics)
)
if any_changed:
parameters_collection.update_one({"_id": doc["_id"]}, {"$set": updates})
selected_fields = list(session.get("SELECTED_FIELDS_LIST", []) or [])
selected_fields, selected_changed = _replace_field_name_in_list(
selected_fields,
old_name=old_name,
new_name=new_name,
)
if selected_changed:
session["SELECTED_FIELDS_LIST"] = selected_fields
diagnostics_only = list(session.get("diagnostics_only", []) or [])
diagnostics_only, diagnostics_changed = _replace_field_name_in_list(
diagnostics_only,
old_name=old_name,
new_name=new_name,
)
if diagnostics_changed:
session["diagnostics_only"] = diagnostics_only
def _remove_field_references(*, field_name: str, parameters_collection) -> None:
db = get_db()
db["campaign_layouts"].delete_many({"field_name": field_name})
db["field_aliases"].delete_many({"field_name": field_name})
invalidate_field_alias_cache()
for doc in parameters_collection.find({}):
updates = {}
any_changed = False
for key in (
"selected_fields",
"selected_parameters",
"selected_diagnostics",
"diagnostics_list",
"always_include",
):
if key not in doc:
continue
current_values = list(doc.get(key) or [])
filtered = [value for value in current_values if value != field_name]
if filtered != current_values:
updates[key] = filtered
any_changed = True
sections = doc.get("field_sections_dict")
if isinstance(sections, dict):
cleaned_sections = {}
sections_changed = False
for section_name, value in sections.items():
if isinstance(value, dict):
current_fields = list(value.get("fields") or [])
filtered_fields = [
value_name
for value_name in current_fields
if value_name != field_name
]
if filtered_fields != current_fields:
sections_changed = True
updated_value = dict(value)
updated_value["fields"] = filtered_fields
cleaned_sections[section_name] = updated_value
else:
current_fields = list(value or [])
filtered_fields = [
value_name
for value_name in current_fields
if value_name != field_name
]
if filtered_fields != current_fields:
sections_changed = True
cleaned_sections[section_name] = filtered_fields
if sections_changed:
updates["field_sections_dict"] = cleaned_sections
any_changed = True
hidden_sections = doc.get("section_hidden_fields")
if isinstance(hidden_sections, dict):
cleaned_hidden = {}
hidden_changed = False
for section_name, values in hidden_sections.items():
filtered_values = [
value_name
for value_name in list(values or [])
if value_name != field_name
]
if filtered_values:
cleaned_hidden[section_name] = filtered_values
if filtered_values != list(values or []):
hidden_changed = True
if hidden_changed:
updates["section_hidden_fields"] = cleaned_hidden
any_changed = True
if any_changed:
if doc.get("template_type") == "field_layout" and (
"selected_parameters" in updates or "selected_diagnostics" in updates
):
selected_parameters, selected_diagnostics, _ = (
resolve_layout_selections({**doc, **updates})
)
updates.update(
build_layout_selection_update(
selected_parameters, selected_diagnostics
)
)
parameters_collection.update_one({"_id": doc["_id"]}, {"$set": updates})
selected_fields = [
value_name
for value_name in list(session.get("SELECTED_FIELDS_LIST", []) or [])
if value_name != field_name
]
session["SELECTED_FIELDS_LIST"] = ordered_unique_strings(selected_fields)
diagnostics_only = [
value_name
for value_name in list(session.get("diagnostics_only", []) or [])
if value_name != field_name
]
session["diagnostics_only"] = ordered_unique_strings(diagnostics_only)
def _ui_custom_field_type_for_bound_field(field_obj) -> str:
if isinstance(field_obj, IntegerField):
return "integer"
if isinstance(field_obj, FloatField):
return "float"
if isinstance(field_obj, BooleanField):
return "boolean"
if isinstance(field_obj, DateField):
return "date"
if isinstance(field_obj, DateTimeField):
return "datetime"
if isinstance(field_obj, SelectMultipleField):
return "SelectMultipleField"
if isinstance(field_obj, SelectField):
return "SelectField"
return "string"
def _validator_name_for_bound_field(field_obj) -> str:
validators = list(getattr(field_obj, "validators", []) or [])
for validator in validators:
if isinstance(validator, (InputRequired, DataRequired)):
return "InputRequired"
return "Optional"
def _default_section_by_field(parameters_collection, mode: str) -> dict[str, str]:
section_map: dict[str, str] = {}
default_doc = parameters_collection.find_one(
{"layout_name": "DEFAULT", "mode": mode},
{"field_sections_dict": 1},
)
if not default_doc:
default_doc = parameters_collection.find_one(
{"layout_name": "DEFAULT"},
{"field_sections_dict": 1},
)
sections = (default_doc or {}).get("field_sections_dict", {}) or {}
for section_name, value in 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 section_map
):
section_map[field_name] = section_name
return section_map
def _base_parameter_rows(mode: str, parameters_db: str) -> list[dict]:
db = get_db()
parameters_collection = db[parameters_db]
section_map = _default_section_by_field(parameters_collection, mode)
visible_fields = list(section_map.keys())
if not visible_fields:
return []
diagnostics = {
str(doc.get("DisplayName") or "").strip()
for doc in db["diagnostics"].find({}, {"DisplayName": 1, "_id": 0})
if str(doc.get("DisplayName") or "").strip()
}
custom_field_names = {
str(doc.get("field_name") or "").strip()
for doc in db["custom_fields_app"].find(
{"field_name": {"$exists": True}},
{"field_name": 1, "_id": 0},
)
if str(doc.get("field_name") or "").strip()
}
form = create_dynamic_form(
collection=parameters_db,
options_collection=None,
mode=mode,
visible_fields=visible_fields,
)
rows = []
for field_name in visible_fields:
if (
field_name in diagnostics
or field_name in custom_field_names
or field_name in _INTERNAL_BASE_FIELDS
):
continue
field_obj = getattr(form, field_name, None)
if field_obj is None:
continue
display_name, unit = infer_label_metadata(
form_field=field_obj,
field_name=field_name,
)
if not unit:
label_text = str(
getattr(getattr(field_obj, "label", None), "text", "")
).strip()
match = _UNIT_SUFFIX_RE.search(label_text)
if match:
unit = match.group(1).strip()
details = str(
(getattr(field_obj, "render_kw", {}) or {}).get("title") or ""
).strip()
rows.append({
"field_name": field_name,
"display_name": display_name if display_name != field_name else "",
"unit": unit,
"details": details,
"section": section_map.get(field_name, ""),
"custom_field_type": _ui_custom_field_type_for_bound_field(field_obj),
"validators": _validator_name_for_bound_field(field_obj),
"responsible_person": "",
"copy_forward": True,
"_source": "base",
})
return sorted(rows, key=lambda row: row.get("field_name", ""))
def _parameter_catalog_rows(mode: str, parameters_db: str) -> list[dict]:
db = get_db()
custom_rows = list(db["custom_fields_app"].find({}, {"_id": 0}))
rows_by_name = {
row.get("field_name"): row for row in _base_parameter_rows(mode, parameters_db)
}
for custom_row in custom_rows:
field_name = (custom_row.get("field_name") or "").strip()
if not field_name:
continue
merged = dict(rows_by_name.get(field_name, {}))
merged.update(custom_row)
merged["field_name"] = field_name
merged.setdefault("copy_forward", True)
merged["_source"] = "custom"
merged.setdefault("display_name", "")
merged.setdefault("unit", "")
merged.setdefault("details", "")
merged.setdefault("section", "")
merged.setdefault("custom_field_type", "string")
merged.setdefault("validators", "Optional")
merged.setdefault("options_text", "")
merged.setdefault("calculated_from", "")
merged.setdefault("calculated_formula", "")
merged.setdefault("calculated_decimal_places", None)
merged.setdefault("responsible_person", "")
rows_by_name[field_name] = merged
rows = []
for row in rows_by_name.values():
row.setdefault("copy_forward", True)
row.setdefault("display_name", "")
row.setdefault("unit", "")
row.setdefault("details", "")
row.setdefault("section", "")
row.setdefault("custom_field_type", "string")
row.setdefault("validators", "Optional")
row.setdefault("options_text", "")
row.setdefault("calculated_from", "")
row.setdefault("calculated_formula", "")
row.setdefault("calculated_decimal_places", None)
row.setdefault("responsible_person", "")
rows.append(row)
return sorted(rows, key=lambda row: row.get("field_name", ""))
def _current_selected_field_names(parameters_collection, mode: str) -> set[str]:
"""Return currently selected field names for the active source layout/campaign."""
source_layout_doc = _active_source_layout_doc(parameters_collection, mode)
if source_layout_doc:
return set(resolve_layout_selections(source_layout_doc)[2])
return {
str(name).strip()
for name in session.get("SELECTED_FIELDS_LIST", []) or []
if str(name).strip()
}
def _management_parameter_rows(
mode: str,
parameters_db: str,
*,
parameters_collection,
) -> list[dict]:
"""Return table rows for Adjust Parameters with locked fields excluded."""
locked_fields = _locked_management_fields(mode, parameters_collection)
selected_fields = _current_selected_field_names(parameters_collection, mode)
rows = []
for row in _parameter_catalog_rows(mode, parameters_db):
field_name = str(row.get("field_name") or "").strip()
if not field_name or field_name in locked_fields:
continue
row_copy = dict(row)
row_copy["_included_in_form"] = field_name in selected_fields
rows.append(row_copy)
return rows
def _catalog_entry_for_field(
mode: str, parameters_db: str, field_name: str
) -> dict | None:
target = (field_name or "").strip()
if not target:
return None
for row in _parameter_catalog_rows(mode, parameters_db):
if row.get("field_name") == target:
return row
return None
[docs]
@login_required
@blueprint.route("/add_new_field/verify_calculated", methods=["POST"]) # legacy alias
@blueprint.route("/new_parameter/verify_calculated", methods=["POST"]) # canonical
def verify_calculated_field():
"""Validate calculated field configuration and return a safe preview value."""
mode = session.get("mode", "shot")
parameters_db = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
payload = request.get_json(silent=True) or {}
source_field = str(payload.get("source_field") or "").strip()
formula = str(payload.get("formula") or "").strip()
calculation_type = normalize_custom_field_type(payload.get("calculation_type"))
sample_x_raw = payload.get("sample_x", 1)
sample_prev_x_raw = payload.get("sample_prev_x", 0)
decimal_places, decimal_error = _normalized_calculated_decimal_places(
payload.get("decimal_places")
)
if decimal_error:
return jsonify({"ok": False, "error": decimal_error}), 400
allowed_calculation_types = {"calculated", "calculated_prev_shot"}
if calculation_type not in allowed_calculation_types:
calculation_type = "calculated"
if calculation_type == "calculated_prev_shot" and not formula:
formula = "x - prev_x"
if calculation_type == "calculated_prev_shot" and mode != "shot":
return (
jsonify({
"ok": False,
"error": (
"Previous-shot calculated fields are only available in shot mode."
),
}),
400,
)
if not source_field:
return (
jsonify({"ok": False, "error": "Select a numeric source field first."}),
400,
)
allowed_sources = {
name for name, _label in _resolve_calculated_source_choices(mode, parameters_db)
}
if source_field not in allowed_sources:
return (
jsonify({
"ok": False,
"error": (
f"Source field '{source_field}' is not an eligible numeric "
"source field."
),
}),
400,
)
allowed_names = CURRENT_CALCULATION_NAMES
if calculation_type == "calculated_prev_shot":
allowed_names = CURRENT_CALCULATION_NAMES | PREVIOUS_CALCULATION_NAMES
is_valid, reason = validate_calculation_expression(
formula, allowed_names=allowed_names
)
if not is_valid:
return (
jsonify({
"ok": False,
"error": f"Calculated equation is invalid: {reason}.",
}),
400,
)
try:
sample_x = float(sample_x_raw)
except (TypeError, ValueError):
return (
jsonify({"ok": False, "error": "Sample x value must be numeric."}),
400,
)
sample_prev_x = None
if calculation_type == "calculated_prev_shot":
try:
sample_prev_x = float(sample_prev_x_raw)
except (TypeError, ValueError):
return (
jsonify({
"ok": False,
"error": "Sample previous x value must be numeric.",
}),
400,
)
try:
preview = evaluate_calculation_expression(
formula,
x=sample_x,
prev_x=sample_prev_x,
allowed_names=allowed_names,
)
if decimal_places is not None:
preview = round(preview, decimal_places)
except (TypeError, ValueError, OverflowError) as exc:
return (
jsonify({"ok": False, "error": f"Calculated equation is invalid: {exc}."}),
400,
)
return jsonify({
"ok": True,
"message": "Calculated field setup is valid.",
"preview": {"x": sample_x, "prev_x": sample_prev_x, "result": preview},
})
[docs]
@login_required
@blueprint.route("/add_new_field", methods=["GET", "POST"]) # legacy alias
@blueprint.route("/new_parameter", methods=["GET", "POST"]) # canonical path
def new_parameter():
"""
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:
render_template: Renders the 'new_parameter.html' template with the appropriate context.
"""
mode = session.get("mode", "shot")
PARAMETERS_DB = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
parameters_collection = get_db()[PARAMETERS_DB]
section_names = _resolve_section_names(parameters_collection, mode)
if section_names:
logging.debug(f"Section Names: {section_names}")
else:
logging.debug("No layout found or 'field_sections_dict' not present.")
calculated_source_choices = _resolve_calculated_source_choices(mode, PARAMETERS_DB)
form = AddNewField(
choices=section_names,
calculated_source_choices=calculated_source_choices,
)
original_field_name = ""
fields_collection = get_db()["custom_fields_app"]
custom_fields = _management_parameter_rows(
mode,
PARAMETERS_DB,
parameters_collection=parameters_collection,
)
current_endpoint = request.endpoint
SELECTED_FIELDS_LIST = session.get("SELECTED_FIELDS_LIST", [])
selected_layout = session.get("selected_layout", "DEFAULT")
return_url = get_management_return_url(
request.values.get("next"),
url_for("field_selection.field_selection"),
)
responsible_person = current_username(default="")
form.responsible_person.data = responsible_person
if request.method == "POST" and form.validate_on_submit():
new_field_name = sanitize_field_name(form.newFieldName.data)
form.newFieldName.data = new_field_name
if not new_field_name:
flash("Field name is required.", "error")
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=False,
)
display_name = form.display_name.data
unit = form.unit.data
details = form.details.data
section = form.section.data
custom_field_type = normalize_custom_field_type(form.custom_field_type.data)
validators = form.validators.data
options_text = form.options_text.data or ""
calculated_from = form.calculated_from.data or ""
calculated_formula = form.calculated_formula.data or ""
calculated_decimal_places = form.calculated_decimal_places.data
copy_forward = (
bool(form.copy_forward.data) if "copy_forward" in request.form else True
)
if custom_field_type in {"calculated", "calculated_prev_shot"}:
section = ""
if not custom_field_uses_options(custom_field_type):
options_text = ""
calculated_from, calculated_formula, calc_error = _normalized_calculated_config(
custom_field_type=custom_field_type,
field_name=new_field_name,
source_field_name=calculated_from,
equation=calculated_formula,
mode=mode,
)
if calc_error:
flash(calc_error, "error")
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=False,
)
calculated_decimal_places, decimal_error = (
_normalized_calculated_decimal_places(calculated_decimal_places)
)
if custom_field_type not in {"calculated", "calculated_prev_shot"}:
calculated_decimal_places = None
if decimal_error:
flash(decimal_error, "error")
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=False,
)
new_data = {
"field_name": new_field_name,
"display_name": display_name,
"unit": unit,
"details": details,
"section": section,
"custom_field_type": custom_field_type,
"validators": validators,
"options_text": options_text,
"calculated_from": calculated_from,
"calculated_formula": calculated_formula,
"calculated_decimal_places": calculated_decimal_places,
"responsible_person": responsible_person,
"copy_forward": copy_forward,
}
# Check if field already exists
existing_field = fields_collection.find_one({"field_name": new_field_name})
if existing_field:
fields_collection.update_one(
{"field_name": new_field_name}, {"$set": new_data}
)
# assume they want to add it to SELECTED_FIELDS
if new_field_name not in SELECTED_FIELDS_LIST:
SELECTED_FIELDS_LIST.append(new_field_name)
session["SELECTED_FIELDS_LIST"] = ordered_unique_strings(
SELECTED_FIELDS_LIST
)
flash(
f"Field {form.newFieldName.data} modified successfully!",
"success",
)
else:
# Save the new field to MongoDB
fields_collection.insert_one(new_data)
SELECTED_FIELDS_LIST.append(new_field_name)
session["SELECTED_FIELDS_LIST"] = ordered_unique_strings(
SELECTED_FIELDS_LIST
)
flash(f"Field {new_field_name} added successfully!", "success")
_sync_current_field_templates(
parameters_collection=parameters_collection,
mode=mode,
field_name=new_field_name,
section_name=section,
username=responsible_person,
add_to_selected=True,
)
# Ensure same-request catalog generation sees the just-saved field metadata.
invalidate_custom_field_doc_cache()
# update available custom_fields
custom_fields = _management_parameter_rows(
mode,
PARAMETERS_DB,
parameters_collection=parameters_collection,
)
else:
flash_form_errors(form, logger=logging.getLogger(__name__))
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=False,
)
[docs]
@login_required
@blueprint.route("/modify_field/<field_name>", methods=["GET", "POST"])
def modify_field(field_name):
"""
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:
render_template: Renders the 'new_parameter.html' template with the updated context.
"""
mode = session.get("mode", "shot")
PARAMETERS_DB = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
db = get_db()
parameters_collection = db[PARAMETERS_DB]
fields_collection = db["custom_fields_app"]
current_endpoint = request.endpoint
original_field_name = field_name
selected_layout = session.get("selected_layout", "DEFAULT")
return_url = get_management_return_url(
request.values.get("next"),
url_for("field_selection.field_selection"),
)
responsible_person = current_username(default="")
if field_name in _locked_management_fields(mode, parameters_collection):
flash(f"Field {field_name} is locked and cannot be modified.", "error")
return redirect(url_for("new_parameter.new_parameter", next=return_url))
existing_override = fields_collection.find_one({"field_name": field_name})
catalog_entry = existing_override or _catalog_entry_for_field(
mode, PARAMETERS_DB, field_name
)
if not catalog_entry:
abort(404)
is_base_field = not bool(existing_override)
custom_fields = _management_parameter_rows(
mode,
PARAMETERS_DB,
parameters_collection=parameters_collection,
)
section_names = _resolve_section_names(parameters_collection, mode)
if section_names:
logging.debug(f"Section Names: {section_names}")
else:
logging.debug("No layout found or 'field_sections_dict' not present.")
calculated_source_choices = _resolve_calculated_source_choices(mode, PARAMETERS_DB)
form = AddNewField(
choices=section_names,
calculated_source_choices=calculated_source_choices,
)
if request.method == "GET":
form.newFieldName.data = catalog_entry.get("field_name", "")
form.display_name.data = catalog_entry.get("display_name", "")
form.unit.data = catalog_entry.get("unit", "")
form.details.data = catalog_entry.get("details", "")
form.section.data = catalog_entry.get("section", "")
form.custom_field_type.data = _display_custom_field_type(
catalog_entry.get("custom_field_type", "")
)
form.validators.data = catalog_entry.get("validators", "Optional")
form.options_text.data = catalog_entry.get("options_text", "")
form.calculated_from.data = catalog_entry.get("calculated_from", "")
form.calculated_formula.data = catalog_entry.get("calculated_formula", "")
form.calculated_decimal_places.data = catalog_entry.get(
"calculated_decimal_places"
)
form.copy_forward.data = bool(catalog_entry.get("copy_forward", True))
form.responsible_person.data = responsible_person
# If it's a POST request, upsert the data in MongoDB
if request.method == "POST" and form.validate_on_submit():
sanitized_field_name = sanitize_field_name(form.newFieldName.data)
form.newFieldName.data = sanitized_field_name
if not sanitized_field_name:
flash("Field name is required.", "error")
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=is_base_field,
)
if sanitized_field_name != original_field_name and fields_collection.find_one({
"field_name": sanitized_field_name
}):
flash(
f"Field name '{sanitized_field_name}' already exists. Choose another name.",
"error",
)
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=is_base_field,
)
normalized_field_type = normalize_custom_field_type(form.custom_field_type.data)
options_text = form.options_text.data or ""
calculated_from = form.calculated_from.data or ""
calculated_formula = form.calculated_formula.data or ""
calculated_decimal_places = form.calculated_decimal_places.data
section_value = form.section.data
copy_forward = (
bool(form.copy_forward.data) if "copy_forward" in request.form else True
)
if normalized_field_type in {"calculated", "calculated_prev_shot"}:
section_value = ""
if not custom_field_uses_options(normalized_field_type):
options_text = ""
calculated_from, calculated_formula, calc_error = _normalized_calculated_config(
custom_field_type=normalized_field_type,
field_name=sanitized_field_name,
source_field_name=calculated_from,
equation=calculated_formula,
mode=mode,
)
if calc_error:
flash(calc_error, "error")
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=is_base_field,
)
calculated_decimal_places, decimal_error = (
_normalized_calculated_decimal_places(calculated_decimal_places)
)
if normalized_field_type not in {"calculated", "calculated_prev_shot"}:
calculated_decimal_places = None
if decimal_error:
flash(decimal_error, "error")
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=is_base_field,
)
new_data = {
"field_name": sanitized_field_name,
"display_name": form.display_name.data,
"unit": form.unit.data,
"details": form.details.data,
"section": section_value,
"custom_field_type": normalized_field_type,
"validators": form.validators.data,
"options_text": options_text,
"calculated_from": calculated_from,
"calculated_formula": calculated_formula,
"calculated_decimal_places": calculated_decimal_places,
"responsible_person": responsible_person,
"copy_forward": copy_forward,
}
fields_collection.update_one(
{"field_name": field_name}, {"$set": new_data}, upsert=True
)
selected_fields = list(session.get("SELECTED_FIELDS_LIST", []) or [])
selected_fields, selected_changed = _replace_field_name_in_list(
selected_fields,
old_name=field_name,
new_name=new_data.get("field_name"),
)
if selected_changed:
session["SELECTED_FIELDS_LIST"] = selected_fields
if field_name != new_data.get("field_name"):
_rename_field_references(
old_name=field_name,
new_name=new_data.get("field_name"),
parameters_collection=parameters_collection,
)
_sync_current_field_templates(
parameters_collection=parameters_collection,
mode=mode,
field_name=new_data.get("field_name"),
section_name=section_value,
username=responsible_person,
add_to_selected=False,
)
invalidate_custom_field_doc_cache()
refresh_dynamic_form_bindings()
flash(
f"Field {field_name} modified successfully!",
"success",
)
# Send the user back to the canonical page while keeping the legacy route live.
return redirect(url_for("new_parameter.new_parameter", next=return_url))
flash_form_errors(form, logger=logging.getLogger(__name__))
return render_template(
"new_parameter.html",
selected_layout=selected_layout,
active_tab="new_parameter",
original_field_name=original_field_name,
form=form,
custom_fields=custom_fields,
current_endpoint=current_endpoint,
return_url=return_url,
is_base_field=is_base_field,
)
[docs]
@login_required
@blueprint.route("/delete_field/<field_name>", methods=["POST"])
def delete_field(field_name):
"""
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:
redirect: Redirects the user back to the field management page.
"""
return_url = get_management_return_url(
request.values.get("next"),
url_for("field_selection.field_selection"),
)
try:
parameters_db = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
parameters_collection = get_db()[parameters_db]
fields_collection = get_db()["custom_fields_app"]
result = fields_collection.delete_one({"field_name": field_name})
if result.deleted_count == 0:
flash(f"Field {field_name} was not found!", "error")
else:
flash(f"Field {field_name} deleted successfully!", "success")
_remove_field_references(
field_name=field_name,
parameters_collection=parameters_collection,
)
invalidate_custom_field_doc_cache()
refresh_dynamic_form_bindings()
except Exception as e:
logging.error(f"EXCEPTION DELETING {e}")
return redirect(url_for("new_parameter.new_parameter", next=return_url))
[docs]
@login_required
@blueprint.route(
"/new_parameter/add_to_current_templates/<field_name>",
methods=["POST"],
)
def add_field_to_current_templates(field_name):
"""Add one field into the active campaign's Selected Fields and Organized Form."""
mode = session.get("mode", "shot")
parameters_db = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
db = get_db()
parameters_collection = db[parameters_db]
return_url = get_management_return_url(
request.values.get("next"),
url_for("field_selection.field_selection"),
)
normalized_field_name = str(field_name or "").strip()
if not normalized_field_name:
flash("Field name is required.", "error")
return redirect(url_for("new_parameter.new_parameter", next=return_url))
if normalized_field_name in _locked_management_fields(mode, parameters_collection):
flash(
f"Field {normalized_field_name} is locked and cannot be changed here.",
"error",
)
return redirect(url_for("new_parameter.new_parameter", next=return_url))
catalog_entry = _catalog_entry_for_field(mode, parameters_db, normalized_field_name)
if not catalog_entry:
flash(f"Field {normalized_field_name} was not found.", "error")
return redirect(url_for("new_parameter.new_parameter", next=return_url))
section_value = str(catalog_entry.get("section") or "").strip()
if not section_value:
section_value = _default_section_by_field(parameters_collection, mode).get(
normalized_field_name, ""
)
if not section_value:
known_sections = _resolve_section_names(parameters_collection, mode)
section_value = (
known_sections[0] if known_sections else "Shot Details"
).strip()
_set_session_field_inclusion(normalized_field_name, include_in_form=True)
_sync_current_field_templates(
parameters_collection=parameters_collection,
mode=mode,
field_name=normalized_field_name,
section_name=section_value,
username=current_username(default=""),
add_to_selected=True,
include_in_form=True,
)
invalidate_custom_field_doc_cache()
refresh_dynamic_form_bindings()
flash(
f"Field {normalized_field_name} added to current Selected Fields and Organized Form.",
"success",
)
return redirect(url_for("new_parameter.new_parameter", next=return_url))
[docs]
@login_required
@blueprint.route(
"/new_parameter/toggle_current_templates/<field_name>",
methods=["POST"],
)
def toggle_field_in_current_templates(field_name):
"""Toggle one field on/off in the active campaign's Selected Fields and Organized Form."""
mode = session.get("mode", "shot")
parameters_db = current_app.config["MONGODB_COLLECTION_FOR_SETTINGS"]
db = get_db()
parameters_collection = db[parameters_db]
return_url = get_management_return_url(
request.values.get("next"),
url_for("field_selection.field_selection"),
)
normalized_field_name = str(field_name or "").strip()
if not normalized_field_name:
flash("Field name is required.", "error")
return redirect(url_for("new_parameter.new_parameter", next=return_url))
if normalized_field_name in _locked_management_fields(mode, parameters_collection):
flash(
f"Field {normalized_field_name} is locked and cannot be changed here.",
"error",
)
return redirect(url_for("new_parameter.new_parameter", next=return_url))
catalog_entry = _catalog_entry_for_field(mode, parameters_db, normalized_field_name)
if not catalog_entry:
flash(f"Field {normalized_field_name} was not found.", "error")
return redirect(url_for("new_parameter.new_parameter", next=return_url))
section_value = str(catalog_entry.get("section") or "").strip()
if not section_value:
section_value = _default_section_by_field(parameters_collection, mode).get(
normalized_field_name, ""
)
if not section_value:
known_sections = _resolve_section_names(parameters_collection, mode)
section_value = (
known_sections[0] if known_sections else "Shot Details"
).strip()
truthy_values = {"1", "true", "yes", "on"}
include_values = [
str(value).strip().lower() for value in request.form.getlist("include")
]
if not include_values:
include_values = [str(request.form.get("include", "0")).strip().lower()]
include_in_form = any(value in truthy_values for value in include_values)
_set_session_field_inclusion(normalized_field_name, include_in_form=include_in_form)
_sync_current_field_templates(
parameters_collection=parameters_collection,
mode=mode,
field_name=normalized_field_name,
section_name=section_value,
username=current_username(default=""),
add_to_selected=include_in_form,
include_in_form=include_in_form,
)
invalidate_custom_field_doc_cache()
refresh_dynamic_form_bindings()
flash(
(
f"Field {normalized_field_name} included in current form layout."
if include_in_form
else f"Field {normalized_field_name} removed from current form layout."
),
"success",
)
return redirect(url_for("new_parameter.new_parameter", next=return_url))