refactor(wizard): generic adapter handling with Literal types

- Add Literal type support to form_descriptors.py
  - Literal fields map to select widget
  - list[Literal] fields map to checkboxes widget
  - Options list extracted from Literal type args
- Update FIRMS adapter: satellites is now list[Literal[...]]
- Update USGS adapter: feed is now Literal[...]
- Refactor wizard to use wizard_order for adapter filtering
- Replace hardcoded adapter lists with dynamic discovery
- Remove _get_valid_satellites() and _get_valid_feeds() helpers
- Generic field parsing using describe_fields() pattern
- Update templates for generic widget rendering
- Add select/checkboxes widgets to adapters_edit.html
- Update tests for new widget types

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-05-19 00:38:06 +00:00
commit 08eb729979
8 changed files with 476 additions and 229 deletions

View file

@ -7,7 +7,7 @@ from collections.abc import AsyncIterator
from datetime import datetime, timezone
from io import StringIO
from pathlib import Path
from typing import Any
from typing import Any, Literal
import aiohttp
from tenacity import (
@ -54,7 +54,7 @@ SEVERITY_MAP = {
class FIRMSSettings(BaseModel):
"""Settings schema for FIRMS adapter."""
api_key_alias: str = "firms"
satellites: list[str] = ["VIIRS_SNPP_NRT", "VIIRS_NOAA20_NRT"]
satellites: list[Literal["VIIRS_SNPP_NRT", "VIIRS_NOAA20_NRT", "VIIRS_NOAA21_NRT"]] = ["VIIRS_SNPP_NRT", "VIIRS_NOAA20_NRT"]
region: RegionConfig | None = None

View file

@ -5,7 +5,7 @@ import sqlite3
from collections.abc import AsyncIterator
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from typing import Any, Literal
import aiohttp
from shapely.geometry import Point, box as shapely_box
@ -64,7 +64,7 @@ def magnitude_to_severity(mag: float) -> int:
class USGSQuakeSettings(BaseModel):
"""Settings schema for USGS quake adapter."""
feed: str = "all_hour"
feed: Literal["all_hour", "all_day", "all_week", "all_month"] = "all_hour"
region: RegionConfig | None = None

View file

@ -247,18 +247,37 @@ def _create_app() -> FastAPI:
except Exception:
pass
# Import helper functions for valid values
from central.gui.routes import _get_valid_satellites, _get_valid_feeds
# Add field descriptors to adapters
from central.gui.routes import _adapter_classes
from central.gui.form_descriptors import describe_fields
adapter_classes = _adapter_classes()
wizard_adapters = sorted(
[(name, cls) for name, cls in adapter_classes.items() if cls.wizard_order is not None],
key=lambda nc: nc[1].wizard_order
)
# Rebuild adapters with fields
enriched_adapters = []
for name, cls in wizard_adapters:
adapter_data = next((a for a in adapters if a["name"] == name), None)
if adapter_data:
settings_dict = adapter_data.get("settings", {})
fields = describe_fields(cls.settings_schema, settings_dict)
enriched_adapters.append({
"name": name,
"display_name": cls.display_name,
"enabled": adapter_data.get("enabled", False),
"cadence_s": adapter_data.get("cadence_s", 300),
"settings": settings_dict,
"fields": fields,
})
response = templates.TemplateResponse(
request=request,
name="setup_adapters.html",
context={
"csrf_token": csrf_token,
"adapters": adapters,
"adapters": enriched_adapters,
"api_keys": api_keys,
"valid_satellites": _get_valid_satellites(),
"valid_feeds": sorted(_get_valid_feeds()),
"tile_url": tile_url,
"tile_attribution": tile_attribution,
"error": error_msg,

View file

@ -4,8 +4,8 @@ If a second nested settings type beyond RegionConfig appears,
refactor this helper to recurse over nested models.
"""
from dataclasses import dataclass
from typing import Any, Union, get_args, get_origin
from dataclasses import dataclass, field
from typing import Any, Literal, Union, get_args, get_origin
from pydantic import BaseModel
from pydantic.fields import FieldInfo
@ -19,15 +19,30 @@ class FieldDescriptor:
"""Describes a form field for rendering."""
name: str
label: str
widget: str # "text", "number", "checkbox", "csv", "region"
widget: str # "text", "number", "checkbox", "csv", "select", "checkboxes", "region"
current_value: Any
default: Any
description: str
required: bool
options: list[str] | None = None # For select/checkboxes widgets
def _type_to_widget(field_name: str, field_type: type) -> str:
"""Map a Python type to a widget type."""
def _is_literal(tp: type) -> bool:
"""Check if a type is a Literal type."""
return get_origin(tp) is Literal
def _get_literal_values(tp: type) -> list[str]:
"""Extract the literal values from a Literal type."""
return list(get_args(tp))
def _type_to_widget_and_options(field_name: str, field_type: type) -> tuple[str, list[str] | None]:
"""Map a Python type to a widget type and optional options list.
Returns:
Tuple of (widget_type, options_list_or_none)
"""
# Handle Optional/Union types
origin = get_origin(field_type)
args = get_args(field_type)
@ -39,24 +54,38 @@ def _type_to_widget(field_name: str, field_type: type) -> str:
if non_none_args:
inner_type = non_none_args[0]
# Recursively determine widget for the inner type
return _type_to_widget(field_name, inner_type)
return _type_to_widget_and_options(field_name, inner_type)
# Check for Literal type (single select)
if _is_literal(field_type):
options = _get_literal_values(field_type)
return "select", [str(o) for o in options]
# Direct type checks
if field_type is str:
return "text"
return "text", None
if field_type is int:
return "number"
return "number", None
if field_type is bool:
return "checkbox"
return "checkbox", None
if field_type is RegionConfig:
return "region"
return "region", None
# Check for list[str]
# Check for list types
if origin is list:
if args and args[0] is str:
return "csv"
inner_type = args[0] if args else None
# list[Literal[...]] -> checkboxes
if inner_type is not None and _is_literal(inner_type):
options = _get_literal_values(inner_type)
return "checkboxes", [str(o) for o in options]
# list[str] -> csv
if inner_type is str:
return "csv", None
raise NotImplementedError(
f"Field '{field_name}' has unsupported list type: list[{args[0].__name__ if args else '?'}]"
f"Field '{field_name}' has unsupported list type: list[{inner_type.__name__ if inner_type else '?'}]"
)
# Check if it's a BaseModel subclass (nested model other than RegionConfig)
@ -98,8 +127,8 @@ def describe_fields(model_cls: type[BaseModel], current: dict) -> list[FieldDesc
# Get the field type
field_type = field_info.annotation
# Determine widget
widget = _type_to_widget(field_name, field_type)
# Determine widget and options
widget, options = _type_to_widget_and_options(field_name, field_type)
# Get current value, falling back to default
if field_name in current:
@ -128,6 +157,7 @@ def describe_fields(model_cls: type[BaseModel], current: dict) -> list[FieldDesc
default=default,
description=description,
required=required,
options=options,
))
return descriptors

View file

@ -73,18 +73,6 @@ ALIAS_REGEX = re.compile(r"^[a-zA-Z0-9_]+$")
EMAIL_REGEX = re.compile(r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
def _get_valid_satellites() -> list[str]:
"""Get valid satellite identifiers from firms adapter."""
from central.adapters.firms import SATELLITE_SHORT
return list(SATELLITE_SHORT.keys())
def _get_valid_feeds() -> set[str]:
"""Get valid feed values from usgs_quake adapter."""
from central.adapters.usgs_quake import VALID_FEEDS
return VALID_FEEDS
def _get_templates():
"""Get templates instance (deferred import to avoid circular)."""
from central.gui import templates
@ -647,18 +635,31 @@ async def setup_adapters_form(request: Request) -> HTMLResponse:
templates = _get_templates()
pool = get_pool()
# Get wizard adapters (filtered by wizard_order)
adapter_classes = _adapter_classes()
wizard_adapters = sorted(
[(name, cls) for name, cls in adapter_classes.items() if cls.wizard_order is not None],
key=lambda nc: nc[1].wizard_order
)
# Pre-fill from cookie state or DB defaults
if state.adapters:
adapters = []
for name in ["firms", "nws", "usgs_quake"]:
for name, cls in wizard_adapters:
if name in state.adapters:
a = state.adapters[name]
adapters.append({
"name": name,
"enabled": a["enabled"],
"cadence_s": a["cadence_s"],
"settings": a["settings"],
})
settings_dict = a["settings"]
else:
settings_dict = {}
fields = describe_fields(cls.settings_schema, settings_dict)
adapters.append({
"name": name,
"display_name": cls.display_name,
"enabled": a["enabled"] if name in state.adapters else False,
"cadence_s": a["cadence_s"] if name in state.adapters else 300,
"settings": settings_dict,
"fields": fields,
})
else:
async with pool.acquire() as conn:
rows = await conn.fetch(
@ -668,15 +669,28 @@ async def setup_adapters_form(request: Request) -> HTMLResponse:
ORDER BY name
"""
)
adapters = []
for row in rows:
settings_data = row["settings"] or {}
adapters.append({
"name": row["name"],
"enabled": row["enabled"],
"cadence_s": row["cadence_s"],
"settings": settings_data,
})
db_adapters = {row["name"]: row for row in rows}
adapters = []
for name, cls in wizard_adapters:
if name in db_adapters:
row = db_adapters[name]
settings_dict = row["settings"] or {}
enabled = row["enabled"]
cadence_s = row["cadence_s"]
else:
settings_dict = {}
enabled = False
cadence_s = 300
fields = describe_fields(cls.settings_schema, settings_dict)
adapters.append({
"name": name,
"display_name": cls.display_name,
"enabled": enabled,
"cadence_s": cadence_s,
"settings": settings_dict,
"fields": fields,
})
# Get API keys from wizard state (not DB)
api_keys = [{"alias": k["alias"]} for k in state.api_keys]
@ -701,8 +715,6 @@ async def setup_adapters_form(request: Request) -> HTMLResponse:
"csrf_token": csrf_token,
"adapters": adapters,
"api_keys": api_keys,
"valid_satellites": _get_valid_satellites(),
"valid_feeds": sorted(_get_valid_feeds()),
"tile_url": tile_url,
"tile_attribution": tile_attribution,
"error": None,
@ -755,7 +767,14 @@ async def setup_adapters_submit(request: Request) -> Response:
"settings": row["settings"] or {},
}
for adapter_name in ["firms", "nws", "usgs_quake"]:
# Get wizard adapters (filtered by wizard_order)
adapter_classes = _adapter_classes()
wizard_adapters = sorted(
[(name, cls) for name, cls in adapter_classes.items() if cls.wizard_order is not None],
key=lambda nc: nc[1].wizard_order
)
for adapter_name, adapter_cls in wizard_adapters:
current = current_adapters.get(adapter_name, {"enabled": False, "cadence_s": 300, "settings": {}})
current_settings = current.get("settings", {})
new_settings = dict(current_settings)
@ -777,73 +796,101 @@ async def setup_adapters_submit(request: Request) -> Response:
errors[f"{adapter_name}_cadence_s"] = "Cadence must be a valid integer"
cadence_s = current.get("cadence_s", 300)
# Adapter-specific validation
if adapter_name == "nws":
contact_email = form.get(f"{adapter_name}_contact_email", "").strip()
if enabled:
if not contact_email:
errors[f"{adapter_name}_contact_email"] = "Contact email is required when enabled"
elif not EMAIL_REGEX.match(contact_email):
errors[f"{adapter_name}_contact_email"] = "Invalid email format"
# Generic field parsing using describe_fields
fields = describe_fields(adapter_cls.settings_schema, current_settings)
for field in fields:
form_key = f"{adapter_name}_{field.name}"
if field.widget == "text":
value = form.get(form_key, "").strip()
# Special validation for contact_email
if field.name == "contact_email":
if enabled:
if not value:
errors[form_key] = "Contact email is required when enabled"
elif not EMAIL_REGEX.match(value):
errors[form_key] = "Invalid email format"
else:
new_settings[field.name] = value
else:
new_settings[field.name] = value if value else current_settings.get(field.name)
# Special validation for api_key_alias
elif field.name == "api_key_alias":
if value:
if not any(k["alias"] == value for k in state.api_keys):
errors[form_key] = "API key alias does not exist"
else:
new_settings[field.name] = value
else:
new_settings[field.name] = None
else:
new_settings["contact_email"] = contact_email
else:
new_settings["contact_email"] = contact_email if contact_email else current_settings.get("contact_email")
new_settings[field.name] = value if value else current_settings.get(field.name)
elif adapter_name == "firms":
api_key_alias = form.get(f"{adapter_name}_api_key_alias", "").strip()
satellites = form.getlist(f"{adapter_name}_satellites")
if api_key_alias:
# Validate against wizard state keys
if not any(k["alias"] == api_key_alias for k in state.api_keys):
errors[f"{adapter_name}_api_key_alias"] = f"API key alias does not exist"
elif field.widget == "number":
value_str = form.get(form_key, "").strip()
if value_str:
try:
new_settings[field.name] = int(value_str)
except ValueError:
errors[form_key] = f"{field.label} must be a valid number"
else:
new_settings["api_key_alias"] = api_key_alias
else:
new_settings["api_key_alias"] = None
new_settings[field.name] = current_settings.get(field.name)
# Validate satellites
valid_sats = set(_get_valid_satellites())
invalid_sats = [s for s in satellites if s not in valid_sats]
if invalid_sats:
errors[f"{adapter_name}_satellites"] = f"Invalid satellites: " + ", ".join(invalid_sats)
else:
new_settings["satellites"] = satellites
elif field.widget == "checkbox":
new_settings[field.name] = form_key in form
elif adapter_name == "usgs_quake":
feed = form.get(f"{adapter_name}_feed", "").strip()
valid_feeds = _get_valid_feeds()
if feed not in valid_feeds:
errors[f"{adapter_name}_feed"] = "Invalid feed"
else:
new_settings["feed"] = feed
elif field.widget == "csv":
value = form.get(form_key, "").strip()
if value:
new_settings[field.name] = [v.strip() for v in value.split(",") if v.strip()]
else:
new_settings[field.name] = []
# Region validation (all adapters)
region_north_str = form.get(f"{adapter_name}_region_north", "").strip()
region_south_str = form.get(f"{adapter_name}_region_south", "").strip()
region_east_str = form.get(f"{adapter_name}_region_east", "").strip()
region_west_str = form.get(f"{adapter_name}_region_west", "").strip()
elif field.widget == "select":
value = form.get(form_key, "").strip()
if value and field.options and value not in field.options:
errors[form_key] = f"Invalid {field.label.lower()}"
else:
new_settings[field.name] = value
try:
region_north = float(region_north_str)
region_south = float(region_south_str)
region_east = float(region_east_str)
region_west = float(region_west_str)
elif field.widget == "checkboxes":
# Use getlist for checkbox groups - absence means empty list
values = form.getlist(form_key)
if field.options:
invalid = [v for v in values if v not in field.options]
if invalid:
errors[form_key] = f"Invalid values: {', '.join(invalid)}"
else:
new_settings[field.name] = values
else:
new_settings[field.name] = values
if not (-90 <= region_south < region_north <= 90):
errors[f"{adapter_name}_region"] = "Invalid latitude: south < north, both -90 to 90"
elif not (-180 <= region_west < region_east <= 180):
errors[f"{adapter_name}_region"] = "Invalid longitude: west < east, both -180 to 180"
else:
new_settings["region"] = {
"north": region_north,
"south": region_south,
"east": region_east,
"west": region_west,
}
except ValueError:
errors[f"{adapter_name}_region"] = "Region coordinates must be valid numbers"
elif field.widget == "region":
# Region validation
region_north_str = form.get(f"{adapter_name}_{field.name}_north", "").strip()
region_south_str = form.get(f"{adapter_name}_{field.name}_south", "").strip()
region_east_str = form.get(f"{adapter_name}_{field.name}_east", "").strip()
region_west_str = form.get(f"{adapter_name}_{field.name}_west", "").strip()
try:
region_north = float(region_north_str)
region_south = float(region_south_str)
region_east = float(region_east_str)
region_west = float(region_west_str)
if not (-90 <= region_south < region_north <= 90):
errors[f"{adapter_name}_{field.name}"] = "Invalid latitude: south < north, both -90 to 90"
elif not (-180 <= region_west < region_east <= 180):
errors[f"{adapter_name}_{field.name}"] = "Invalid longitude: west < east, both -180 to 180"
else:
new_settings[field.name] = {
"north": region_north,
"south": region_south,
"east": region_east,
"west": region_west,
}
except ValueError:
errors[f"{adapter_name}_{field.name}"] = "Region coordinates must be valid numbers"
new_adapters[adapter_name] = {
"enabled": enabled,
@ -918,10 +965,20 @@ async def setup_finish_form(request: Request) -> HTMLResponse:
adapters = []
if state.adapters:
for name in ["firms", "nws", "usgs_quake"]:
adapter_classes = _adapter_classes()
wizard_adapters = sorted(
[(name, cls) for name, cls in adapter_classes.items() if cls.wizard_order is not None],
key=lambda nc: nc[1].wizard_order
)
for name, cls in wizard_adapters:
if name in state.adapters:
a = state.adapters[name]
adapters.append({"name": name, "enabled": a["enabled"], "cadence_s": a["cadence_s"]})
adapters.append({
"name": name,
"display_name": cls.display_name,
"enabled": a["enabled"],
"cadence_s": a["cadence_s"],
})
csrf_token, signed_token = reuse_or_generate_pre_auth_csrf(request, settings.csrf_secret)
response = templates.TemplateResponse(
@ -1441,6 +1498,24 @@ async def adapters_edit_submit(
parsed_values[field.name] = [v.strip() for v in raw.split(",") if v.strip()]
else:
parsed_values[field.name] = []
elif field.widget == "select":
value = raw.strip() if raw else None
if value and field.options and value not in field.options:
errors[field.name] = f"Invalid {field.label.lower()}"
else:
parsed_values[field.name] = value
elif field.widget == "checkboxes":
# Use getlist for checkbox groups
values = form.getlist(field.name)
form_data[field.name] = values # Override raw value
if field.options:
invalid = [v for v in values if v not in field.options]
if invalid:
errors[field.name] = f"Invalid values: {', '.join(invalid)}"
else:
parsed_values[field.name] = values
else:
parsed_values[field.name] = values
elif field.widget == "region":
# Region handled separately below
pass

View file

@ -100,6 +100,40 @@
{% if errors and errors[field.name] %}
<small style="color: var(--pico-color-red-500);">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "select" %}
<label for="{{ field.name }}">{{ field.label }}</label>
<select id="{{ field.name }}" name="{{ field.name }}">
{% for opt in field.options %}
<option value="{{ opt }}"
{% if (form_data[field.name] if form_data and field.name in form_data else field.current_value) == opt %}selected{% endif %}>
{{ opt }}
</option>
{% endfor %}
</select>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small style="color: var(--pico-color-red-500);">{{ errors[field.name] }}</small>
{% endif %}
{% elif field.widget == "checkboxes" %}
<label>{{ field.label }}</label>
{% set current_values = form_data.getlist(field.name) if form_data and form_data.getlist else (field.current_value or []) %}
{% for opt in field.options %}
<label style="display: inline-block; margin-right: 1rem;">
<input type="checkbox" name="{{ field.name }}" value="{{ opt }}"
{% if opt in current_values %}checked{% endif %}>
{{ opt }}
</label>
{% endfor %}
{% if field.description %}
<small style="display: block;">{{ field.description }}</small>
{% endif %}
{% if errors and errors[field.name] %}
<small style="color: var(--pico-color-red-500); display: block;">{{ errors[field.name] }}</small>
{% endif %}
{% endif %}
{% endfor %}
</fieldset>

View file

@ -29,7 +29,7 @@
{% for adapter in adapters %}
<details open style="margin-bottom: 2rem;">
<summary><strong>{{ adapter.name }}</strong></summary>
<summary><strong>{{ adapter.display_name or adapter.name }}</strong></summary>
<div style="padding: 1rem; border-left: 3px solid var(--pico-primary);">
<label>
@ -44,100 +44,154 @@
<label for="{{ adapter.name }}_cadence_s">Cadence (seconds)</label>
<input type="number" id="{{ adapter.name }}_cadence_s" name="{{ adapter.name }}_cadence_s"
value="{{ form_data.get(adapter.name + '_cadence_s') if form_data else adapter.cadence_s }}"
>
value="{{ form_data.get(adapter.name + '_cadence_s') if form_data else adapter.cadence_s }}">
{% if errors and errors.get(adapter.name + '_cadence_s') %}
<small style="color: var(--pico-color-red-500);">{{ errors[adapter.name + '_cadence_s'] }}</small>
{% endif %}
{% if adapter.name == 'nws' %}
<label for="{{ adapter.name }}_contact_email">Contact Email</label>
<input type="email" id="{{ adapter.name }}_contact_email" name="{{ adapter.name }}_contact_email"
value="{{ form_data.get(adapter.name + '_contact_email') if form_data else adapter.settings.contact_email }}">
{% if errors and errors.get(adapter.name + '_contact_email') %}
<small style="color: var(--pico-color-red-500);">{{ errors[adapter.name + '_contact_email'] }}</small>
{% endif %}
{% endif %}
{% for field in adapter.fields %}
{% set form_key = adapter.name + '_' + field.name %}
{% if adapter.name == 'firms' %}
<label for="{{ adapter.name }}_api_key_alias">API Key Alias</label>
<select id="{{ adapter.name }}_api_key_alias" name="{{ adapter.name }}_api_key_alias">
<option value="">(none)</option>
{% for key in api_keys %}
<option value="{{ key.alias }}"
{% if (form_data.get(adapter.name + '_api_key_alias') if form_data else adapter.settings.api_key_alias) == key.alias %}selected{% endif %}>
{{ key.alias }}
</option>
{% endfor %}
</select>
{% if errors and errors.get(adapter.name + '_api_key_alias') %}
<small style="color: var(--pico-color-red-500);">{{ errors[adapter.name + '_api_key_alias'] }}</small>
{% endif %}
{% if field.widget == "text" %}
{# Special handling for api_key_alias - render as select from wizard API keys #}
{% if field.name == "api_key_alias" %}
<label for="{{ form_key }}">{{ field.label }}</label>
<select id="{{ form_key }}" name="{{ form_key }}">
<option value="">(none)</option>
{% for key in api_keys %}
<option value="{{ key.alias }}"
{% if (form_data.get(form_key) if form_data else field.current_value) == key.alias %}selected{% endif %}>
{{ key.alias }}
</option>
{% endfor %}
</select>
{% else %}
<label for="{{ form_key }}">{{ field.label }}</label>
<input type="text" id="{{ form_key }}" name="{{ form_key }}"
value="{{ form_data.get(form_key) if form_data else field.current_value or '' }}"
{% if field.required %}required{% endif %}>
{% endif %}
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors.get(form_key) %}
<small style="color: var(--pico-color-red-500);">{{ errors[form_key] }}</small>
{% endif %}
<label>Satellites</label>
{% for sat in valid_satellites %}
<label style="display: inline-block; margin-right: 1rem;">
<input type="checkbox" name="{{ adapter.name }}_satellites" value="{{ sat }}"
{% if sat in (form_data.getlist(adapter.name + '_satellites') if form_data else adapter.settings.satellites or []) %}checked{% endif %}>
{{ sat }}
</label>
{% endfor %}
{% endif %}
{% elif field.widget == "number" %}
<label for="{{ form_key }}">{{ field.label }}</label>
<input type="number" id="{{ form_key }}" name="{{ form_key }}"
value="{{ form_data.get(form_key) if form_data else field.current_value or '' }}"
{% if field.required %}required{% endif %}>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors.get(form_key) %}
<small style="color: var(--pico-color-red-500);">{{ errors[form_key] }}</small>
{% endif %}
{% if adapter.name == 'usgs_quake' %}
<label for="{{ adapter.name }}_feed">Feed</label>
<select id="{{ adapter.name }}_feed" name="{{ adapter.name }}_feed">
{% for f in valid_feeds %}
<option value="{{ f }}"
{% if (form_data.get(adapter.name + '_feed') if form_data else adapter.settings.feed) == f %}selected{% endif %}>
{{ f }}
</option>
{% endfor %}
</select>
{% if errors and errors.get(adapter.name + '_feed') %}
<small style="color: var(--pico-color-red-500);">{{ errors[adapter.name + '_feed'] }}</small>
{% endif %}
{% endif %}
{% elif field.widget == "checkbox" %}
<label>
<input type="checkbox" name="{{ form_key }}"
{% if form_data and form_data.get(form_key) %}checked
{% elif not form_data and field.current_value %}checked{% endif %}>
{{ field.label }}
</label>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors.get(form_key) %}
<small style="color: var(--pico-color-red-500);">{{ errors[form_key] }}</small>
{% endif %}
<h4>Region</h4>
{% set region = form_data if form_data else adapter.settings.region %}
<div id="region-picker-{{ adapter.name }}"
data-adapter="{{ adapter.name }}"
data-north="{{ form_data.get(adapter.name + '_region_north') if form_data else (adapter.settings.region.north if adapter.settings.region else 49.5) }}"
data-south="{{ form_data.get(adapter.name + '_region_south') if form_data else (adapter.settings.region.south if adapter.settings.region else 31.0) }}"
data-east="{{ form_data.get(adapter.name + '_region_east') if form_data else (adapter.settings.region.east if adapter.settings.region else -102.0) }}"
data-west="{{ form_data.get(adapter.name + '_region_west') if form_data else (adapter.settings.region.west if adapter.settings.region else -124.5) }}"
data-tile-url="{{ tile_url }}"
data-tile-attr="{{ tile_attribution }}">
{% elif field.widget == "csv" %}
<label for="{{ form_key }}">{{ field.label }}</label>
<input type="text" id="{{ form_key }}" name="{{ form_key }}"
value="{{ form_data.get(form_key) if form_data else (field.current_value | join(',') if field.current_value else '') }}"
{% if field.required %}required{% endif %}>
<small>Comma-separated values{% if field.description %} — {{ field.description }}{% endif %}</small>
{% if errors and errors.get(form_key) %}
<small style="color: var(--pico-color-red-500);">{{ errors[form_key] }}</small>
{% endif %}
<div id="region-map-{{ adapter.name }}" style="height: 300px; margin-bottom: 1rem;"></div>
{% elif field.widget == "select" %}
<label for="{{ form_key }}">{{ field.label }}</label>
<select id="{{ form_key }}" name="{{ form_key }}">
{% for opt in field.options %}
<option value="{{ opt }}"
{% if (form_data.get(form_key) if form_data else field.current_value) == opt %}selected{% endif %}>
{{ opt }}
</option>
{% endfor %}
</select>
{% if field.description %}
<small>{{ field.description }}</small>
{% endif %}
{% if errors and errors.get(form_key) %}
<small style="color: var(--pico-color-red-500);">{{ errors[form_key] }}</small>
{% endif %}
<div class="grid">
<div>
<label>North</label>
<input type="number" name="{{ adapter.name }}_region_north" step="0.0001" min="-90" max="90" readonly
value="{{ form_data.get(adapter.name + '_region_north') if form_data else (adapter.settings.region.north if adapter.settings.region else 49.5) }}">
{% elif field.widget == "checkboxes" %}
<label>{{ field.label }}</label>
{% set current_values = form_data.getlist(form_key) if form_data else (field.current_value or []) %}
{% for opt in field.options %}
<label style="display: inline-block; margin-right: 1rem;">
<input type="checkbox" name="{{ form_key }}" value="{{ opt }}"
{% if opt in current_values %}checked{% endif %}>
{{ opt }}
</label>
{% endfor %}
{% if field.description %}
<small style="display: block;">{{ field.description }}</small>
{% endif %}
{% if errors and errors.get(form_key) %}
<small style="color: var(--pico-color-red-500); display: block;">{{ errors[form_key] }}</small>
{% endif %}
{% elif field.widget == "region" %}
<h4>Region</h4>
{% set region_key = adapter.name + '_' + field.name %}
{% set region = field.current_value or {} %}
<div id="region-picker-{{ adapter.name }}"
data-adapter="{{ adapter.name }}"
data-field="{{ field.name }}"
data-north="{{ form_data.get(region_key + '_north') if form_data else (region.north if region else 49.5) }}"
data-south="{{ form_data.get(region_key + '_south') if form_data else (region.south if region else 31.0) }}"
data-east="{{ form_data.get(region_key + '_east') if form_data else (region.east if region else -102.0) }}"
data-west="{{ form_data.get(region_key + '_west') if form_data else (region.west if region else -124.5) }}"
data-tile-url="{{ tile_url }}"
data-tile-attr="{{ tile_attribution }}">
<div id="region-map-{{ adapter.name }}" style="height: 300px; margin-bottom: 1rem;"></div>
<div class="grid">
<div>
<label>North</label>
<input type="number" name="{{ region_key }}_north" step="0.0001" min="-90" max="90" readonly
value="{{ form_data.get(region_key + '_north') if form_data else (region.north if region else 49.5) }}">
</div>
<div>
<label>South</label>
<input type="number" name="{{ region_key }}_south" step="0.0001" min="-90" max="90" readonly
value="{{ form_data.get(region_key + '_south') if form_data else (region.south if region else 31.0) }}">
</div>
<div>
<label>East</label>
<input type="number" name="{{ region_key }}_east" step="0.0001" min="-180" max="180" readonly
value="{{ form_data.get(region_key + '_east') if form_data else (region.east if region else -102.0) }}">
</div>
<div>
<label>West</label>
<input type="number" name="{{ region_key }}_west" step="0.0001" min="-180" max="180" readonly
value="{{ form_data.get(region_key + '_west') if form_data else (region.west if region else -124.5) }}">
</div>
</div>
{% if errors and errors.get(region_key) %}
<small style="color: var(--pico-color-red-500);">{{ errors[region_key] }}</small>
{% endif %}
</div>
<div>
<label>South</label>
<input type="number" name="{{ adapter.name }}_region_south" step="0.0001" min="-90" max="90" readonly
value="{{ form_data.get(adapter.name + '_region_south') if form_data else (adapter.settings.region.south if adapter.settings.region else 31.0) }}">
</div>
<div>
<label>East</label>
<input type="number" name="{{ adapter.name }}_region_east" step="0.0001" min="-180" max="180" readonly
value="{{ form_data.get(adapter.name + '_region_east') if form_data else (adapter.settings.region.east if adapter.settings.region else -102.0) }}">
</div>
<div>
<label>West</label>
<input type="number" name="{{ adapter.name }}_region_west" step="0.0001" min="-180" max="180" readonly
value="{{ form_data.get(adapter.name + '_region_west') if form_data else (adapter.settings.region.west if adapter.settings.region else -124.5) }}">
</div>
</div>
{% if errors and errors.get(adapter.name + '_region') %}
<small style="color: var(--pico-color-red-500);">{{ errors[adapter.name + '_region'] }}</small>
{% endif %}
</div>
{% endfor %}
</div>
</details>
{% endfor %}
@ -151,11 +205,12 @@
<script>
document.addEventListener('DOMContentLoaded', function() {
const adapters = ['nws', 'firms', 'usgs_quake'];
// Find all region pickers dynamically
const regionPickers = document.querySelectorAll('[id^="region-picker-"]');
adapters.forEach(function(adapterName) {
const container = document.getElementById('region-picker-' + adapterName);
if (!container) return;
regionPickers.forEach(function(container) {
const adapterName = container.dataset.adapter;
const fieldName = container.dataset.field || 'region';
const savedNorth = parseFloat(container.dataset.north);
const savedSouth = parseFloat(container.dataset.south);
@ -215,10 +270,11 @@ document.addEventListener('DOMContentLoaded', function() {
rectangle.editing.enable();
const northInput = container.querySelector('input[name="' + adapterName + '_region_north"]');
const southInput = container.querySelector('input[name="' + adapterName + '_region_south"]');
const eastInput = container.querySelector('input[name="' + adapterName + '_region_east"]');
const westInput = container.querySelector('input[name="' + adapterName + '_region_west"]');
const inputPrefix = adapterName + '_' + fieldName;
const northInput = container.querySelector('input[name="' + inputPrefix + '_north"]');
const southInput = container.querySelector('input[name="' + inputPrefix + '_south"]');
const eastInput = container.querySelector('input[name="' + inputPrefix + '_east"]');
const westInput = container.querySelector('input[name="' + inputPrefix + '_west"]');
function updateInputs() {
const b = rectangle.getBounds();

View file

@ -4,7 +4,7 @@ import pytest
from pydantic import BaseModel
from typing import Optional
from central.gui.form_descriptors import describe_fields, FieldDescriptor, _type_to_widget
from central.gui.form_descriptors import describe_fields, FieldDescriptor, _type_to_widget_and_options
from central.config_models import RegionConfig
@ -33,39 +33,39 @@ class SettingsWithRegion(BaseModel):
class TestTypeToWidget:
"""Tests for _type_to_widget function."""
"""Tests for _type_to_widget_and_options function."""
def test_str_maps_to_text(self):
assert _type_to_widget("field", str) == "text"
assert _type_to_widget_and_options("field", str) == ("text", None)
def test_int_maps_to_number(self):
assert _type_to_widget("field", int) == "number"
assert _type_to_widget_and_options("field", int) == ("number", None)
def test_bool_maps_to_checkbox(self):
assert _type_to_widget("field", bool) == "checkbox"
assert _type_to_widget_and_options("field", bool) == ("checkbox", None)
def test_list_str_maps_to_csv(self):
assert _type_to_widget("field", list[str]) == "csv"
assert _type_to_widget_and_options("field", list[str]) == ("csv", None)
def test_region_config_maps_to_region(self):
assert _type_to_widget("field", RegionConfig) == "region"
assert _type_to_widget_and_options("field", RegionConfig) == ("region", None)
def test_optional_region_maps_to_region(self):
assert _type_to_widget("field", Optional[RegionConfig]) == "region"
assert _type_to_widget_and_options("field", Optional[RegionConfig]) == ("region", None)
def test_optional_str_maps_to_text(self):
"""Optional[str] should map to text widget."""
assert _type_to_widget("field", Optional[str]) == "text"
assert _type_to_widget_and_options("field", Optional[str]) == ("text", None)
def test_optional_int_maps_to_number(self):
"""Optional[int] should map to number widget."""
assert _type_to_widget("field", Optional[int]) == "number"
assert _type_to_widget_and_options("field", Optional[int]) == ("number", None)
def test_unsupported_type_raises(self):
class CustomType:
pass
with pytest.raises(NotImplementedError):
_type_to_widget("field", CustomType)
_type_to_widget_and_options("field", CustomType)
class TestDescribeFields:
@ -172,15 +172,17 @@ class TestRealAdapterSchemas:
fields = describe_fields(FIRMSSettings, {
"api_key_alias": "firms_key",
"satellites": ["VIIRS_SNPP"]
"satellites": ["VIIRS_SNPP_NRT"]
})
key_field = next(f for f in fields if f.name == "api_key_alias")
assert key_field.widget == "text"
sat_field = next(f for f in fields if f.name == "satellites")
assert sat_field.widget == "csv"
assert sat_field.current_value == ["VIIRS_SNPP"]
assert sat_field.widget == "checkboxes"
assert sat_field.current_value == ["VIIRS_SNPP_NRT"]
assert sat_field.options is not None
assert "VIIRS_SNPP_NRT" in sat_field.options
def test_usgs_quake_settings(self):
"""USGSQuakeSettings generates correct field descriptors."""
@ -189,8 +191,11 @@ class TestRealAdapterSchemas:
fields = describe_fields(USGSQuakeSettings, {"feed": "all_hour"})
feed_field = next(f for f in fields if f.name == "feed")
assert feed_field.widget == "text"
assert feed_field.widget == "select"
assert feed_field.current_value == "all_hour"
assert feed_field.options is not None
assert "all_hour" in feed_field.options
assert "all_day" in feed_field.options
def test_all_adapters_have_region_field(self):
"""All adapter settings schemas include region field."""
@ -203,3 +208,31 @@ class TestRealAdapterSchemas:
region_field = next((f for f in fields if f.name == "region"), None)
assert region_field is not None, f"{schema.__name__} should have region field"
assert region_field.widget == "region"
class TestLiteralTypes:
"""Tests for Literal type support."""
def test_literal_maps_to_select(self):
"""Literal type maps to select widget with options."""
from typing import Literal
widget, options = _type_to_widget_and_options("field", Literal["a", "b", "c"])
assert widget == "select"
assert options == ["a", "b", "c"]
def test_list_literal_maps_to_checkboxes(self):
"""list[Literal] maps to checkboxes widget with options."""
from typing import Literal
widget, options = _type_to_widget_and_options("field", list[Literal["x", "y", "z"]])
assert widget == "checkboxes"
assert options == ["x", "y", "z"]
def test_optional_literal_maps_to_select(self):
"""Optional[Literal] maps to select widget."""
from typing import Literal, Optional
widget, options = _type_to_widget_and_options("field", Optional[Literal["one", "two"]])
assert widget == "select"
assert options == ["one", "two"]