Files
KSP_project/app/routes/web.py
T

4005 lines
152 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation
from difflib import SequenceMatcher
from math import ceil
from pathlib import Path
import re
from typing import Any
from flask import Blueprint, abort, flash, has_request_context, redirect, render_template, request, session, url_for
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import selectinload
from app.extensions import db
from app.models import Asset, AssetLogEntry, AssetStateNode, CommunicationPart, EngineFamily, EngineVariant, TankSpec, VehicleCost
from app.services.fuel_conversion import list_fuel_factors
from app.services.log_book_importer import import_log_book_data
web_bp = Blueprint("web", __name__)
DEFAULT_PAGE_SIZE = 30
CUSTOM_OPTION_VALUE = "__custom__"
LOG_BOOK_PATH = Path(__file__).resolve().parents[2] / "log_book.xlsx"
SIMULATION_TIME_SESSION_KEY = "simulation_time"
ENGINE_SORT_OPTIONS = [
{"value": "engine_name_asc", "label": "名称 A-Z"},
{"value": "engine_name_desc", "label": "名称 Z-A"},
{"value": "mass_t_asc", "label": "重量从低到高"},
{"value": "mass_t_desc", "label": "重量从高到低"},
{"value": "vac_thrust_asc", "label": "真空推力从低到高"},
{"value": "vac_thrust_desc", "label": "真空推力从高到低"},
{"value": "sl_thrust_asc", "label": "海平面推力从低到高"},
{"value": "sl_thrust_desc", "label": "海平面推力从高到低"},
{"value": "vac_isp_asc", "label": "真空比冲从低到高"},
{"value": "vac_isp_desc", "label": "真空比冲从高到低"},
{"value": "sl_isp_asc", "label": "海平面比冲从低到高"},
{"value": "sl_isp_desc", "label": "海平面比冲从高到低"},
]
ENGINE_SORT_COLUMNS = [
{"key": "engine_name", "label": "Engine"},
{"key": "mass_t", "label": "Mass"},
{"key": "sl_thrust", "label": "SL Thrust"},
{"key": "vac_thrust", "label": "Vac Thrust"},
{"key": "sl_isp", "label": "SL ISP"},
{"key": "vac_isp", "label": "Vac ISP"},
]
COMM_SORT_OPTIONS = [
{"value": "part_name", "label": "名称 A-Z"},
{"value": "range_desc", "label": "通信距离从高到低"},
{"value": "mass_desc", "label": "质量从高到低"},
]
TANK_SORT_OPTIONS = [
{"value": "tank_name", "label": "名称 A-Z"},
{"value": "volume_desc", "label": "容积从高到低"},
{"value": "ratio_desc", "label": "质量比从高到低"},
]
VEHICLE_SORT_OPTIONS = [
{"value": "vehicle_name", "label": "名称 A-Z"},
{"value": "launch_price_desc", "label": "发射成本从高到低"},
{"value": "launch_price_asc", "label": "发射成本从低到高"},
]
ASSET_SORT_OPTIONS = [
{"value": "name_asc", "label": "名称 A-Z"},
{"value": "asset_type_asc", "label": "类型 A-Z"},
{"value": "updated_desc", "label": "最近更新优先"},
]
MISSION_PREVIEW_TIME = "2060-03-12 09:00"
MISSION_PREVIEW_METRICS = [
{"label": "Tracked Assets", "value": "14"},
{"label": "Active Missions", "value": "6"},
{"label": "Upcoming Events (30d)", "value": "9"},
]
MISSION_PREVIEW_FILTERS = {
"simulation_time": MISSION_PREVIEW_TIME,
"asset_type": "All Asset Types",
"location": "Earth System",
"record_scope": "Confirmed + Planned",
}
MISSION_BOARD_RECORD_SCOPE_OPTIONS = [
{"value": "all", "label": "全部资产"},
{"value": "active", "label": "仅活动中"},
{"value": "active_planned", "label": "活动中 + 计划中"},
{"value": "recorded", "label": "仅已记录"},
]
TIMELINE_SCALE_OPTIONS = [
{"value": "day", "label": "按日"},
{"value": "month", "label": "按月"},
{"value": "year", "label": "按年"},
]
MISSION_STATUS_GROUPS = [
{
"title": "In Transit",
"accent": "warning",
"items": [
{
"name": "XH-02 Taibai",
"asset_type": "Exploration Mothership",
"location": "Earth to Mars Transfer",
"mission": "Support Mars One Campus",
"state_window": "2060-02-26 to 2060-03-29",
"next_event": "Mars Arrival · 2060-03-29",
"summary": "31-day transfer, planned delta-v margin 105 km/s.",
"record_status": "Confirmed",
},
{
"name": "Amalthea Crew Vehicle 4",
"asset_type": "Shuttle",
"location": "LKO to Lingxiao Station",
"mission": "Expedition 16 Crew Rotation",
"state_window": "2060-03-28 to 2060-03-30",
"next_event": "Docking Window · 2060-03-30",
"summary": "10 crew manifest uploaded, rendezvous burn pending.",
"record_status": "Planned",
},
],
},
{
"title": "Docked / On Station",
"accent": "primary",
"items": [
{
"name": "Lingxiao Station",
"asset_type": "Station",
"location": "Low Kerbin Orbit",
"mission": "Continuous habitation and logistics hub",
"state_window": "2059-07-23 to present",
"next_event": "Expedition 16 Crew Arrival · 2060-03-30",
"summary": "Expedition 15 is onboard, crew handover prep in progress.",
"record_status": "Active",
},
{
"name": "Tianhe Cargo Pod 7",
"asset_type": "Cargo Shuttle",
"location": "Lingxiao Forward Berth",
"mission": "Dry cargo replenishment",
"state_window": "2060-03-01 to 2060-03-18",
"next_event": "Undock · 2060-03-18",
"summary": "Food, station spares, and EVA consumables unloaded.",
"record_status": "Confirmed",
},
],
},
{
"title": "At Port / Surface Ops",
"accent": "neutral",
"items": [
{
"name": "XH-03 Changxi",
"asset_type": "Exploration Mothership",
"location": "Star Port",
"mission": "Refit for Jupiter relay deployment",
"state_window": "2060-01-08 to 2060-04-12",
"next_event": "Rollout Review · 2060-04-12",
"summary": "Main engine refurbishment at bay 4, payload integration queued.",
"record_status": "Confirmed",
},
{
"name": "Mars One Campus Alpha",
"asset_type": "Surface Outpost",
"location": "Mars Surface",
"mission": "Habitat expansion and ISRU staging",
"state_window": "2059-10-20 to present",
"next_event": "Taibai Supply Window · 2060-03-29",
"summary": "Surface crew awaiting structural truss shipment and greenhouse kit.",
"record_status": "Active",
},
],
},
]
MISSION_LOCATION_GROUPS = [
{
"title": "Earth System",
"subtitle": "Ports, orbital stations, and nearby transfers",
"items": [
{
"name": "Lingxiao Station",
"asset_type": "Station",
"state": "Docked / On Station",
"mission": "Crew rotation and logistics hub",
"location": "Low Kerbin Orbit",
"next_event": "Crew Arrival · 2060-03-30",
},
{
"name": "XH-03 Changxi",
"asset_type": "Exploration Mothership",
"state": "At Port",
"mission": "Refit for Jupiter relay deployment",
"location": "Star Port",
"next_event": "Rollout Review · 2060-04-12",
},
{
"name": "Amalthea Crew Vehicle 4",
"asset_type": "Shuttle",
"state": "In Transit",
"mission": "Expedition 16 Crew Rotation",
"location": "LKO Transfer Corridor",
"next_event": "Docking Window · 2060-03-30",
},
],
},
{
"title": "Mars System",
"subtitle": "Orbital assets and surface support network",
"items": [
{
"name": "Mars One Campus Alpha",
"asset_type": "Surface Outpost",
"state": "Surface Ops",
"mission": "Habitat expansion and ISRU staging",
"location": "Mars Surface",
"next_event": "Taibai Supply Window · 2060-03-29",
},
{
"name": "XH-02 Taibai",
"asset_type": "Exploration Mothership",
"state": "In Transit",
"mission": "Support Mars One Campus",
"location": "Mars Transfer Lane",
"next_event": "Mars Arrival · 2060-03-29",
},
],
},
]
MISSION_UPCOMING_EVENTS = [
{"when": "2060-03-18", "title": "Tianhe Cargo Pod 7 undock from Lingxiao Station", "scope": "Lingxiao Station"},
{"when": "2060-03-29", "title": "XH-02 Taibai arrives at Mars", "scope": "Mars System"},
{"when": "2060-03-30", "title": "Expedition 16 crew arrives via Amalthea Crew Vehicle", "scope": "Lingxiao Station"},
{"when": "2060-04-02", "title": "Expedition 15 crew departs Lingxiao Station", "scope": "Lingxiao Station"},
]
MISSION_LOG_HEADER = {
"name": "XH-02 Taibai",
"asset_type": "Exploration Mothership",
"program": "Mars Expedition Program",
"current_state": "In Transit",
"current_location": "Earth to Mars Transfer",
"current_mission": "Support Mars One Campus",
"state_window": "2060-02-26 06:00 to 2060-03-29 18:00",
"next_event": "Mars Arrival · 2060-03-29 18:00",
}
MISSION_LOG_ENTRIES = [
{
"mode": "event",
"timestamp": "2058-03-15 10:20",
"title": "First module delivered to Star Port",
"location": "Star Port",
"record_status": "Completed",
"summary": "Core transfer spine and habitat ring unloaded for final assembly.",
},
{
"mode": "event",
"timestamp": "2059-08-06 09:00",
"title": "Commissioned",
"location": "Star Port",
"record_status": "Completed",
"summary": "Flight certification signed for Mars transport duties.",
},
{
"mode": "state",
"timestamp": "2059-08-08 04:10",
"end_timestamp": "2059-10-17 13:40",
"title": "Outbound transfer carrying Mars Expedition 3",
"location": "Earth to Mars Transfer",
"record_status": "Completed",
"summary": "Departure from Earth with full crew and descent stack attached.",
},
{
"mode": "event",
"timestamp": "2059-10-17 13:40",
"title": "Mars Arrival",
"location": "Mars Orbit",
"record_status": "Completed",
"summary": "Insertion burn nominal. Crew descent staging began next day.",
},
{
"mode": "state",
"timestamp": "2059-10-18 08:00",
"end_timestamp": "2059-10-19 06:10",
"title": "Mars orbit standby and crew pickup window",
"location": "Mars Orbit",
"record_status": "Completed",
"summary": "Received Mars Expedition 1 return crew after descent vehicle separation.",
},
{
"mode": "state",
"timestamp": "2059-10-19 06:10",
"end_timestamp": "2059-12-25 22:30",
"title": "Return transfer from Mars to Earth",
"location": "Mars to Earth Transfer",
"record_status": "Completed",
"summary": "Return trajectory with Expedition 1 crew aboard.",
},
{
"mode": "state",
"timestamp": "2060-02-26 06:00",
"end_timestamp": "2060-03-29 18:00",
"title": "Support mission transfer to Mars One Campus",
"location": "Earth to Mars Transfer",
"record_status": "Active",
"summary": "Carrying structural truss and surface greenhouse kit; 105 km/s, 31-day transition.",
},
{
"mode": "event",
"timestamp": "2060-03-29 18:00",
"title": "Mars Arrival",
"location": "Mars Orbit",
"record_status": "Planned",
"summary": "Planned capture burn and cargo handoff to surface fleet.",
},
]
MISSION_TIMELINE_ROWS = [
{
"name": "XH-02 Taibai",
"asset_type": "Exploration Mothership",
"segments": [
{"label": "Transit to Mars", "start": 6, "span": 22, "tone": "warning"},
{"label": "Mars Orbit", "start": 29, "span": 6, "tone": "primary"},
{"label": "Return Transit", "start": 38, "span": 21, "tone": "neutral"},
{"label": "Port Stay", "start": 61, "span": 8, "tone": "soft"},
{"label": "Support Transit", "start": 73, "span": 22, "tone": "warning current"},
],
"events": [
{"label": "Commissioned", "offset": 1},
{"label": "Mars Arrival", "offset": 28},
{"label": "Earth Arrival", "offset": 59},
{"label": "Next Mars Arrival", "offset": 95},
],
},
{
"name": "Lingxiao Station",
"asset_type": "Station",
"segments": [
{"label": "Continuous Ops", "start": 0, "span": 96, "tone": "primary current"},
],
"events": [
{"label": "Crew 9 Arrival", "offset": 10},
{"label": "Expedition 15 Arrival", "offset": 62},
{"label": "Expedition 16 Arrival", "offset": 88},
],
},
{
"name": "Mars One Campus Alpha",
"asset_type": "Surface Outpost",
"segments": [
{"label": "Surface Expansion", "start": 34, "span": 62, "tone": "neutral current"},
],
"events": [
{"label": "First Crew Descent", "offset": 37},
{"label": "Taibai Supply Window", "offset": 95},
],
},
]
MISSION_TIMELINE_SCALE = [
"2059-08",
"2059-09",
"2059-10",
"2059-11",
"2059-12",
"2060-01",
"2060-02",
"2060-03",
]
def _mission_preview_context() -> dict[str, object]:
return {
"simulation_time": MISSION_PREVIEW_TIME,
"metrics": MISSION_PREVIEW_METRICS,
"filters": MISSION_PREVIEW_FILTERS,
"upcoming_events": MISSION_UPCOMING_EVENTS,
}
FAMILY_FORM_FIELDS = [
{"name": "engine_name", "label": "Engine", "kind": "text", "required": True},
{"name": "part_name", "label": "Part Name", "kind": "text", "required": False},
{"name": "cycle", "label": "Cycle", "kind": "combo", "required": False},
{"name": "size_m", "label": "Size (m)", "kind": "decimal", "required": False},
{"name": "entry_cost", "label": "Entry Cost", "kind": "integer", "required": False},
]
VARIANT_FORM_FIELDS = [
{"name": "fuel_type", "label": "Fuel Type", "kind": "combo", "required": True},
{"name": "config_name", "label": "Config Name", "kind": "text", "required": False},
{"name": "work_env", "label": "Work Env", "kind": "text", "required": False},
{"name": "sl_isp", "label": "SL ISP", "kind": "decimal", "required": False},
{"name": "vac_isp", "label": "Vac ISP", "kind": "decimal", "required": False},
{"name": "min_thrust_kn", "label": "Min Thrust (kN)", "kind": "decimal", "required": False},
{"name": "max_thrust_kn", "label": "Max Thrust (kN)", "kind": "decimal", "required": False},
{"name": "mass_t", "label": "Mass (t)", "kind": "decimal", "required": False},
{"name": "twr", "label": "TWR", "kind": "decimal", "required": False},
{"name": "throttle_ratio", "label": "Throttle Ratio", "kind": "decimal", "required": False},
{"name": "tvc_deg", "label": "TVC (deg)", "kind": "decimal", "required": False},
{"name": "price_kd", "label": "Price (KD)", "kind": "integer", "required": False},
{"name": "ignitions", "label": "Ignitions", "kind": "integer", "required": False},
{
"name": "has_unlimited_ignitions",
"label": "Unlimited Ignitions",
"kind": "checkbox",
"required": False,
},
{"name": "mod_source", "label": "Mod Source", "kind": "combo", "required": False},
{"name": "tech_required", "label": "Tech Required", "kind": "text", "required": False},
{"name": "engine_note", "label": "Engine Note", "kind": "textarea", "required": False},
{"name": "config_note", "label": "Config Note", "kind": "textarea", "required": False},
]
VARIANT_CREATE_FIELDS = [
{"name": "fuel_type", "label": "Fuel Type", "kind": "combo", "required": True},
{"name": "config_name", "label": "Config Name", "kind": "text", "required": False},
{"name": "work_env", "label": "Work Env", "kind": "text", "required": False},
{"name": "sl_isp", "label": "SL ISP", "kind": "decimal", "required": False},
{"name": "vac_isp", "label": "Vac ISP", "kind": "decimal", "required": False},
{"name": "max_thrust_kn", "label": "Max Thrust (kN)", "kind": "decimal", "required": False},
{"name": "mod_source", "label": "Mod Source", "kind": "combo", "required": False},
]
COMMUNICATION_FIELDS = [
{"name": "part_name", "label": "Part", "kind": "text", "required": True},
{"name": "display_name", "label": "Part Showname", "kind": "text", "required": False},
{"name": "antenna_type", "label": "Type", "kind": "text", "required": False},
{"name": "source", "label": "Source", "kind": "text", "required": False},
{"name": "mass_t", "label": "Mass (t)", "kind": "decimal", "required": False},
{"name": "range_km", "label": "Range km", "kind": "decimal", "required": False},
{"name": "angle_deg", "label": "Angle", "kind": "decimal", "required": False},
{"name": "idle_power_watt", "label": "Idle Power (W)", "kind": "decimal", "required": False},
{
"name": "transmitting_power_watt",
"label": "Transmitting Power (W)",
"kind": "decimal",
"required": False,
},
{"name": "entry_cost", "label": "Entry Cost", "kind": "integer", "required": False},
{"name": "cost", "label": "Cost", "kind": "integer", "required": False},
{"name": "is_active", "label": "Is Active", "kind": "checkbox", "required": False},
{
"name": "is_deployable",
"label": "Is Deployable",
"kind": "checkbox",
"required": False,
},
{"name": "is_feeder", "label": "Is Feeder", "kind": "checkbox", "required": False},
{"name": "description", "label": "Description", "kind": "textarea", "required": False},
{"name": "note", "label": "Note", "kind": "textarea", "required": False},
]
TANK_FIELDS = [
{"name": "tank_name", "label": "Tank", "kind": "text", "required": True},
{"name": "fuel_type", "label": "Fuel Type", "kind": "text", "required": False},
{"name": "vehicle_name", "label": "Vehicle", "kind": "text", "required": False},
{"name": "source", "label": "Source", "kind": "text", "required": False},
{"name": "dry_mass_t", "label": "Dry Mass", "kind": "decimal", "required": False},
{"name": "fuel_mass_t", "label": "Fuel Mass", "kind": "decimal", "required": False},
{"name": "wet_mass_t", "label": "Wet Mass", "kind": "decimal", "required": False},
{"name": "tank_volume_l", "label": "Tank Volume (L)", "kind": "decimal", "required": False},
{"name": "mass_ratio", "label": "Mass Ratio", "kind": "decimal", "required": False},
{
"name": "kiloliters_per_ton",
"label": "KL per ton",
"kind": "decimal",
"required": False,
},
{"name": "note", "label": "Note", "kind": "textarea", "required": False},
]
VEHICLE_FIELDS = [
{"name": "vehicle_name", "label": "Vehicle", "kind": "text", "required": True},
{"name": "launch_price", "label": "Launch Price", "kind": "integer", "required": False},
{"name": "source", "label": "Source", "kind": "text", "required": False},
]
ASSET_FIELDS = [
{"name": "name", "label": "Asset", "kind": "text", "required": True},
{"name": "asset_type", "label": "Asset Type", "kind": "combo", "required": True},
{"name": "is_retired", "label": "Retired", "kind": "checkbox", "required": False},
{"name": "program", "label": "Program", "kind": "text", "required": False},
{"name": "home_region", "label": "Home Region", "kind": "text", "required": False},
{"name": "note", "label": "Note", "kind": "textarea", "required": False},
]
ASSET_TYPE_BASE_OPTIONS = (
"Shuttle",
"Station",
"Surface Outpost",
"Exploration Mothership",
)
ASSET_ENTRY_KIND_OPTIONS = [
{"value": "event", "label": "Event Point"},
{"value": "state", "label": "State Interval"},
]
PROJECT_METRICS = [
{"label": "Engine rows", "value": "478"},
{"label": "Communication rows", "value": "50"},
{"label": "Tank rows", "value": "27"},
{"label": "Vehicle cost rows", "value": "10"},
]
MODULES = [
{
"title": "Engine catalog",
"status": "live",
"description": "Core engine families and variants with filters, pagination, create/delete, and editable variants.",
"endpoint": "web.engine_list",
},
{
"title": "Communication parts",
"status": "live",
"description": "Paginated communication part list with source/type filters and edit forms.",
"endpoint": "web.communication_list",
},
{
"title": "Tank specs",
"status": "live",
"description": "Paginated tank list with fuel/source filters and edit forms.",
"endpoint": "web.tank_list",
},
{
"title": "Vehicle costs",
"status": "live",
"description": "Paginated vehicle cost list with sorting and edit forms.",
"endpoint": "web.vehicle_list",
},
{
"title": "Mission assets",
"status": "beta",
"description": "Asset CRUD with real asset log entries and simulation-time-derived current state.",
"endpoint": "web.asset_list",
},
{
"title": "Fuel converter",
"status": "live",
"description": "A working liters-to-tons and tons-to-liters converter based on the workbook coefficients.",
"endpoint": "web.fuel_converter",
},
]
@dataclass(slots=True)
class EngineCatalogRow:
family: EngineFamily
config_count: int
fuel_types: tuple[str, ...]
mod_sources: tuple[str, ...]
mass_t: Decimal | None
twr: Decimal | None
vac_thrust_kn: Decimal | None
sl_thrust_kn: Decimal | None
vac_isp: Decimal | None
sl_isp: Decimal | None
def _engine_family_statement() -> object:
return select(EngineFamily).options(selectinload(EngineFamily.variants))
def _load_model_or_404(model: type[Any], object_id: object) -> Any:
instance = db.session.get(model, object_id)
if instance is None:
abort(404)
return instance
def _load_engine_family_or_404(family_id: object) -> EngineFamily:
statement = _engine_family_statement().where(EngineFamily.id == family_id)
family = db.session.execute(statement).scalar_one_or_none()
if family is None:
abort(404)
return family
def _normalized_text(raw_value: str | None) -> str | None:
if raw_value is None:
return None
cleaned = raw_value.strip()
return cleaned or None
def _parse_decimal(raw_value: str | None, label: str) -> Decimal | None:
cleaned = _normalized_text(raw_value)
if cleaned is None:
return None
try:
return Decimal(cleaned)
except InvalidOperation as exc:
raise ValueError(f"{label} 必须是数值") from exc
def _parse_integer(raw_value: str | None, label: str) -> int | None:
cleaned = _normalized_text(raw_value)
if cleaned is None:
return None
try:
decimal_value = Decimal(cleaned)
except InvalidOperation as exc:
raise ValueError(f"{label} 必须是整数") from exc
if decimal_value != decimal_value.to_integral_value():
raise ValueError(f"{label} 必须是整数")
return int(decimal_value)
def _as_utc(value: datetime | None) -> datetime | None:
if value is None:
return None
if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc)
return value.astimezone(timezone.utc)
def _parse_datetime(raw_value: str | None, label: str) -> datetime | None:
cleaned = _normalized_text(raw_value)
if cleaned is None:
return None
try:
parsed = datetime.fromisoformat(cleaned.replace(" ", "T"))
except ValueError as exc:
normalized = (
cleaned.replace("年", "/")
.replace("月", "/")
.replace("日", " ")
.replace("T", " ")
)
match = re.fullmatch(r"(\d{4})[/-](\d{1,2})[/-](\d{1,2})\s+(\d{1,2}):(\d{1,2})", " ".join(normalized.split()))
if match is None:
raise ValueError(f"{label} 必须是合法时间") from exc
try:
parsed = datetime(
int(match.group(1)),
int(match.group(2)),
int(match.group(3)),
int(match.group(4)),
int(match.group(5)),
)
except ValueError as inner_exc:
raise ValueError(f"{label} 必须是合法时间") from inner_exc
return _as_utc(parsed)
def _default_simulation_time() -> datetime:
return datetime(2060, 3, 12, 9, 0, tzinfo=timezone.utc)
def _remember_simulation_time(value: datetime) -> None:
if not has_request_context():
return
session[SIMULATION_TIME_SESSION_KEY] = _format_datetime_input(value)
def _saved_simulation_time_value() -> str | None:
if not has_request_context():
return None
saved_value = session.get(SIMULATION_TIME_SESSION_KEY)
return saved_value if isinstance(saved_value, str) else None
def _resolve_simulation_time(raw_value: str | None) -> datetime:
try:
parsed_value = _parse_datetime(raw_value, "Simulation Time")
except ValueError:
parsed_value = None
if parsed_value is not None:
_remember_simulation_time(parsed_value)
return parsed_value
saved_value = _saved_simulation_time_value()
if saved_value is not None:
try:
saved_datetime = _parse_datetime(saved_value, "Simulation Time")
except ValueError:
saved_datetime = None
if saved_datetime is not None:
return saved_datetime
default_value = _default_simulation_time()
_remember_simulation_time(default_value)
return default_value
def _format_datetime_display(value: datetime | None) -> str:
normalized = _as_utc(value)
if normalized is None:
return "-"
return normalized.strftime("%Y-%m-%d %H:%M")
def _format_datetime_input(value: datetime | None) -> str:
normalized = _as_utc(value)
if normalized is None:
return ""
return normalized.strftime("%Y-%m-%dT%H:%M")
def _datetime_input_parts(value: datetime | None) -> dict[str, str]:
normalized = _as_utc(value)
if normalized is None:
return {"year": "", "month": "", "day": "", "hour": "", "minute": ""}
return {
"year": normalized.strftime("%Y"),
"month": normalized.strftime("%m"),
"day": normalized.strftime("%d"),
"hour": normalized.strftime("%H"),
"minute": normalized.strftime("%M"),
}
def _parse_datetime_form_value(
form: object,
field_name: str,
label: str,
*,
required: bool = False,
) -> datetime | None:
part_names = ("year", "month", "day", "hour", "minute")
part_values = {part: _normalized_text(form.get(f"{field_name}_{part}")) for part in part_names}
if any(value is not None for value in part_values.values()):
if any(value is None for value in part_values.values()):
raise ValueError(f"{label} 需要完整填写年月日时分")
try:
return datetime(
int(part_values["year"]),
int(part_values["month"]),
int(part_values["day"]),
int(part_values["hour"]),
int(part_values["minute"]),
tzinfo=timezone.utc,
)
except ValueError as exc:
raise ValueError(f"{label} 必须是合法时间") from exc
parsed_value = _parse_datetime(form.get(field_name), label)
if required and parsed_value is None:
raise ValueError(f"{label} 不能为空")
return parsed_value
class FormValidationError(ValueError):
def __init__(
self,
message: str,
*,
field_errors: dict[str, str] | None = None,
messages: list[str] | None = None,
) -> None:
super().__init__(message)
self.field_errors = field_errors or {}
self.messages = messages or [message]
def _build_validation_modal(
error: Exception | str,
*,
title: str = "本次提交未保存",
field_errors: dict[str, str] | None = None,
) -> dict[str, object]:
if isinstance(error, FormValidationError):
merged_field_errors = {**error.field_errors, **(field_errors or {})}
messages = list(error.messages)
else:
merged_field_errors = field_errors or {}
messages = [str(error)]
for message in merged_field_errors.values():
if message not in messages:
messages.append(message)
return {
"title": title,
"messages": messages,
"field_errors": merged_field_errors,
}
def _collect_submitted_form_values(
form: object,
field_specs: list[dict[str, object]],
prefix: str = "",
) -> dict[str, object]:
values: dict[str, object] = {}
for field_spec in field_specs:
field_name = field_spec["name"]
full_name = f"{prefix}{field_name}"
field_kind = field_spec["kind"]
if field_kind == "checkbox":
values[field_name] = full_name in form
continue
if field_kind == "combo":
selected_value = form.get(full_name)
custom_value = form.get(f"{full_name}__custom")
if _normalized_text(custom_value) is not None:
values[field_name] = custom_value or ""
elif selected_value == CUSTOM_OPTION_VALUE:
values[field_name] = ""
else:
values[field_name] = selected_value or ""
continue
values[field_name] = form.get(full_name) or ""
return values
def _apply_fields(
target: object,
form: object,
field_specs: list[dict[str, object]],
prefix: str = "",
) -> None:
parsed_values: dict[str, object] = {}
for field_spec in field_specs:
field_name = field_spec["name"]
full_name = f"{prefix}{field_name}"
field_kind = field_spec["kind"]
label = field_spec["label"]
required = bool(field_spec.get("required"))
try:
if field_kind == "checkbox":
value = full_name in form
elif field_kind == "combo":
selected_value = _normalized_text(form.get(full_name))
custom_value = _normalized_text(form.get(f"{full_name}__custom"))
if custom_value is not None:
value = custom_value
elif selected_value == CUSTOM_OPTION_VALUE:
value = None
else:
value = selected_value
elif field_kind == "decimal":
value = _parse_decimal(form.get(full_name), label)
elif field_kind == "integer":
value = _parse_integer(form.get(full_name), label)
else:
value = _normalized_text(form.get(full_name))
except ValueError as exc:
raise FormValidationError(str(exc), field_errors={field_name: str(exc)}) from exc
if required and value in (None, ""):
message = f"{label} 不能为空"
raise FormValidationError(message, field_errors={field_name: message})
parsed_values[field_name] = value
for field_name, value in parsed_values.items():
setattr(target, field_name, value)
def _collect_form_values(entity: object, field_specs: list[dict[str, object]]) -> dict[str, object]:
return {field["name"]: getattr(entity, field["name"], None) for field in field_specs}
def _load_asset_or_404(asset_id: object) -> Asset:
statement = select(Asset).options(selectinload(Asset.log_entries).selectinload(AssetLogEntry.state_nodes)).where(Asset.id == asset_id)
asset = db.session.execute(statement).scalar_one_or_none()
if asset is None:
abort(404)
return asset
def _load_asset_entry_or_404(asset_id: object, entry_id: object) -> AssetLogEntry:
entry = db.session.get(AssetLogEntry, entry_id)
if entry is None or entry.asset_id != asset_id:
abort(404)
return entry
def _apply_asset_entry_fields(target: AssetLogEntry, form: object) -> None:
entry_kind = _normalized_text(form.get("entry_kind")) or "event"
allowed_kinds = {option["value"] for option in ASSET_ENTRY_KIND_OPTIONS}
if entry_kind not in allowed_kinds:
message = "Entry Type 无效"
raise FormValidationError(message, field_errors={"entry_kind": message})
title = _normalized_text(form.get("title"))
if title is None:
message = "Title 不能为空"
raise FormValidationError(message, field_errors={"title": message})
try:
start_at = _parse_datetime_form_value(form, "start_at", "Start At", required=True)
except ValueError as exc:
raise FormValidationError(str(exc), field_errors={"start_at": str(exc)}) from exc
if start_at is None:
message = "Start At 不能为空"
raise FormValidationError(message, field_errors={"start_at": message})
try:
end_at = _parse_datetime_form_value(form, "end_at", "End At")
except ValueError as exc:
raise FormValidationError(str(exc), field_errors={"end_at": str(exc)}) from exc
if entry_kind == "event":
end_at = None
elif end_at is not None and end_at <= start_at:
message = "End At 必须晚于 Start At"
raise FormValidationError(message, field_errors={"start_at": message, "end_at": message})
state_node_rows = []
if entry_kind == "state":
state_node_rows = _parse_state_node_rows(form, start_at, end_at)
target.entry_kind = entry_kind
target.title = title
target.state_label = _normalize_multi_value_text(form.get("state_label"))
target.mission_label = _normalize_multi_value_text(form.get("mission_label"))
target.location = _normalized_text(form.get("location"))
target.start_at = start_at
target.end_at = end_at
target.summary = _normalized_text(form.get("summary"))
target.note = _normalized_text(form.get("note"))
if target.entry_kind == "state" and target.state_label is None:
target.state_label = title
_sync_state_nodes(target, state_node_rows if entry_kind == "state" else [])
def _asset_entry_form_values(entry: AssetLogEntry) -> dict[str, object]:
return {
"entry_kind": entry.entry_kind or "event",
"title": entry.title or "",
"state_label": entry.state_label or "",
"mission_label": entry.mission_label or "",
"location": entry.location or "",
"start_at": _format_datetime_input(entry.start_at),
"start_at_parts": _datetime_input_parts(entry.start_at),
"end_at": _format_datetime_input(entry.end_at),
"end_at_parts": _datetime_input_parts(entry.end_at),
"summary": entry.summary or "",
"note": entry.note or "",
"state_nodes": [
{
"id": str(node.id) if node.id is not None else "",
"title": node.title or "",
"detail": node.detail or "",
"at": _format_datetime_input(node.at_time),
}
for node in sorted(entry.state_nodes, key=lambda item: _as_utc(item.at_time) or _default_simulation_time())
],
}
def _asset_entry_form_values_from_form(form: object) -> dict[str, object]:
node_ids = form.getlist("state_node_id")
node_titles = form.getlist("state_node_title")
node_details = form.getlist("state_node_detail")
node_times = form.getlist("state_node_at")
row_count = max(len(node_ids), len(node_titles), len(node_details), len(node_times))
state_nodes = []
for index in range(row_count):
state_nodes.append(
{
"id": node_ids[index] if index < len(node_ids) else "",
"title": node_titles[index] if index < len(node_titles) else "",
"detail": node_details[index] if index < len(node_details) else "",
"at": node_times[index] if index < len(node_times) else "",
}
)
return {
"entry_kind": form.get("entry_kind") or "event",
"title": form.get("title") or "",
"state_label": form.get("state_label") or "",
"mission_label": form.get("mission_label") or "",
"location": form.get("location") or "",
"start_at": form.get("start_at") or "",
"start_at_parts": {"year": "", "month": "", "day": "", "hour": "", "minute": ""},
"end_at": form.get("end_at") or "",
"end_at_parts": {"year": "", "month": "", "day": "", "hour": "", "minute": ""},
"summary": form.get("summary") or "",
"note": form.get("note") or "",
"state_nodes": state_nodes,
}
def _parse_state_node_rows(
form: object,
start_at: datetime,
end_at: datetime | None,
) -> list[dict[str, object]]:
node_ids = form.getlist("state_node_id")
node_titles = form.getlist("state_node_title")
node_details = form.getlist("state_node_detail")
node_times = form.getlist("state_node_at")
if not node_ids and not node_titles and not node_details and not node_times:
return []
if (
len(node_ids) != len(node_titles)
or len(node_titles) != len(node_details)
or len(node_details) != len(node_times)
):
message = "State Node 数据不完整"
raise FormValidationError(message, field_errors={"state_nodes": message})
state_nodes = []
for index, (raw_id, raw_title, raw_detail, raw_at_time) in enumerate(
zip(node_ids, node_titles, node_details, node_times),
start=1,
):
node_id = _normalized_text(raw_id)
title = _normalized_text(raw_title)
detail = _normalized_text(raw_detail)
row_index = index - 1
try:
at_time = _parse_datetime(raw_at_time, f"State Node {index} At")
except ValueError as exc:
raise FormValidationError(
str(exc),
field_errors={f"state_node_at:{row_index}": str(exc)},
) from exc
if title is None and detail is None and at_time is None:
continue
if title is None or at_time is None:
message = f"State Node {index} 需要同时填写 Title 和 At"
field_errors: dict[str, str] = {}
if title is None:
field_errors[f"state_node_title:{row_index}"] = message
if at_time is None:
field_errors[f"state_node_at:{row_index}"] = message
raise FormValidationError(message, field_errors=field_errors)
if at_time < start_at:
message = f"State Node {index} 时间不能早于 Start At"
raise FormValidationError(
message,
field_errors={"start_at": message, f"state_node_at:{row_index}": message},
)
if end_at is not None and at_time > end_at:
message = f"State Node {index} 时间不能晚于 End At"
raise FormValidationError(
message,
field_errors={"end_at": message, f"state_node_at:{row_index}": message},
)
state_nodes.append(
{
"id": node_id,
"title": title,
"detail": detail,
"at_time": at_time,
}
)
return state_nodes
def _sync_state_nodes(target: AssetLogEntry, state_node_rows: list[dict[str, object]]) -> None:
existing_nodes = {str(node.id): node for node in target.state_nodes if node.id is not None}
next_nodes = []
for row in state_node_rows:
raw_node_id = row["id"]
if isinstance(raw_node_id, str) and raw_node_id:
node = existing_nodes.pop(raw_node_id, None)
if node is None:
message = "State Node 无效"
raise FormValidationError(message, field_errors={"state_nodes": message})
else:
node = AssetStateNode()
node.title = str(row["title"])
node.detail = row["detail"]
node.at_time = row["at_time"]
next_nodes.append(node)
next_nodes.sort(key=lambda item: _as_utc(item.at_time) or _default_simulation_time())
target.state_nodes = next_nodes
def _normalize_variant_ignitions(variant: EngineVariant) -> None:
if variant.has_unlimited_ignitions:
variant.ignitions = None
return
if variant.ignitions is None:
variant.has_unlimited_ignitions = True
def _calculate_variant_twr(max_thrust: Decimal | None, mass: Decimal | None) -> Decimal | None:
if max_thrust is None or mass in (None, Decimal("0")):
return None
return max_thrust / Decimal("9.80665") / mass
def _calculate_variant_throttle_ratio(
min_thrust: Decimal | None, max_thrust: Decimal | None
) -> Decimal | None:
if min_thrust is None or max_thrust in (None, Decimal("0")):
return None
return min_thrust / max_thrust
def _refresh_variant_derived_fields(variant: EngineVariant) -> None:
variant.twr = _calculate_variant_twr(variant.max_thrust_kn, variant.mass_t)
variant.throttle_ratio = _calculate_variant_throttle_ratio(
variant.min_thrust_kn,
variant.max_thrust_kn,
)
def _distinct_variant_values(family: EngineFamily, attribute_name: str) -> tuple[str, ...]:
values = {
getattr(variant, attribute_name)
for variant in family.variants
if getattr(variant, attribute_name) not in (None, "")
}
return tuple(sorted(values, key=str.casefold))
def _distinct_text_values(values: list[str | None]) -> tuple[str, ...]:
cleaned_values = {value.strip() for value in values if value and value.strip()}
return tuple(sorted(cleaned_values, key=str.casefold))
def _engine_form_options() -> dict[str, tuple[str, ...]]:
cycle_values = db.session.execute(select(EngineFamily.cycle)).scalars().all()
fuel_values = db.session.execute(select(EngineVariant.fuel_type)).scalars().all()
mod_values = db.session.execute(select(EngineVariant.mod_source)).scalars().all()
return {
"cycle": _distinct_text_values(cycle_values),
"fuel_type": _distinct_text_values(fuel_values),
"mod_source": _distinct_text_values(mod_values),
}
def _asset_form_options() -> dict[str, tuple[str, ...]]:
asset_type_values = db.session.execute(select(Asset.asset_type)).scalars().all()
return {
"asset_type": _distinct_text_values([*ASSET_TYPE_BASE_OPTIONS, *asset_type_values]),
}
def _max_variant_decimal(family: EngineFamily, attribute_name: str) -> Decimal | None:
values = [
getattr(variant, attribute_name)
for variant in family.variants
if getattr(variant, attribute_name) is not None
]
return max(values) if values else None
def _estimate_sl_thrust_kn(variant: EngineVariant) -> Decimal | None:
if variant.max_thrust_kn is None:
return None
if variant.sl_isp is None or variant.vac_isp in (None, Decimal("0")):
return None
return variant.max_thrust_kn * variant.sl_isp / variant.vac_isp
def _build_catalog_row(family: EngineFamily) -> EngineCatalogRow:
sl_thrust_values = [
value for value in (_estimate_sl_thrust_kn(variant) for variant in family.variants) if value is not None
]
return EngineCatalogRow(
family=family,
config_count=len(family.variants),
fuel_types=_distinct_variant_values(family, "fuel_type"),
mod_sources=_distinct_variant_values(family, "mod_source"),
mass_t=_max_variant_decimal(family, "mass_t"),
twr=_max_variant_decimal(family, "twr"),
vac_thrust_kn=_max_variant_decimal(family, "max_thrust_kn"),
sl_thrust_kn=max(sl_thrust_values) if sl_thrust_values else None,
vac_isp=_max_variant_decimal(family, "vac_isp"),
sl_isp=_max_variant_decimal(family, "sl_isp"),
)
def _build_catalog_rows() -> list[EngineCatalogRow]:
families = db.session.execute(
_engine_family_statement().order_by(EngineFamily.engine_name.asc())
).scalars().unique().all()
return [_build_catalog_row(family) for family in families]
def _matches_catalog_search(row: EngineCatalogRow, search_term: str) -> bool:
normalized = search_term.casefold()
search_targets = [
row.family.engine_name,
row.family.part_name or "",
row.family.cycle or "",
*row.fuel_types,
*row.mod_sources,
*(variant.config_name or "" for variant in row.family.variants),
]
return any(normalized in target.casefold() for target in search_targets if target)
def _decimal_sort_key(value: Decimal | None, descending: bool = False) -> tuple[int, Decimal]:
if value is None:
return (1, Decimal("0"))
return (0, -value if descending else value)
def _sort_catalog_rows(rows: list[EngineCatalogRow], sort_option: str) -> str:
allowed = {option["value"] for option in ENGINE_SORT_OPTIONS}
normalized = sort_option if sort_option in allowed else "engine_name_asc"
if normalized == "engine_name_desc":
rows.sort(key=lambda row: row.family.engine_name.casefold(), reverse=True)
return normalized
if normalized == "vac_thrust_asc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.vac_thrust_kn),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "vac_thrust_desc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.vac_thrust_kn, descending=True),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "sl_thrust_asc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.sl_thrust_kn),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "sl_thrust_desc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.sl_thrust_kn, descending=True),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "mass_t_asc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.mass_t),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "mass_t_desc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.mass_t, descending=True),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "vac_isp_asc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.vac_isp),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "vac_isp_desc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.vac_isp, descending=True),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "sl_isp_asc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.sl_isp),
row.family.engine_name.casefold(),
)
)
return normalized
if normalized == "sl_isp_desc":
rows.sort(
key=lambda row: (
_decimal_sort_key(row.sl_isp, descending=True),
row.family.engine_name.casefold(),
)
)
return normalized
rows.sort(key=lambda row: row.family.engine_name.casefold())
return normalized
def _build_engine_sort_links(query_args: dict[str, object], current_sort: str) -> list[dict[str, object]]:
columns = []
for column in ENGINE_SORT_COLUMNS:
asc_value = f"{column['key']}_asc"
desc_value = f"{column['key']}_desc"
columns.append(
{
"key": column["key"],
"label": column["label"],
"asc_url": _page_url("web.engine_list", {**query_args, "sort": asc_value}, 1),
"desc_url": _page_url("web.engine_list", {**query_args, "sort": desc_value}, 1),
"asc_active": current_sort == asc_value,
"desc_active": current_sort == desc_value,
}
)
return columns
def _sort_communications(rows: list[CommunicationPart], sort_option: str) -> str:
allowed = {option["value"] for option in COMM_SORT_OPTIONS}
normalized = sort_option if sort_option in allowed else "part_name"
if normalized == "range_desc":
rows.sort(
key=lambda item: (
_decimal_sort_key(item.range_km, descending=True),
item.part_name.casefold(),
)
)
return normalized
if normalized == "mass_desc":
rows.sort(
key=lambda item: (
_decimal_sort_key(item.mass_t, descending=True),
item.part_name.casefold(),
)
)
return normalized
rows.sort(key=lambda item: item.part_name.casefold())
return normalized
def _sort_tanks(rows: list[TankSpec], sort_option: str) -> str:
allowed = {option["value"] for option in TANK_SORT_OPTIONS}
normalized = sort_option if sort_option in allowed else "tank_name"
if normalized == "volume_desc":
rows.sort(
key=lambda item: (
_decimal_sort_key(item.tank_volume_l, descending=True),
item.tank_name.casefold(),
)
)
return normalized
if normalized == "ratio_desc":
rows.sort(
key=lambda item: (
_decimal_sort_key(item.mass_ratio, descending=True),
item.tank_name.casefold(),
)
)
return normalized
rows.sort(key=lambda item: item.tank_name.casefold())
return normalized
def _sort_vehicles(rows: list[VehicleCost], sort_option: str) -> str:
allowed = {option["value"] for option in VEHICLE_SORT_OPTIONS}
normalized = sort_option if sort_option in allowed else "vehicle_name"
if normalized == "launch_price_desc":
rows.sort(
key=lambda item: (
item.launch_price is None,
-(item.launch_price or 0),
item.vehicle_name.casefold(),
)
)
return normalized
if normalized == "launch_price_asc":
rows.sort(
key=lambda item: (
item.launch_price is None,
item.launch_price or 0,
item.vehicle_name.casefold(),
)
)
return normalized
rows.sort(key=lambda item: item.vehicle_name.casefold())
return normalized
def _sort_assets(rows: list[Asset], sort_option: str) -> str:
allowed = {option["value"] for option in ASSET_SORT_OPTIONS}
normalized = sort_option if sort_option in allowed else "name_asc"
if normalized == "updated_desc":
rows.sort(
key=lambda item: (
_as_utc(item.updated_at) or datetime.min.replace(tzinfo=timezone.utc),
item.name.casefold(),
),
reverse=True,
)
elif normalized == "asset_type_asc":
rows.sort(key=lambda item: ((item.asset_type or "").casefold(), item.name.casefold()))
else:
rows.sort(key=lambda item: item.name.casefold())
rows.sort(key=lambda item: bool(item.is_retired))
return normalized
def _classify_asset_entry(entry: AssetLogEntry, simulation_time: datetime) -> dict[str, str]:
start_at = _as_utc(entry.start_at)
end_at = _as_utc(entry.end_at)
if entry.entry_kind == "state":
if start_at and start_at <= simulation_time and (end_at is None or simulation_time < end_at):
return {"label": "Current Interval", "tone": "active"}
if start_at and simulation_time < start_at:
return {"label": "Upcoming Interval", "tone": "planned"}
return {"label": "Past Interval", "tone": "completed"}
if start_at and simulation_time < start_at:
return {"label": "Upcoming Event", "tone": "planned"}
return {"label": "Past Event", "tone": "completed"}
def _entry_matches_date_range(
entry: AssetLogEntry,
range_start: datetime | None,
range_end: datetime | None,
) -> bool:
start_at = _as_utc(entry.start_at)
if start_at is None:
return False
window_start = range_start or datetime.min.replace(tzinfo=timezone.utc)
window_end = range_end or datetime.max.replace(tzinfo=timezone.utc)
if entry.entry_kind == "event":
return window_start <= start_at <= window_end
effective_end = _as_utc(entry.end_at) or datetime.max.replace(tzinfo=timezone.utc)
return start_at <= window_end and effective_end >= window_start
def _build_asset_future_events(
asset: Asset,
simulation_time: datetime,
active_state: AssetLogEntry | None,
) -> list[dict[str, object]]:
future_events: list[dict[str, object]] = []
if active_state is not None:
for state_node in sorted(
active_state.state_nodes,
key=lambda item: _as_utc(item.at_time) or simulation_time,
):
node_time = _as_utc(state_node.at_time)
if node_time is None or node_time <= simulation_time:
continue
future_events.append(
{
"sort_at": node_time,
"title": state_node.title,
"when": _format_datetime_display(state_node.at_time),
"scope": active_state.title or asset.name,
"summary": state_node.detail or "在所属 State Interval 中编辑该节点。",
"kind_label": "State Node",
"status_label": "Upcoming Node",
"state_labels": _split_multi_value_text(active_state.state_label),
"mission_labels": _split_multi_value_text(active_state.mission_label),
"source_entry_id": active_state.id,
"action_label": "编辑所属区间",
"edit_hint": "State Node 需要在所属 State Interval 中修改。",
}
)
for entry in sorted(
asset.log_entries,
key=lambda item: _as_utc(item.start_at) or _default_simulation_time(),
):
start_at = _as_utc(entry.start_at)
if start_at is None or start_at <= simulation_time:
continue
status = _classify_asset_entry(entry, simulation_time)
future_events.append(
{
"sort_at": start_at,
"title": entry.title,
"when": _format_datetime_display(entry.start_at),
"scope": entry.location or _display_multi_value_text(entry.mission_label) or asset.name,
"summary": entry.summary or "未填写摘要。",
"kind_label": "Event Point" if entry.entry_kind == "event" else "State Interval",
"status_label": status["label"],
"state_labels": _split_multi_value_text(entry.state_label),
"mission_labels": _split_multi_value_text(entry.mission_label),
"source_entry_id": entry.id,
"action_label": "编辑条目",
"edit_hint": None,
}
)
future_events.sort(key=lambda item: (item["sort_at"], str(item["title"])))
return future_events
def _build_asset_snapshot(asset: Asset, simulation_time: datetime) -> dict[str, object]:
entries = sorted(asset.log_entries, key=lambda item: _as_utc(item.start_at) or _default_simulation_time())
active_state = None
for entry in entries:
if entry.entry_kind != "state":
continue
start_at = _as_utc(entry.start_at)
end_at = _as_utc(entry.end_at)
if start_at and start_at <= simulation_time and (end_at is None or simulation_time < end_at):
active_state = entry
past_entries = [entry for entry in entries if (_as_utc(entry.start_at) or simulation_time) <= simulation_time]
last_entry = past_entries[-1] if past_entries else None
future_entries = [entry for entry in entries if (_as_utc(entry.start_at) or simulation_time) > simulation_time]
next_entry = future_entries[0] if future_entries else None
future_events = _build_asset_future_events(asset, simulation_time, active_state)
if active_state is not None:
current_state = _display_multi_value_text(active_state.state_label) or active_state.title
current_location = active_state.location or asset.home_region or "-"
current_mission = _display_multi_value_text(active_state.mission_label) or asset.program or "-"
state_window = f"{_format_datetime_display(active_state.start_at)}{_format_datetime_display(active_state.end_at) if active_state.end_at else 'Open'}"
else:
current_state = "No Active Interval" if last_entry is not None else "No Logged Timeline"
current_location = (
(last_entry.location if last_entry is not None else None)
or asset.home_region
or "-"
)
current_mission = (
(_display_multi_value_text(last_entry.mission_label) if last_entry is not None else None)
or asset.program
or "-"
)
if last_entry is None:
state_window = "No entry covers selected Simulation Time"
elif last_entry.entry_kind == "state":
state_window = (
f"Last interval · {_format_datetime_display(last_entry.start_at)}"
f" → {_format_datetime_display(last_entry.end_at) if last_entry.end_at else 'Open'}"
)
else:
state_window = f"Last event · {_format_datetime_display(last_entry.start_at)}"
current_event = active_state.title if active_state is not None else (last_entry.title if last_entry is not None else "No Logged Timeline")
next_event_label = "No future event recorded"
if future_events:
next_event_label = f"{future_events[0]['title']} · {future_events[0]['when']}"
return {
"current_state": current_state,
"current_event": current_event,
"current_location": current_location,
"current_mission": current_mission,
"state_window": state_window,
"next_event": next_event_label,
"next_entry": next_entry,
"active_state": active_state,
"last_entry": last_entry,
"future_events": future_events,
"upcoming_events": future_events[:4],
}
def _build_asset_catalog_row(asset: Asset, simulation_time: datetime) -> dict[str, object]:
snapshot = _build_asset_snapshot(asset, simulation_time)
return {
"asset": asset,
"is_retired": asset.is_retired,
"current_event": snapshot["current_event"],
"current_location": snapshot["current_location"],
"next_event": snapshot["next_event"],
"entry_count": len(asset.log_entries),
}
def _load_assets_with_logs() -> list[Asset]:
return db.session.execute(
select(Asset).options(selectinload(Asset.log_entries).selectinload(AssetLogEntry.state_nodes)).order_by(Asset.name.asc())
).scalars().all()
def _classify_asset_board_record(snapshot: dict[str, object]) -> dict[str, object]:
if snapshot["active_state"] is not None:
return {"label": "Active", "rank": 0}
if snapshot["next_entry"] is not None:
return {"label": "Planned", "rank": 1}
if snapshot["last_entry"] is not None:
return {"label": "Recorded", "rank": 2}
return {"label": "Unlogged", "rank": 3}
def _derive_status_group_title(snapshot: dict[str, object], record_status: dict[str, object]) -> str:
if snapshot["active_state"] is not None:
return str(snapshot["current_state"])
if record_status["label"] == "Planned":
return "Upcoming / Not Yet Active"
if record_status["label"] == "Recorded":
return "No Active Interval"
return "No Logged Timeline"
def _status_group_accent(group_title: str, record_label: str) -> str:
normalized = group_title.casefold()
if record_label == "Planned":
return "warning"
if "transit" in normalized or "transfer" in normalized or "cruise" in normalized:
return "warning"
if "station" in normalized or "dock" in normalized or "orbit" in normalized:
return "primary"
if "surface" in normalized or "port" in normalized or "outpost" in normalized or "base" in normalized:
return "neutral"
if "no active" in normalized or "no logged" in normalized:
return "soft"
if record_label == "Active":
return "primary"
if record_label == "Recorded":
return "soft"
return "neutral"
def _asset_board_summary(asset: Asset, snapshot: dict[str, object]) -> str:
summary_source = snapshot["active_state"] or snapshot["last_entry"] or snapshot["next_entry"]
if summary_source is not None and summary_source.summary:
return summary_source.summary
if asset.note:
return asset.note
if snapshot["last_entry"] is not None:
return f"Last logged entry: {snapshot['last_entry'].title}."
if snapshot["next_entry"] is not None:
return f"Next recorded entry: {snapshot['next_entry'].title}."
return "No summary recorded."
def _normalize_location_group(location: str) -> str:
if not location or location == "-":
return "Unknown Location"
return location
def _build_mission_board_rows(assets: list[Asset], simulation_time: datetime) -> list[dict[str, object]]:
rows = []
for asset in assets:
snapshot = _build_asset_snapshot(asset, simulation_time)
record_status = _classify_asset_board_record(snapshot)
state_group = _derive_status_group_title(snapshot, record_status)
current_location = _normalize_location_group(str(snapshot["current_location"]))
state_display = state_group if snapshot["active_state"] is None else str(snapshot["current_state"])
rows.append(
{
"asset": asset,
"snapshot": snapshot,
"record_status": record_status,
"name": asset.name,
"asset_id": asset.id,
"asset_type": asset.asset_type,
"location": current_location,
"mission": snapshot["current_mission"],
"state_window": snapshot["state_window"],
"next_event": snapshot["next_event"],
"summary": _asset_board_summary(asset, snapshot),
"status_group": state_group,
"status_accent": _status_group_accent(state_group, str(record_status["label"])),
"state_display": state_display,
}
)
return rows
def _normalize_record_scope(raw_value: str | None) -> str:
allowed = {option["value"] for option in MISSION_BOARD_RECORD_SCOPE_OPTIONS}
normalized = _normalized_text(raw_value) or "all"
return normalized if normalized in allowed else "all"
def _matches_record_scope(row: dict[str, object], record_scope: str) -> bool:
record_label = row["record_status"]["label"]
if record_scope == "active":
return record_label == "Active"
if record_scope == "active_planned":
return record_label in {"Active", "Planned"}
if record_scope == "recorded":
return record_label == "Recorded"
return True
def _filter_mission_board_rows(
rows: list[dict[str, object]],
asset_type_filter: str,
location_filter: str,
record_scope: str,
) -> list[dict[str, object]]:
filtered_rows = []
for row in rows:
if asset_type_filter and asset_type_filter != (row["asset_type"] or ""):
continue
if location_filter and location_filter != row["location"]:
continue
if not _matches_record_scope(row, record_scope):
continue
filtered_rows.append(row)
return filtered_rows
def _build_status_board_groups(rows: list[dict[str, object]]) -> list[dict[str, object]]:
groups_by_title: dict[str, dict[str, object]] = {}
for row in rows:
group = groups_by_title.setdefault(
row["status_group"],
{
"title": row["status_group"],
"accent": row["status_accent"],
"rank": row["record_status"]["rank"],
"items": [],
},
)
group["rank"] = min(group["rank"], row["record_status"]["rank"])
group["items"].append(
{
"asset_id": row["asset_id"],
"name": row["name"],
"asset_type": row["asset_type"],
"location": row["location"],
"mission": row["mission"],
"state_window": row["state_window"],
"next_event": row["next_event"],
"summary": row["summary"],
"record_status": row["record_status"]["label"],
"record_sort_rank": row["record_status"]["rank"],
}
)
groups = list(groups_by_title.values())
for group in groups:
group["items"].sort(key=lambda item: (item["record_sort_rank"], item["name"].casefold()))
groups.sort(key=lambda item: (item["rank"], item["title"].casefold()))
return groups
def _build_location_board_groups(rows: list[dict[str, object]]) -> list[dict[str, object]]:
groups_by_location: dict[str, dict[str, object]] = {}
for row in rows:
group = groups_by_location.setdefault(
row["location"],
{
"title": row["location"],
"subtitle": "Derived from active interval or latest logged event.",
"items": [],
},
)
group["items"].append(
{
"asset_id": row["asset_id"],
"name": row["name"],
"asset_type": row["asset_type"],
"state": row["state_display"],
"mission": row["mission"],
"location": row["location"],
"next_event": row["next_event"],
"record_sort_rank": row["record_status"]["rank"],
}
)
groups = list(groups_by_location.values())
for group in groups:
item_count = len(group["items"])
if group["title"] == "Unknown Location":
group["subtitle"] = "No active interval or logged location is available yet."
elif item_count == 1:
group["subtitle"] = "1 asset currently attributed to this location."
else:
group["subtitle"] = f"{item_count} assets currently attributed to this location."
group["items"].sort(key=lambda item: (item["record_sort_rank"], item["name"].casefold()))
groups.sort(key=lambda item: (item["title"] == "Unknown Location", item["title"].casefold()))
return groups
def _build_board_upcoming_events(rows: list[dict[str, object]], simulation_time: datetime) -> list[dict[str, str]]:
future_entries = []
for row in rows:
asset = row["asset"]
for entry in asset.log_entries:
start_at = _as_utc(entry.start_at)
if start_at is None or start_at <= simulation_time:
continue
future_entries.append(
{
"when": _format_datetime_display(entry.start_at),
"title": f"{asset.name} · {entry.title}",
"scope": entry.location or row["location"],
"sort_at": start_at,
}
)
future_entries.sort(key=lambda item: item["sort_at"])
return [
{"when": item["when"], "title": item["title"], "scope": item["scope"]}
for item in future_entries[:6]
]
def _build_board_metrics(all_rows: list[dict[str, object]], visible_rows: list[dict[str, object]], simulation_time: datetime) -> list[dict[str, str]]:
upcoming_window_limit = simulation_time + timedelta(days=30)
upcoming_30d_count = 0
for row in visible_rows:
for entry in row["asset"].log_entries:
start_at = _as_utc(entry.start_at)
if start_at is not None and simulation_time < start_at <= upcoming_window_limit:
upcoming_30d_count += 1
return [
{"label": "Total Assets", "value": str(len(all_rows))},
{"label": "Visible Cards", "value": str(len(visible_rows))},
{"label": "30-Day Events", "value": str(upcoming_30d_count)},
]
def _build_asset_catalog_metrics(all_assets: list[Asset], visible_assets: list[Asset], simulation_time: datetime) -> list[dict[str, str]]:
upcoming_window_limit = simulation_time + timedelta(days=30)
upcoming_30d_count = 0
for asset in visible_assets:
for entry in asset.log_entries:
start_at = _as_utc(entry.start_at)
if start_at is not None and simulation_time < start_at <= upcoming_window_limit:
upcoming_30d_count += 1
return [
{"label": "Total Assets", "value": str(len(all_assets))},
{"label": "Current Results", "value": str(len(visible_assets))},
{"label": "30-Day Events", "value": str(upcoming_30d_count)},
]
def _build_header_hidden_fields(values: dict[str, object]) -> list[dict[str, str]]:
hidden_fields: list[dict[str, str]] = []
for name, value in values.items():
if value in (None, "", [], (), set()):
continue
if isinstance(value, (list, tuple, set)):
for item in value:
if item in (None, ""):
continue
hidden_fields.append({"name": name, "value": str(item)})
continue
hidden_fields.append({"name": name, "value": str(value)})
return hidden_fields
def _mission_board_query_args(
simulation_time: datetime,
asset_type_filter: str,
location_filter: str,
record_scope: str,
) -> dict[str, str]:
params = {
"sim_time": _format_datetime_input(simulation_time),
"record_scope": record_scope,
}
if asset_type_filter:
params["asset_type"] = asset_type_filter
if location_filter:
params["location"] = location_filter
return params
def _mission_board_context(
endpoint: str,
simulation_time: datetime,
asset_type_filter: str,
location_filter: str,
record_scope: str,
all_rows: list[dict[str, object]],
visible_rows: list[dict[str, object]],
) -> dict[str, object]:
query_args = _mission_board_query_args(simulation_time, asset_type_filter, location_filter, record_scope)
return {
"simulation_time": _format_datetime_display(simulation_time),
"simulation_time_display": _format_datetime_display(simulation_time),
"simulation_time_input": _format_datetime_input(simulation_time),
"metrics": _build_board_metrics(all_rows, visible_rows, simulation_time),
"preview_eyebrow": "Mission Ops",
"header_time_action": url_for(endpoint),
"header_time_field_id": f"{endpoint.rsplit('.', 1)[-1]}-header-sim-time",
"header_time_submit_label": "Apply Time",
"header_hidden_fields": _build_header_hidden_fields(
{
"asset_type": asset_type_filter,
"location": location_filter,
"record_scope": record_scope,
}
),
"upcoming_events": _build_board_upcoming_events(visible_rows, simulation_time),
"asset_type_filter": asset_type_filter,
"location_filter": location_filter,
"record_scope": record_scope,
"asset_type_options": sorted({row["asset_type"] for row in all_rows if row["asset_type"]}, key=str.casefold),
"location_options": sorted({row["location"] for row in all_rows if row["location"]}, key=str.casefold),
"record_scope_options": MISSION_BOARD_RECORD_SCOPE_OPTIONS,
"time_shift_urls": {
"minus_7": url_for(endpoint, **{**query_args, "sim_time": _format_datetime_input(simulation_time - timedelta(days=7))}),
"plus_7": url_for(endpoint, **{**query_args, "sim_time": _format_datetime_input(simulation_time + timedelta(days=7))}),
},
"reset_url": url_for(endpoint, sim_time=_format_datetime_input(simulation_time)),
"board_nav_urls": {
"status": url_for("web.mission_status_board_preview", **query_args),
"location": url_for("web.mission_location_board_preview", **query_args),
"mission": url_for("web.mission_list", sim_time=_format_datetime_input(simulation_time)),
"asset": url_for("web.asset_list", sim_time=_format_datetime_input(simulation_time)),
"timeline": url_for("web.mission_timeline_preview", sim_time=_format_datetime_input(simulation_time)),
},
}
def _mission_ops_nav_urls(simulation_time: datetime) -> dict[str, str]:
simulation_time_input = _format_datetime_input(simulation_time)
return {
"status": url_for("web.mission_status_board_preview", sim_time=simulation_time_input),
"location": url_for("web.mission_location_board_preview", sim_time=simulation_time_input),
"mission": url_for("web.mission_list", sim_time=simulation_time_input),
"asset": url_for("web.asset_list", sim_time=simulation_time_input),
"timeline": url_for("web.mission_timeline_preview", sim_time=simulation_time_input),
}
def _month_start(value: datetime) -> datetime:
normalized = _as_utc(value) or _default_simulation_time()
return normalized.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
def _month_end(value: datetime) -> datetime:
start = _month_start(value)
if start.month == 12:
return start.replace(year=start.year + 1, month=1)
return start.replace(month=start.month + 1)
def _month_delta(start: datetime, end: datetime) -> int:
return (end.year - start.year) * 12 + (end.month - start.month)
def _day_start(value: datetime) -> datetime:
normalized = _as_utc(value) or _default_simulation_time()
return normalized.replace(hour=0, minute=0, second=0, microsecond=0)
def _days_in_month(year: int, month: int) -> int:
if month == 12:
next_month = datetime(year + 1, 1, 1)
else:
next_month = datetime(year, month + 1, 1)
return (next_month - datetime(year, month, 1)).days
def _timeline_add_units(value: datetime, scale: str, count: int = 1) -> datetime:
normalized = _as_utc(value) or _default_simulation_time()
if scale == "day":
return normalized + timedelta(days=count)
if scale == "year":
target_year = normalized.year + count
target_day = min(normalized.day, _days_in_month(target_year, normalized.month))
return normalized.replace(year=target_year, day=target_day)
total_month = (normalized.year * 12 + (normalized.month - 1)) + count
target_year = total_month // 12
target_month = (total_month % 12) + 1
target_day = min(normalized.day, _days_in_month(target_year, target_month))
return normalized.replace(year=target_year, month=target_month, day=target_day)
def _timeline_bucket_start(value: datetime, scale: str) -> datetime:
normalized = _as_utc(value) or _default_simulation_time()
if scale == "year":
return normalized.replace(month=1, day=1, hour=0, minute=0, second=0, microsecond=0)
if scale == "day":
return _day_start(normalized)
return _month_start(normalized)
def _timeline_next_bucket(value: datetime, scale: str) -> datetime:
bucket_start = _timeline_bucket_start(value, scale)
if scale == "year":
return bucket_start.replace(year=bucket_start.year + 1)
if scale == "day":
return bucket_start + timedelta(days=1)
return _month_end(bucket_start)
def _timeline_bucket_delta(start: datetime, end: datetime, scale: str) -> int:
start_bucket = _timeline_bucket_start(start, scale)
end_bucket = _timeline_bucket_start(end, scale)
if scale == "year":
return end_bucket.year - start_bucket.year
if scale == "day":
return (end_bucket - start_bucket).days
return _month_delta(start_bucket, end_bucket)
def _resolve_timeline_scale(raw_value: str | None, range_start: datetime, range_end: datetime) -> str:
allowed = {option["value"] for option in TIMELINE_SCALE_OPTIONS}
normalized = _normalized_text(raw_value)
if normalized in allowed:
return str(normalized)
total_days = max(1, int((_timeline_bucket_start(range_end, "day") - _timeline_bucket_start(range_start, "day")).days) + 1)
if total_days <= 120:
return "day"
if total_days <= 365 * 4:
return "month"
return "year"
def _resolve_timeline_boundary(raw_value: str | None, fallback: datetime) -> datetime:
try:
return _parse_datetime(raw_value, "Timeline Boundary") or fallback
except ValueError:
return fallback
def _timeline_default_range(rows: list[dict[str, object]], simulation_time: datetime) -> tuple[datetime, datetime]:
return _timeline_default_range_for_scale(rows, simulation_time, "month")
def _timeline_default_range_for_scale(rows: list[dict[str, object]], simulation_time: datetime, scale: str) -> tuple[datetime, datetime]:
start_timestamps = []
effective_end_timestamps = []
for row in rows:
for entry in row["asset"].log_entries:
start_at = _as_utc(entry.start_at)
end_at = _as_utc(entry.end_at)
if start_at is not None:
start_timestamps.append(start_at)
effective_end_timestamps.append(end_at or start_at)
all_timestamps = [*start_timestamps, *effective_end_timestamps]
if not all_timestamps:
fallback_start = _day_start(simulation_time)
return fallback_start, _timeline_add_units(fallback_start, scale)
earliest_start = min(start_timestamps or all_timestamps)
latest_finish = max(effective_end_timestamps or all_timestamps)
return earliest_start, _timeline_add_units(latest_finish, scale)
def _timeline_scale_labels(start_at: datetime, end_at: datetime, scale: str) -> list[str]:
labels = []
cursor = _timeline_bucket_start(start_at, scale)
boundary = _timeline_bucket_start(end_at, scale)
while cursor <= boundary:
if scale == "year":
labels.append(cursor.strftime("%Y"))
elif scale == "day":
labels.append(cursor.strftime("%Y-%m-%d"))
else:
labels.append(cursor.strftime("%Y-%m"))
cursor = _timeline_next_bucket(cursor, scale)
return labels
def _timeline_range_display(start_at: datetime, end_at: datetime, scale: str) -> str:
start_bucket = _timeline_bucket_start(start_at, scale)
end_bucket = _timeline_bucket_start(end_at, scale)
if scale == "year":
return f"{start_bucket.strftime('%Y')} to {end_bucket.strftime('%Y')}"
if scale == "day":
return f"{start_bucket.strftime('%Y-%m-%d')} to {end_bucket.strftime('%Y-%m-%d')}"
return f"{start_bucket.strftime('%Y-%m')} to {end_bucket.strftime('%Y-%m')}"
def _timeline_column_width(scale: str) -> int:
if scale == "day":
return 48
if scale == "year":
return 84
return 72
def _timeline_segment_tone(entry: AssetLogEntry, simulation_time: datetime) -> str:
title = (entry.title or "").casefold()
location = (entry.location or "").casefold()
tone = "neutral"
if any(token in title for token in ("transfer", "transit", "outbound", "return", "departure")):
tone = "warning"
elif any(token in title for token in ("operation", "ops", "orbit", "station", "trial", "expedition")):
tone = "primary"
elif any(token in title for token in ("maintenance", "upgrade", "construction", "fitting", "reconfig")):
tone = "soft"
elif any(token in location for token in ("mars", "europa", "jupiter", "saturn", "neptune", "surface", "orbit")):
tone = "primary"
start_at = _as_utc(entry.start_at)
end_at = _as_utc(entry.end_at)
if start_at is not None and start_at <= simulation_time and (end_at is None or simulation_time < end_at):
return f"{tone} current"
return tone
def _entry_overlaps_timeline(entry: AssetLogEntry, range_start: datetime, range_end: datetime) -> bool:
start_at = _as_utc(entry.start_at)
end_at = _as_utc(entry.end_at)
if start_at is None:
return False
if entry.entry_kind == "event":
return range_start <= start_at <= range_end
effective_end = end_at or range_end
return start_at <= range_end and effective_end > range_start
def _timeline_entry_edit_url(asset_id: object, entry_id: object, simulation_time: datetime) -> str:
return url_for(
"web.asset_entry_edit",
asset_id=asset_id,
entry_id=entry_id,
sim_time=_format_datetime_input(simulation_time),
)
def _timeline_state_windows(asset: Asset) -> dict[object, tuple[datetime, datetime | None]]:
state_entries = [
entry
for entry in sorted(asset.log_entries, key=lambda item: _as_utc(item.start_at) or _default_simulation_time())
if entry.entry_kind == "state" and _as_utc(entry.start_at) is not None
]
windows: dict[object, tuple[datetime, datetime | None]] = {}
for entry in state_entries:
start_at = _as_utc(entry.start_at)
if start_at is None:
continue
windows[entry.id] = (start_at, _as_utc(entry.end_at))
return windows
def _timeline_state_is_active(
entry: AssetLogEntry,
at_time: datetime,
state_windows: dict[object, tuple[datetime, datetime | None]],
) -> bool:
start_at, end_at = state_windows.get(entry.id, (_as_utc(entry.start_at) or at_time, _as_utc(entry.end_at)))
return start_at <= at_time and (end_at is None or at_time < end_at)
def _timeline_state_overlaps(
entry: AssetLogEntry,
range_start: datetime,
range_end: datetime,
state_windows: dict[object, tuple[datetime, datetime | None]],
) -> bool:
start_at, end_at = state_windows.get(entry.id, (_as_utc(entry.start_at) or range_start, _as_utc(entry.end_at)))
effective_end = end_at or range_end
return start_at <= range_end and effective_end > range_start
def _timeline_position(value: datetime, bucket_origin: datetime, scale: str, bucket_count: int) -> float:
normalized = _as_utc(value) or bucket_origin
if normalized <= bucket_origin:
return 0.0
bucket_start = _timeline_bucket_start(normalized, scale)
bucket_end = _timeline_next_bucket(bucket_start, scale)
bucket_index = _timeline_bucket_delta(bucket_origin, bucket_start, scale)
bucket_duration = max((bucket_end - bucket_start).total_seconds(), 1.0)
fraction = max(0.0, min(1.0, (normalized - bucket_start).total_seconds() / bucket_duration))
return max(0.0, min(float(bucket_count), bucket_index + fraction))
def _timeline_percent(position: float, bucket_count: int) -> str:
if bucket_count <= 0:
return "0%"
return f"{(position / bucket_count) * 100:.6f}%"
def _assign_timeline_tracks(
items: list[dict[str, object]],
*,
start_key: str,
end_key: str,
minimum_span: float,
gap: float = 0.0,
span_key: str | None = None,
) -> int:
if not items:
return 1
track_ends: list[float] = []
for item in items:
start_position = float(item[start_key])
item_span = minimum_span
if span_key is not None:
span_value = item.get(span_key)
if span_value is not None:
item_span = max(item_span, float(span_value))
end_position = max(float(item[end_key]), start_position + item_span)
track_index = 0
while track_index < len(track_ends) and start_position < track_ends[track_index] + gap:
track_index += 1
if track_index == len(track_ends):
track_ends.append(end_position)
else:
track_ends[track_index] = end_position
item["track"] = track_index
return max(1, len(track_ends))
def _timeline_state_track_settings(scale: str) -> tuple[float, float]:
if scale == "year":
return 0.9, 0.04
if scale == "day":
return 0.18, 0.02
return 0.24, 0.03
def _timeline_event_track_settings(scale: str) -> tuple[float, float]:
label_span = 110 / max(_timeline_column_width(scale), 1)
if scale == "year":
return max(0.95, label_span), 0.08
if scale == "day":
return max(1.75, label_span), 0.04
return max(1.35, label_span), 0.06
def _timeline_event_label_width_px(label: str, scale: str) -> int:
weighted_length = 0.0
for character in label.strip():
if character.isspace():
weighted_length += 0.3
elif ord(character) < 128:
weighted_length += 0.85
else:
weighted_length += 1.55
if scale == "year":
minimum_width, maximum_width = 110, 190
elif scale == "day":
minimum_width, maximum_width = 150, 260
else:
minimum_width, maximum_width = 132, 220
estimated_width = int(28 + weighted_length * 7)
return max(minimum_width, min(maximum_width, estimated_width))
def _timeline_event_label_span(width_px: int, scale: str) -> float:
return width_px / max(_timeline_column_width(scale), 1)
def _timeline_state_lane_height(track_count: int) -> int:
return max(1, track_count) * 38 + 22
def _timeline_state_segment_center(track_index: int) -> int:
return 25 + max(track_index, 0) * 38
def _timeline_event_node_anchor(track_index: int) -> int:
return 23 + max(track_index, 0) * 54
def _timeline_state_node_connector_height(
state_track_count: int,
state_track_index: int,
event_track_index: int,
) -> int:
return (
_timeline_state_lane_height(state_track_count)
+ 8
+ _timeline_event_node_anchor(event_track_index)
- _timeline_state_segment_center(state_track_index)
)
def _build_timeline_rows(
rows: list[dict[str, object]],
simulation_time: datetime,
range_start: datetime,
range_end: datetime,
scale: str,
) -> tuple[list[dict[str, object]], dict[str, int]]:
bucket_origin = _timeline_bucket_start(range_start, scale)
bucket_count = len(_timeline_scale_labels(range_start, range_end, scale))
timeline_rows = []
state_count = 0
event_count = 0
for row in rows:
asset = row["asset"]
state_windows = _timeline_state_windows(asset)
segments = []
events = []
for entry in sorted(asset.log_entries, key=lambda item: _as_utc(item.start_at) or range_start):
if entry.entry_kind == "state":
if not _timeline_state_overlaps(entry, range_start, range_end, state_windows):
continue
state_start_at, state_end_at = state_windows.get(entry.id, (_as_utc(entry.start_at) or range_start, _as_utc(entry.end_at)))
start_at = max(state_start_at, range_start)
exclusive_end = state_end_at
if exclusive_end is not None:
exclusive_end = exclusive_end - timedelta(microseconds=1)
end_at = min(exclusive_end or range_end, range_end)
if end_at < start_at:
end_at = start_at
actual_start_position = _timeline_position(start_at, bucket_origin, scale, bucket_count)
actual_end_position = _timeline_position(end_at, bucket_origin, scale, bucket_count)
if actual_end_position <= actual_start_position:
actual_end_position = min(float(bucket_count), actual_start_position + 0.02)
state_count += 1
segments.append(
{
"entry_id": entry.id,
"label": entry.title,
"tone": _timeline_segment_tone(entry, simulation_time),
"url": _timeline_entry_edit_url(entry.asset_id, entry.id, simulation_time),
"start_position": actual_start_position,
"body_end_position": actual_end_position,
"display_end_position": actual_end_position,
}
)
for state_node in sorted(entry.state_nodes, key=lambda item: _as_utc(item.at_time) or range_start):
node_at = _as_utc(state_node.at_time)
if node_at is None or node_at < range_start or node_at > range_end:
continue
full_label_parts = [state_node.title, entry.title]
if state_node.detail:
full_label_parts.append(state_node.detail)
event_position = _timeline_position(node_at, bucket_origin, scale, bucket_count)
label_width_px = _timeline_event_label_width_px(state_node.title, scale)
label_span = _timeline_event_label_span(label_width_px, scale)
label_side = "left" if event_position >= bucket_count * 0.6 else "right"
track_start_position = max(0.0, event_position - label_span) if label_side == "left" else event_position
track_end_position = event_position if label_side == "left" else min(float(bucket_count), event_position + label_span)
events.append(
{
"label": state_node.title,
"full_label": " · ".join(full_label_parts),
"url": _timeline_entry_edit_url(asset.id, entry.id, simulation_time),
"position": event_position,
"label_width_px": label_width_px,
"label_span": label_span,
"label_side": label_side,
"track_start_position": track_start_position,
"track_end_position": track_end_position,
"state_node": True,
"parent_entry_id": entry.id,
}
)
event_count += 1
continue
if not _entry_overlaps_timeline(entry, range_start, range_end):
continue
event_position = _timeline_position(_as_utc(entry.start_at) or range_start, bucket_origin, scale, bucket_count)
label_width_px = _timeline_event_label_width_px(entry.title, scale)
label_span = _timeline_event_label_span(label_width_px, scale)
label_side = "left" if event_position >= bucket_count * 0.6 else "right"
track_start_position = max(0.0, event_position - label_span) if label_side == "left" else event_position
track_end_position = event_position if label_side == "left" else min(float(bucket_count), event_position + label_span)
events.append(
{
"label": entry.title,
"full_label": entry.title,
"url": _timeline_entry_edit_url(asset.id, entry.id, simulation_time),
"position": event_position,
"label_width_px": label_width_px,
"label_span": label_span,
"label_side": label_side,
"track_start_position": track_start_position,
"track_end_position": track_end_position,
"state_node": False,
}
)
event_count += 1
segments.sort(key=lambda item: float(item["start_position"]))
events.sort(key=lambda item: float(item["position"]))
state_minimum_span, state_gap = _timeline_state_track_settings(scale)
event_minimum_span, event_gap = _timeline_event_track_settings(scale)
state_track_count = _assign_timeline_tracks(
segments,
start_key="start_position",
end_key="display_end_position",
minimum_span=state_minimum_span,
gap=state_gap,
)
segment_tracks = {segment["entry_id"]: int(segment["track"]) for segment in segments}
event_track_count = _assign_timeline_tracks(
events,
start_key="track_start_position",
end_key="track_end_position",
minimum_span=event_minimum_span,
gap=event_gap,
)
for segment in segments:
display_span = max(float(segment["display_end_position"]) - float(segment["start_position"]), 0.02)
body_span = max(float(segment["body_end_position"]) - float(segment["start_position"]), 0.02)
segment["left_pct"] = _timeline_percent(float(segment["start_position"]), bucket_count)
segment["track_width_pct"] = _timeline_percent(display_span, bucket_count)
segment["body_width_pct"] = f"{(body_span / display_span) * 100:.6f}%"
segment["compact"] = (scale == "year" and body_span < 0.45) or (state_track_count > 1 and body_span < 0.75)
for event in events:
event["left_pct"] = _timeline_percent(float(event["position"]), bucket_count)
if not event.get("state_node"):
continue
parent_track = segment_tracks.get(event["parent_entry_id"])
if parent_track is None:
continue
event["connector_height_px"] = f"{_timeline_state_node_connector_height(state_track_count, parent_track, int(event['track']))}px"
if not segments and not events:
continue
timeline_rows.append(
{
"asset_id": asset.id,
"name": asset.name,
"asset_type": asset.asset_type,
"current_state": row["state_display"],
"segments": segments,
"events": events,
"state_track_count": state_track_count,
"event_track_count": event_track_count,
}
)
timeline_rows.sort(key=lambda item: item["name"].casefold())
return timeline_rows, {"states": state_count, "events": event_count}
def _page_url(endpoint: str, query_args: dict[str, object], page_number: int) -> str:
params = {key: value for key, value in query_args.items() if value not in (None, "")}
params["page"] = page_number
return url_for(endpoint, **params)
def _page_markers(total_pages: int, current_page: int) -> list[int | None]:
if total_pages <= 9:
return list(range(1, total_pages + 1))
markers: list[int | None] = [1]
start = max(2, current_page - 2)
end = min(total_pages - 1, current_page + 2)
if start > 2:
markers.append(None)
markers.extend(range(start, end + 1))
if end < total_pages - 1:
markers.append(None)
markers.append(total_pages)
return markers
def _paginate(
items: list[Any],
endpoint: str,
query_args: dict[str, object],
requested_page: int,
) -> tuple[list[Any], dict[str, Any]]:
total_count = len(items)
total_pages = max(1, ceil(total_count / DEFAULT_PAGE_SIZE))
page = min(max(requested_page, 1), total_pages)
start_index = (page - 1) * DEFAULT_PAGE_SIZE
page_items = items[start_index : start_index + DEFAULT_PAGE_SIZE]
markers = _page_markers(total_pages, page)
page_links = []
for marker in markers:
if marker is None:
page_links.append({"ellipsis": True})
continue
page_links.append(
{
"ellipsis": False,
"number": marker,
"url": _page_url(endpoint, query_args, marker),
"current": marker == page,
}
)
context = {
"page": page,
"total_pages": total_pages,
"total_count": total_count,
"has_prev": page > 1,
"has_next": page < total_pages,
"prev_url": _page_url(endpoint, query_args, page - 1) if page > 1 else None,
"next_url": _page_url(endpoint, query_args, page + 1) if page < total_pages else None,
"page_links": page_links,
}
return page_items, context
def _entity_matches_query(values: list[str | None], search_term: str) -> bool:
normalized = search_term.casefold()
return any(normalized in (value or "").casefold() for value in values)
def _split_multi_value_text(value: str | None) -> list[str]:
if value is None:
return []
labels: list[str] = []
for part in str(value).split(";"):
cleaned = _normalized_text(part)
if cleaned is None or cleaned in labels:
continue
labels.append(cleaned)
return labels
def _normalize_multi_value_text(value: str | None) -> str | None:
labels = _split_multi_value_text(value)
if not labels:
return None
return "; ".join(labels)
def _display_multi_value_text(value: str | None, separator: str = " · ") -> str | None:
labels = _split_multi_value_text(value)
if not labels:
return None
return separator.join(labels)
def _mission_label_search_key(value: str | None) -> str:
return re.sub(r"[\s\-_]+", "", (value or "").casefold())
def _mission_label_matches_query(mission_label: str | None, search_term: str) -> bool:
if mission_label is None:
return False
raw_query = search_term.casefold().strip()
raw_label = mission_label.casefold().strip()
if not raw_query or not raw_label:
return False
normalized_query = _mission_label_search_key(search_term)
normalized_label = _mission_label_search_key(mission_label)
if normalized_query and normalized_query in normalized_label:
return True
if raw_query in raw_label:
return True
query_tokens = [token for token in re.split(r"[\s\-_]+", raw_query) if token]
label_tokens = [token for token in re.split(r"[\s\-_]+", raw_label) if token]
if query_tokens and all(
any(
token in label_token
or label_token in token
or SequenceMatcher(None, token, label_token).ratio() >= 0.86
for label_token in label_tokens
)
for token in query_tokens
):
return True
return False
@web_bp.get("/")
def dashboard() -> str:
return render_template(
"dashboard.html",
title="Dashboard",
metrics=PROJECT_METRICS,
modules=MODULES,
duplicate_conflicts=48,
duplicate_conflicts_with_fuel=19,
)
@web_bp.get("/fuel-converter")
def fuel_converter() -> str:
return render_template(
"fuel_converter.html",
title="Fuel Converter",
fuel_factors=list_fuel_factors(),
)
@web_bp.get("/engines")
def engine_list() -> str:
search_term = request.args.get("q", "").strip()
fuel_filter = request.args.get("fuel_type", "").strip()
cycle_filter = request.args.get("cycle", "").strip()
mod_source_filter = request.args.get("mod_source", "").strip()
sort_option = request.args.get("sort", "engine_name_asc").strip()
requested_page = request.args.get("page", default=1, type=int) or 1
catalog_rows = _build_catalog_rows()
fuel_options = sorted({fuel for row in catalog_rows for fuel in row.fuel_types}, key=str.casefold)
cycle_options = sorted({row.family.cycle for row in catalog_rows if row.family.cycle}, key=str.casefold)
mod_source_options = sorted({source for row in catalog_rows for source in row.mod_sources}, key=str.casefold)
filtered_rows = []
for row in catalog_rows:
if search_term and not _matches_catalog_search(row, search_term):
continue
if fuel_filter and fuel_filter not in row.fuel_types:
continue
if cycle_filter and cycle_filter != (row.family.cycle or ""):
continue
if mod_source_filter and mod_source_filter not in row.mod_sources:
continue
filtered_rows.append(row)
normalized_sort = _sort_catalog_rows(filtered_rows, sort_option)
query_args = {
"q": search_term,
"fuel_type": fuel_filter,
"cycle": cycle_filter,
"mod_source": mod_source_filter,
}
page_rows, paging = _paginate(filtered_rows, "web.engine_list", query_args, requested_page)
sort_columns = _build_engine_sort_links(query_args, normalized_sort)
return render_template(
"engine_list.html",
title="Engine Catalog",
catalog_rows=page_rows,
search_term=search_term,
fuel_filter=fuel_filter,
cycle_filter=cycle_filter,
mod_source_filter=mod_source_filter,
sort_option=normalized_sort,
sort_columns=sort_columns,
fuel_options=fuel_options,
cycle_options=cycle_options,
mod_source_options=mod_source_options,
total_catalog_count=len(catalog_rows),
**paging,
)
@web_bp.route("/engines/new", methods=["GET", "POST"])
def engine_new() -> str | object:
family = EngineFamily()
variant = EngineVariant()
form_options = _engine_form_options()
if request.method == "POST":
try:
_apply_fields(family, request.form, FAMILY_FORM_FIELDS)
_apply_fields(variant, request.form, VARIANT_CREATE_FIELDS, prefix="initial-")
_normalize_variant_ignitions(variant)
variant.family = family
db.session.add(family)
db.session.add(variant)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
except IntegrityError:
db.session.rollback()
flash("创建失败:Engine 名称可能已存在。", "error")
else:
flash("引擎已创建。", "success")
return redirect(url_for("web.engine_detail", family_id=family.id))
return render_template(
"engine_create.html",
title="Create Engine",
family_fields=FAMILY_FORM_FIELDS,
variant_fields=VARIANT_CREATE_FIELDS,
form_options=form_options,
custom_option_value=CUSTOM_OPTION_VALUE,
)
@web_bp.get("/engines/<uuid:family_id>")
def engine_detail(family_id: object) -> str:
family = _load_engine_family_or_404(family_id)
variant_rows = [
{
"variant": variant,
"sl_thrust_kn": _estimate_sl_thrust_kn(variant),
"vac_thrust_kn": variant.max_thrust_kn,
}
for variant in family.variants
]
return render_template(
"engine_detail.html",
title=family.engine_name,
family=family,
variant_rows=variant_rows,
)
@web_bp.route("/engines/<uuid:family_id>/edit", methods=["GET", "POST"])
def engine_edit(family_id: object) -> str:
family = _load_engine_family_or_404(family_id)
form_options = _engine_form_options()
if request.method == "POST":
try:
_apply_fields(family, request.form, FAMILY_FORM_FIELDS)
for variant in family.variants:
_apply_fields(
variant,
request.form,
VARIANT_FORM_FIELDS,
prefix=f"variant-{variant.id}-",
)
_normalize_variant_ignitions(variant)
_refresh_variant_derived_fields(variant)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
except IntegrityError:
db.session.rollback()
flash("保存失败:Engine 名称必须唯一。", "error")
else:
flash("引擎数据已保存。", "success")
return redirect(url_for("web.engine_detail", family_id=family.id))
return render_template(
"engine_edit.html",
title=f"Edit · {family.engine_name}",
family=family,
family_fields=FAMILY_FORM_FIELDS,
variant_fields=VARIANT_FORM_FIELDS,
variant_create_fields=VARIANT_CREATE_FIELDS,
form_options=form_options,
custom_option_value=CUSTOM_OPTION_VALUE,
)
@web_bp.post("/engines/<uuid:family_id>/variants/add")
def engine_variant_add(family_id: object) -> object:
family = _load_engine_family_or_404(family_id)
variant = EngineVariant(family=family)
try:
_apply_fields(variant, request.form, VARIANT_CREATE_FIELDS, prefix="add-")
_normalize_variant_ignitions(variant)
db.session.add(variant)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
else:
flash("已新增配置项。", "success")
return redirect(url_for("web.engine_edit", family_id=family.id))
@web_bp.post("/engines/<uuid:family_id>/variants/<uuid:variant_id>/delete")
def engine_variant_delete(family_id: object, variant_id: object) -> object:
family = _load_engine_family_or_404(family_id)
variant = db.session.get(EngineVariant, variant_id)
if variant is None or variant.family_id != family.id:
abort(404)
db.session.delete(variant)
db.session.commit()
flash("配置项已删除。", "success")
return redirect(url_for("web.engine_edit", family_id=family.id))
@web_bp.post("/engines/<uuid:family_id>/delete")
def engine_delete(family_id: object) -> object:
family = _load_engine_family_or_404(family_id)
db.session.delete(family)
db.session.commit()
flash(f"已删除引擎:{family.engine_name}", "success")
return redirect(url_for("web.engine_list"))
@web_bp.get("/communications")
def communication_list() -> str:
search_term = request.args.get("q", "").strip()
source_filter = request.args.get("source", "").strip()
type_filter = request.args.get("antenna_type", "").strip()
sort_option = request.args.get("sort", "part_name").strip()
requested_page = request.args.get("page", default=1, type=int) or 1
rows = db.session.execute(select(CommunicationPart).order_by(CommunicationPart.part_name.asc())).scalars().all()
source_options = sorted({row.source for row in rows if row.source}, key=str.casefold)
type_options = sorted({row.antenna_type for row in rows if row.antenna_type}, key=str.casefold)
filtered_rows = []
for row in rows:
if search_term and not _entity_matches_query(
[row.part_name, row.display_name, row.source, row.antenna_type],
search_term,
):
continue
if source_filter and source_filter != (row.source or ""):
continue
if type_filter and type_filter != (row.antenna_type or ""):
continue
filtered_rows.append(row)
normalized_sort = _sort_communications(filtered_rows, sort_option)
query_args = {
"q": search_term,
"source": source_filter,
"antenna_type": type_filter,
"sort": normalized_sort,
}
page_rows, paging = _paginate(filtered_rows, "web.communication_list", query_args, requested_page)
return render_template(
"communication_list.html",
title="Communication Parts",
rows=page_rows,
search_term=search_term,
source_filter=source_filter,
antenna_type_filter=type_filter,
sort_option=normalized_sort,
sort_options=COMM_SORT_OPTIONS,
source_options=source_options,
type_options=type_options,
total_catalog_count=len(rows),
**paging,
)
@web_bp.route("/communications/new", methods=["GET", "POST"])
def communication_new() -> str | object:
part = CommunicationPart()
if request.method == "POST":
try:
_apply_fields(part, request.form, COMMUNICATION_FIELDS)
db.session.add(part)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
else:
flash("已新增通信部件。", "success")
return redirect(url_for("web.communication_edit", part_id=part.id))
return render_template(
"resource_edit.html",
title="Create Communication",
eyebrow="Communication",
heading="新增通信部件",
description="新建一条通信部件记录。",
fields=COMMUNICATION_FIELDS,
values=_collect_form_values(part, COMMUNICATION_FIELDS),
cancel_url=url_for("web.communication_list"),
submit_label="创建",
delete_url=None,
)
@web_bp.route("/communications/<uuid:part_id>/edit", methods=["GET", "POST"])
def communication_edit(part_id: object) -> str | object:
part = _load_model_or_404(CommunicationPart, part_id)
if request.method == "POST":
try:
_apply_fields(part, request.form, COMMUNICATION_FIELDS)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
else:
flash("通信部件已保存。", "success")
return redirect(url_for("web.communication_list"))
return render_template(
"resource_edit.html",
title=f"Edit · {part.part_name}",
eyebrow="Communication",
heading=f"编辑 {part.part_name}",
description="修改通信部件字段。",
fields=COMMUNICATION_FIELDS,
values=_collect_form_values(part, COMMUNICATION_FIELDS),
cancel_url=url_for("web.communication_list"),
submit_label="保存",
delete_url=url_for("web.communication_delete", part_id=part.id),
delete_label="删除通信部件",
delete_confirm="确认删除这个通信部件?",
)
@web_bp.post("/communications/<uuid:part_id>/delete")
def communication_delete(part_id: object) -> object:
part = _load_model_or_404(CommunicationPart, part_id)
db.session.delete(part)
db.session.commit()
flash("通信部件已删除。", "success")
return redirect(url_for("web.communication_list"))
@web_bp.get("/tanks")
def tank_list() -> str:
search_term = request.args.get("q", "").strip()
fuel_filter = request.args.get("fuel_type", "").strip()
source_filter = request.args.get("source", "").strip()
sort_option = request.args.get("sort", "tank_name").strip()
requested_page = request.args.get("page", default=1, type=int) or 1
rows = db.session.execute(select(TankSpec).order_by(TankSpec.tank_name.asc())).scalars().all()
fuel_options = sorted({row.fuel_type for row in rows if row.fuel_type}, key=str.casefold)
source_options = sorted({row.source for row in rows if row.source}, key=str.casefold)
filtered_rows = []
for row in rows:
if search_term and not _entity_matches_query(
[row.tank_name, row.vehicle_name, row.source, row.fuel_type],
search_term,
):
continue
if fuel_filter and fuel_filter != (row.fuel_type or ""):
continue
if source_filter and source_filter != (row.source or ""):
continue
filtered_rows.append(row)
normalized_sort = _sort_tanks(filtered_rows, sort_option)
query_args = {
"q": search_term,
"fuel_type": fuel_filter,
"source": source_filter,
"sort": normalized_sort,
}
page_rows, paging = _paginate(filtered_rows, "web.tank_list", query_args, requested_page)
return render_template(
"tank_list.html",
title="Tank Specs",
rows=page_rows,
search_term=search_term,
fuel_filter=fuel_filter,
source_filter=source_filter,
sort_option=normalized_sort,
sort_options=TANK_SORT_OPTIONS,
fuel_options=fuel_options,
source_options=source_options,
total_catalog_count=len(rows),
**paging,
)
@web_bp.route("/tanks/new", methods=["GET", "POST"])
def tank_new() -> str | object:
tank = TankSpec()
if request.method == "POST":
try:
_apply_fields(tank, request.form, TANK_FIELDS)
db.session.add(tank)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
else:
flash("已新增燃料箱记录。", "success")
return redirect(url_for("web.tank_edit", tank_id=tank.id))
return render_template(
"resource_edit.html",
title="Create Tank",
eyebrow="Tank",
heading="新增燃料箱记录",
description="录入一条 Tank Chart 数据。",
fields=TANK_FIELDS,
values=_collect_form_values(tank, TANK_FIELDS),
cancel_url=url_for("web.tank_list"),
submit_label="创建",
delete_url=None,
)
@web_bp.route("/tanks/<uuid:tank_id>/edit", methods=["GET", "POST"])
def tank_edit(tank_id: object) -> str | object:
tank = _load_model_or_404(TankSpec, tank_id)
if request.method == "POST":
try:
_apply_fields(tank, request.form, TANK_FIELDS)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
else:
flash("燃料箱记录已保存。", "success")
return redirect(url_for("web.tank_list"))
return render_template(
"resource_edit.html",
title=f"Edit · {tank.tank_name}",
eyebrow="Tank",
heading=f"编辑 {tank.tank_name}",
description="修改燃料箱字段。",
fields=TANK_FIELDS,
values=_collect_form_values(tank, TANK_FIELDS),
cancel_url=url_for("web.tank_list"),
submit_label="保存",
delete_url=url_for("web.tank_delete", tank_id=tank.id),
delete_label="删除燃料箱记录",
delete_confirm="确认删除这个燃料箱记录?",
)
@web_bp.post("/tanks/<uuid:tank_id>/delete")
def tank_delete(tank_id: object) -> object:
tank = _load_model_or_404(TankSpec, tank_id)
db.session.delete(tank)
db.session.commit()
flash("燃料箱记录已删除。", "success")
return redirect(url_for("web.tank_list"))
@web_bp.get("/vehicles")
def vehicle_list() -> str:
search_term = request.args.get("q", "").strip()
source_filter = request.args.get("source", "").strip()
sort_option = request.args.get("sort", "vehicle_name").strip()
requested_page = request.args.get("page", default=1, type=int) or 1
rows = db.session.execute(select(VehicleCost).order_by(VehicleCost.vehicle_name.asc())).scalars().all()
source_options = sorted({row.source for row in rows if row.source}, key=str.casefold)
filtered_rows = []
for row in rows:
if search_term and not _entity_matches_query([row.vehicle_name, row.source], search_term):
continue
if source_filter and source_filter != (row.source or ""):
continue
filtered_rows.append(row)
normalized_sort = _sort_vehicles(filtered_rows, sort_option)
query_args = {"q": search_term, "source": source_filter, "sort": normalized_sort}
page_rows, paging = _paginate(filtered_rows, "web.vehicle_list", query_args, requested_page)
return render_template(
"vehicle_list.html",
title="Vehicle Costs",
rows=page_rows,
search_term=search_term,
source_filter=source_filter,
sort_option=normalized_sort,
sort_options=VEHICLE_SORT_OPTIONS,
source_options=source_options,
total_catalog_count=len(rows),
**paging,
)
@web_bp.route("/vehicles/new", methods=["GET", "POST"])
def vehicle_new() -> str | object:
vehicle = VehicleCost()
if request.method == "POST":
try:
_apply_fields(vehicle, request.form, VEHICLE_FIELDS)
db.session.add(vehicle)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
else:
flash("已新增载具成本记录。", "success")
return redirect(url_for("web.vehicle_edit", vehicle_id=vehicle.id))
return render_template(
"resource_edit.html",
title="Create Vehicle",
eyebrow="Vehicle",
heading="新增载具成本",
description="录入一条 Vehicle Cost 数据。",
fields=VEHICLE_FIELDS,
values=_collect_form_values(vehicle, VEHICLE_FIELDS),
cancel_url=url_for("web.vehicle_list"),
submit_label="创建",
delete_url=None,
)
@web_bp.route("/vehicles/<uuid:vehicle_id>/edit", methods=["GET", "POST"])
def vehicle_edit(vehicle_id: object) -> str | object:
vehicle = _load_model_or_404(VehicleCost, vehicle_id)
if request.method == "POST":
try:
_apply_fields(vehicle, request.form, VEHICLE_FIELDS)
db.session.commit()
except ValueError as exc:
db.session.rollback()
flash(str(exc), "error")
else:
flash("载具成本记录已保存。", "success")
return redirect(url_for("web.vehicle_list"))
return render_template(
"resource_edit.html",
title=f"Edit · {vehicle.vehicle_name}",
eyebrow="Vehicle",
heading=f"编辑 {vehicle.vehicle_name}",
description="修改发射成本字段。",
fields=VEHICLE_FIELDS,
values=_collect_form_values(vehicle, VEHICLE_FIELDS),
cancel_url=url_for("web.vehicle_list"),
submit_label="保存",
delete_url=url_for("web.vehicle_delete", vehicle_id=vehicle.id),
delete_label="删除载具成本记录",
delete_confirm="确认删除这个载具成本记录?",
)
@web_bp.post("/vehicles/<uuid:vehicle_id>/delete")
def vehicle_delete(vehicle_id: object) -> object:
vehicle = _load_model_or_404(VehicleCost, vehicle_id)
db.session.delete(vehicle)
db.session.commit()
flash("载具成本记录已删除。", "success")
return redirect(url_for("web.vehicle_list"))
@web_bp.get("/assets")
def asset_list() -> str:
search_term = request.args.get("q", "").strip()
asset_type_filter = request.args.get("asset_type", "").strip()
sort_option = request.args.get("sort", "name_asc").strip()
requested_page = request.args.get("page", default=1, type=int) or 1
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
rows = db.session.execute(
select(Asset).options(selectinload(Asset.log_entries)).order_by(Asset.name.asc())
).scalars().all()
asset_type_options = sorted({row.asset_type for row in rows if row.asset_type}, key=str.casefold)
filtered_rows = []
for row in rows:
if search_term and not _entity_matches_query([row.name, row.asset_type, row.program, row.home_region], search_term):
continue
if asset_type_filter and asset_type_filter != (row.asset_type or ""):
continue
filtered_rows.append(row)
normalized_sort = _sort_assets(filtered_rows, sort_option)
query_args = {
"q": search_term,
"asset_type": asset_type_filter,
"sort": normalized_sort,
"sim_time": _format_datetime_input(simulation_time),
}
page_rows, paging = _paginate(filtered_rows, "web.asset_list", query_args, requested_page)
catalog_rows = [_build_asset_catalog_row(asset, simulation_time) for asset in page_rows]
return render_template(
"asset_list.html",
title="任务资产",
preview_title="任务资产目录",
preview_eyebrow="Mission Asset",
rows=catalog_rows,
search_term=search_term,
asset_type_filter=asset_type_filter,
sort_option=normalized_sort,
sort_options=ASSET_SORT_OPTIONS,
asset_type_options=asset_type_options,
total_catalog_count=len(rows),
metrics=_build_asset_catalog_metrics(rows, filtered_rows, simulation_time),
preview_tab="asset",
board_nav_urls=_mission_ops_nav_urls(simulation_time),
simulation_time_display=_format_datetime_display(simulation_time),
simulation_time_input=_format_datetime_input(simulation_time),
header_time_action=url_for("web.asset_list"),
header_time_field_id="asset-list-header-sim-time",
header_time_submit_label="Apply Time",
header_hidden_fields=_build_header_hidden_fields(
{
"q": search_term,
"asset_type": asset_type_filter,
"sort": normalized_sort,
}
),
**paging,
)
@web_bp.get("/missions")
def mission_list() -> str:
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
search_term = request.args.get("q", "").strip()
asset_id_filter = request.args.get("asset_id", "").strip()
mission_label_filter = request.args.get("mission_label", "").strip()
location_filter = request.args.get("location", "").strip()
try:
range_start = _parse_datetime(request.args.get("start"), "任务开始时间")
except ValueError as exc:
range_start = None
flash(str(exc), "error")
try:
range_end = _parse_datetime(request.args.get("end"), "任务结束时间")
except ValueError as exc:
range_end = None
flash(str(exc), "error")
if range_start is not None and range_end is not None and range_end < range_start:
flash("任务结束时间不能早于开始时间。", "error")
range_end = range_start
all_rows = []
for asset in _load_assets_with_logs():
if asset.is_retired:
continue
for entry in sorted(
asset.log_entries,
key=lambda item: _as_utc(item.start_at) or _default_simulation_time(),
reverse=True,
):
mission_labels = _split_multi_value_text(entry.mission_label)
if not mission_labels:
continue
state_labels = _split_multi_value_text(entry.state_label)
status = _classify_asset_entry(entry, simulation_time)
for mission_label in mission_labels:
all_rows.append(
{
"asset": asset,
"asset_id": str(asset.id),
"asset_name": asset.name,
"asset_type": asset.asset_type or "-",
"entry": entry,
"entry_id": entry.id,
"mission_label": mission_label,
"mission_labels": mission_labels,
"state_labels": state_labels,
"title": entry.title,
"mode_label": "Event Point" if entry.entry_kind == "event" else "State Interval",
"status_label": status["label"],
"status_tone": status["tone"],
"timestamp": _format_datetime_display(entry.start_at),
"end_timestamp": _format_datetime_display(entry.end_at) if entry.end_at else None,
"location": entry.location or "-",
"summary": entry.summary or "未填写摘要。",
"sort_at": _as_utc(entry.start_at) or _default_simulation_time(),
}
)
asset_options = [
{"value": asset_id, "label": asset_name}
for asset_id, asset_name in sorted(
{item["asset_id"]: item["asset_name"] for item in all_rows}.items(),
key=lambda item: item[1].casefold(),
)
]
mission_label_options = sorted({row["mission_label"] for row in all_rows}, key=str.casefold)
location_options = sorted(
{row["location"] for row in all_rows if row["location"] and row["location"] != "-"},
key=str.casefold,
)
filtered_rows = []
for row in all_rows:
if asset_id_filter and asset_id_filter != row["asset_id"]:
continue
if mission_label_filter and mission_label_filter != row["mission_label"]:
continue
if location_filter and location_filter != row["location"]:
continue
if search_term and not _mission_label_matches_query(row["mission_label"], search_term):
continue
if not _entry_matches_date_range(row["entry"], range_start, range_end):
continue
filtered_rows.append(row)
grouped_rows: dict[str, dict[str, object]] = {}
for row in filtered_rows:
mission_label = str(row["mission_label"])
group = grouped_rows.setdefault(
mission_label,
{
"mission_label": mission_label,
"entries": [],
"asset_names": set(),
"locations": set(),
"active_count": 0,
"planned_count": 0,
"latest_at": row["sort_at"],
},
)
group["entries"].append(row)
group["asset_names"].add(str(row["asset_name"]))
if row["location"] != "-":
group["locations"].add(str(row["location"]))
if row["status_tone"] == "active":
group["active_count"] += 1
elif row["status_tone"] == "planned":
group["planned_count"] += 1
if row["sort_at"] > group["latest_at"]:
group["latest_at"] = row["sort_at"]
mission_groups = []
for group in grouped_rows.values():
entries = sorted(group["entries"], key=lambda item: item["sort_at"], reverse=True)
asset_names = sorted(group["asset_names"], key=str.casefold)
locations = sorted(group["locations"], key=str.casefold)
if group["active_count"]:
group_status = "Active"
group_rank = 0
elif group["planned_count"]:
group_status = "Upcoming"
group_rank = 1
else:
group_status = "Recorded"
group_rank = 2
mission_groups.append(
{
"mission_label": group["mission_label"],
"entries": entries,
"entry_count": len(entries),
"asset_names": asset_names,
"asset_count": len(asset_names),
"locations": locations or ["-"],
"status_label": group_status,
"group_rank": group_rank,
"latest_timestamp": _format_datetime_display(group["latest_at"]),
}
)
mission_groups.sort(key=lambda item: (item["group_rank"], item["mission_label"].casefold()))
selected_asset_name = next(
(option["label"] for option in asset_options if option["value"] == asset_id_filter),
"全部资产",
)
return render_template(
"mission_list.html",
title="任务运营 · 任务",
preview_title="任务",
preview_eyebrow="Mission Label",
preview_tab="mission",
mission_groups=mission_groups,
search_term=search_term,
asset_id_filter=asset_id_filter,
mission_label_filter=mission_label_filter,
location_filter=location_filter,
asset_options=asset_options,
mission_label_options=mission_label_options,
location_options=location_options,
selected_asset_name=selected_asset_name,
total_entry_count=len(all_rows),
filtered_entry_count=len(filtered_rows),
metrics=[
{"label": "Visible Missions", "value": str(len(mission_groups))},
{"label": "Visible Entries", "value": str(len(filtered_rows))},
{"label": "Active Missions", "value": str(sum(1 for group in mission_groups if group["status_label"] == "Active"))},
],
simulation_time_display=_format_datetime_display(simulation_time),
simulation_time_input=_format_datetime_input(simulation_time),
range_start_input=_format_datetime_input(range_start),
range_end_input=_format_datetime_input(range_end),
header_time_action=url_for("web.mission_list"),
header_time_field_id="mission-list-header-sim-time",
header_time_submit_label="Apply Time",
header_hidden_fields=_build_header_hidden_fields(
{
"q": search_term,
"asset_id": asset_id_filter,
"mission_label": mission_label_filter,
"location": location_filter,
"start": _format_datetime_input(range_start),
"end": _format_datetime_input(range_end),
}
),
board_nav_urls=_mission_ops_nav_urls(simulation_time),
)
@web_bp.route("/assets/new", methods=["GET", "POST"])
def asset_new() -> str | object:
asset = Asset()
form_options = _asset_form_options()
simulation_time = _resolve_simulation_time(request.form.get("sim_time") if request.method == "POST" else request.args.get("sim_time"))
simulation_time_input = _format_datetime_input(simulation_time)
values = _collect_form_values(asset, ASSET_FIELDS)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
_apply_fields(asset, request.form, ASSET_FIELDS)
db.session.add(asset)
db.session.commit()
except FormValidationError as exc:
db.session.rollback()
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
except IntegrityError:
db.session.rollback()
validation_modal = _build_validation_modal(
"创建失败:Asset 名称必须唯一。",
field_errors={"name": "Asset 名称必须唯一。"},
)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
else:
flash("资产已创建。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
return render_template(
"resource_edit.html",
title="新增任务资产",
eyebrow="Mission Asset",
heading="新增任务资产",
description=None,
fields=ASSET_FIELDS,
values=values,
form_options=form_options,
custom_option_value=CUSTOM_OPTION_VALUE,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_list", sim_time=simulation_time_input),
submit_label="创建",
delete_url=None,
hidden_fields={"sim_time": simulation_time_input},
mission_ops_tabs=True,
preview_tab="asset",
board_nav_urls=_mission_ops_nav_urls(simulation_time),
)
@web_bp.route("/assets/<uuid:asset_id>/edit", methods=["GET", "POST"])
def asset_edit(asset_id: object) -> str | object:
asset = _load_asset_or_404(asset_id)
form_options = _asset_form_options()
simulation_time = _resolve_simulation_time(request.form.get("sim_time") if request.method == "POST" else request.args.get("sim_time"))
simulation_time_input = _format_datetime_input(simulation_time)
values = _collect_form_values(asset, ASSET_FIELDS)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
_apply_fields(asset, request.form, ASSET_FIELDS)
db.session.commit()
except FormValidationError as exc:
db.session.rollback()
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
except IntegrityError:
db.session.rollback()
validation_modal = _build_validation_modal(
"保存失败:Asset 名称必须唯一。",
field_errors={"name": "Asset 名称必须唯一。"},
)
field_errors = dict(validation_modal["field_errors"])
values = _collect_submitted_form_values(request.form, ASSET_FIELDS)
else:
flash("资产信息已保存。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
return render_template(
"resource_edit.html",
title=f"编辑任务资产 · {asset.name}",
eyebrow="Mission Asset",
heading=f"编辑 {asset.name}",
description=None,
fields=ASSET_FIELDS,
values=values,
form_options=form_options,
custom_option_value=CUSTOM_OPTION_VALUE,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input),
submit_label="保存",
delete_url=url_for("web.asset_delete", asset_id=asset.id),
delete_label="删除资产",
delete_confirm="确认删除这个资产及其日志?",
hidden_fields={"sim_time": simulation_time_input},
delete_hidden_fields={"sim_time": simulation_time_input},
mission_ops_tabs=True,
preview_tab="asset",
board_nav_urls=_mission_ops_nav_urls(simulation_time),
)
@web_bp.post("/assets/<uuid:asset_id>/delete")
def asset_delete(asset_id: object) -> object:
asset = _load_asset_or_404(asset_id)
simulation_time_input = _normalized_text(request.form.get("sim_time"))
db.session.delete(asset)
db.session.commit()
flash("资产已删除。", "success")
redirect_kwargs: dict[str, str] = {}
if simulation_time_input is not None:
redirect_kwargs["sim_time"] = simulation_time_input
return redirect(url_for("web.asset_list", **redirect_kwargs))
@web_bp.post("/assets/import-log-book")
def asset_import_log_book() -> object:
redirect_kwargs: dict[str, str] = {}
simulation_time_input = _normalized_text(request.form.get("sim_time"))
if simulation_time_input is not None:
redirect_kwargs["sim_time"] = simulation_time_input
try:
summary = import_log_book_data(LOG_BOOK_PATH)
except FileNotFoundError as exc:
flash(str(exc), "error")
except Exception as exc:
flash(f"导入 log_book 失败:{exc}", "error")
else:
imported_sheet_names = ", ".join(item.sheet_name for item in summary.imported_assets) or "无"
total_states = sum(item.state_entries for item in summary.imported_assets)
total_events = sum(item.event_entries for item in summary.imported_assets)
flash(
f"已从 {summary.workbook} 导入 {len(summary.imported_assets)} 个载具 sheet{imported_sheet_names}),共 {total_states} 条 state 和 {total_events} 条 event。",
"success",
)
return redirect(url_for("web.asset_list", **redirect_kwargs))
@web_bp.get("/assets/<uuid:asset_id>")
def asset_detail(asset_id: object) -> str:
asset = _load_asset_or_404(asset_id)
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
requested_page = request.args.get("page", default=1, type=int) or 1
snapshot = _build_asset_snapshot(asset, simulation_time)
simulation_time_input = _format_datetime_input(simulation_time)
future_events = [
{
**item,
"edit_url": url_for(
"web.asset_entry_edit",
asset_id=asset.id,
entry_id=item["source_entry_id"],
sim_time=simulation_time_input,
),
}
for item in snapshot["future_events"]
]
try:
entry_range_start = _parse_datetime(request.args.get("start"), "日志开始时间")
except ValueError as exc:
entry_range_start = None
flash(str(exc), "error")
try:
entry_range_end = _parse_datetime(request.args.get("end"), "日志结束时间")
except ValueError as exc:
entry_range_end = None
flash(str(exc), "error")
if entry_range_start is not None and entry_range_end is not None and entry_range_end < entry_range_start:
flash("日志结束时间不能早于开始时间。", "error")
entry_range_end = entry_range_start
entry_rows = []
filtered_entries = [
entry
for entry in sorted(
asset.log_entries,
key=lambda item: _as_utc(item.start_at) or _default_simulation_time(),
reverse=True,
)
if _entry_matches_date_range(entry, entry_range_start, entry_range_end)
]
for entry in filtered_entries:
status = _classify_asset_entry(entry, simulation_time)
entry_rows.append(
{
"id": entry.id,
"mode": entry.entry_kind,
"mode_label": "Event Point" if entry.entry_kind == "event" else "State Interval",
"status_label": status["label"],
"title": entry.title,
"timestamp": _format_datetime_display(entry.start_at),
"end_timestamp": _format_datetime_display(entry.end_at) if entry.end_at else None,
"location": entry.location or "-",
"summary": entry.summary or "未填写摘要。",
"mission_label": _display_multi_value_text(entry.mission_label),
"state_label": _display_multi_value_text(entry.state_label),
"mission_labels": _split_multi_value_text(entry.mission_label),
"state_labels": _split_multi_value_text(entry.state_label),
}
)
paged_entries, pagination_context = _paginate(
entry_rows,
"web.asset_detail",
{
"asset_id": asset.id,
"sim_time": _format_datetime_input(simulation_time),
"start": _format_datetime_input(entry_range_start),
"end": _format_datetime_input(entry_range_end),
},
requested_page,
)
return render_template(
"asset_detail.html",
title=asset.name,
preview_title=asset.name,
preview_eyebrow="Mission Asset",
asset=asset,
asset_summary=snapshot,
future_events=future_events,
entries=paged_entries,
upcoming_events=future_events[:4],
total_entry_count=len(asset.log_entries),
filtered_entry_count=len(entry_rows),
metrics=[
{"label": "Asset Type", "value": asset.asset_type},
{"label": "Entries", "value": str(len(asset.log_entries))},
{"label": "Current State", "value": str(snapshot["current_state"])},
],
entry_filter_start_input=_format_datetime_input(entry_range_start),
entry_filter_end_input=_format_datetime_input(entry_range_end),
entry_page=pagination_context["page"],
entry_total_pages=pagination_context["total_pages"],
entry_has_prev=pagination_context["has_prev"],
entry_has_next=pagination_context["has_next"],
entry_prev_url=pagination_context["prev_url"],
entry_next_url=pagination_context["next_url"],
entry_page_links=pagination_context["page_links"],
preview_tab="asset",
board_nav_urls=_mission_ops_nav_urls(simulation_time),
simulation_time_display=_format_datetime_display(simulation_time),
simulation_time_input=simulation_time_input,
header_time_action=url_for("web.asset_detail", asset_id=asset.id),
header_time_field_id="asset-detail-header-sim-time",
header_time_submit_label="Apply Time",
header_hidden_fields=_build_header_hidden_fields(
{
"start": _format_datetime_input(entry_range_start),
"end": _format_datetime_input(entry_range_end),
"page": requested_page if requested_page > 1 else None,
}
),
)
@web_bp.route("/assets/<uuid:asset_id>/entries/new", methods=["GET", "POST"])
def asset_entry_new(asset_id: object) -> str | object:
asset = _load_asset_or_404(asset_id)
simulation_time = _resolve_simulation_time(request.form.get("sim_time") if request.method == "POST" else request.args.get("sim_time"))
simulation_time_input = _format_datetime_input(simulation_time)
requested_kind = _normalized_text(request.args.get("kind")) or "event"
if requested_kind not in {option["value"] for option in ASSET_ENTRY_KIND_OPTIONS}:
requested_kind = "event"
entry = AssetLogEntry(entry_kind=requested_kind)
form_values = _asset_entry_form_values(entry)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
_apply_asset_entry_fields(entry, request.form)
entry.asset = asset
db.session.add(entry)
db.session.commit()
except FormValidationError as exc:
db.session.rollback()
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
form_values = _asset_entry_form_values_from_form(request.form)
else:
flash("日志条目已新增。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
return render_template(
"asset_entry_form.html",
title=f"新增日志 · {asset.name}",
asset=asset,
heading=f"为 {asset.name} 新增日志条目",
description=None,
submit_label="创建条目",
form_values=form_values,
kind_options=ASSET_ENTRY_KIND_OPTIONS,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input),
delete_url=None,
simulation_time_input=simulation_time_input,
preview_tab="asset",
board_nav_urls=_mission_ops_nav_urls(simulation_time),
)
@web_bp.route("/assets/<uuid:asset_id>/entries/<uuid:entry_id>/edit", methods=["GET", "POST"])
def asset_entry_edit(asset_id: object, entry_id: object) -> str | object:
asset = _load_asset_or_404(asset_id)
entry = _load_asset_entry_or_404(asset.id, entry_id)
simulation_time = _resolve_simulation_time(request.form.get("sim_time") if request.method == "POST" else request.args.get("sim_time"))
simulation_time_input = _format_datetime_input(simulation_time)
form_values = _asset_entry_form_values(entry)
validation_modal: dict[str, object] | None = None
field_errors: dict[str, str] = {}
if request.method == "POST":
try:
_apply_asset_entry_fields(entry, request.form)
db.session.commit()
except FormValidationError as exc:
db.session.rollback()
validation_modal = _build_validation_modal(exc)
field_errors = dict(validation_modal["field_errors"])
form_values = _asset_entry_form_values_from_form(request.form)
else:
flash("日志条目已保存。", "success")
return redirect(url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input))
return render_template(
"asset_entry_form.html",
title=f"编辑日志 · {asset.name}",
asset=asset,
heading=f"编辑 {asset.name} 的日志条目",
description=None,
submit_label="保存条目",
form_values=form_values,
kind_options=ASSET_ENTRY_KIND_OPTIONS,
validation_modal=validation_modal,
field_errors=field_errors,
cancel_url=url_for("web.asset_detail", asset_id=asset.id, sim_time=simulation_time_input),
delete_url=url_for("web.asset_entry_delete", asset_id=asset.id, entry_id=entry.id),
simulation_time_input=simulation_time_input,
preview_tab="asset",
board_nav_urls=_mission_ops_nav_urls(simulation_time),
)
@web_bp.post("/assets/<uuid:asset_id>/entries/<uuid:entry_id>/delete")
def asset_entry_delete(asset_id: object, entry_id: object) -> object:
asset = _load_asset_or_404(asset_id)
entry = _load_asset_entry_or_404(asset.id, entry_id)
simulation_time_input = _normalized_text(request.form.get("sim_time"))
db.session.delete(entry)
db.session.commit()
flash("日志条目已删除。", "success")
redirect_kwargs: dict[str, str | object] = {"asset_id": asset.id}
if simulation_time_input is not None:
redirect_kwargs["sim_time"] = simulation_time_input
return redirect(url_for("web.asset_detail", **redirect_kwargs))
@web_bp.get("/mission-preview")
def mission_preview_home() -> object:
return redirect(url_for("web.mission_status_board_preview"))
@web_bp.get("/mission-preview/status-board")
def mission_status_board_preview() -> str:
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
asset_type_filter = request.args.get("asset_type", "").strip()
location_filter = request.args.get("location", "").strip()
record_scope = _normalize_record_scope(request.args.get("record_scope"))
all_rows = [
row
for row in _build_mission_board_rows(_load_assets_with_logs(), simulation_time)
if not row["asset"].is_retired
]
visible_rows = _filter_mission_board_rows(all_rows, asset_type_filter, location_filter, record_scope)
return render_template(
"mission_status_board_preview.html",
title="任务运营 · 状态板",
preview_title="状态板",
preview_tab="status",
status_groups=_build_status_board_groups(visible_rows),
**_mission_board_context(
"web.mission_status_board_preview",
simulation_time,
asset_type_filter,
location_filter,
record_scope,
all_rows,
visible_rows,
),
)
@web_bp.get("/mission-preview/location-board")
def mission_location_board_preview() -> str:
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
asset_type_filter = request.args.get("asset_type", "").strip()
location_filter = request.args.get("location", "").strip()
record_scope = _normalize_record_scope(request.args.get("record_scope"))
all_rows = [
row
for row in _build_mission_board_rows(_load_assets_with_logs(), simulation_time)
if not row["asset"].is_retired
]
visible_rows = _filter_mission_board_rows(all_rows, asset_type_filter, location_filter, record_scope)
return render_template(
"mission_location_board_preview.html",
title="任务运营 · 地点板",
preview_title="地点板",
preview_tab="location",
location_groups=_build_location_board_groups(visible_rows),
**_mission_board_context(
"web.mission_location_board_preview",
simulation_time,
asset_type_filter,
location_filter,
record_scope,
all_rows,
visible_rows,
),
)
@web_bp.get("/mission-preview/asset-log")
def mission_asset_log_preview() -> object:
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
return redirect(url_for("web.asset_list", sim_time=_format_datetime_input(simulation_time)))
@web_bp.get("/mission-preview/timeline")
def mission_timeline_preview() -> str:
simulation_time = _resolve_simulation_time(request.args.get("sim_time"))
asset_type_filter = request.args.get("asset_type", "").strip()
location_filter = request.args.get("location", "").strip()
record_scope = _normalize_record_scope(request.args.get("record_scope"))
requested_asset_ids = {asset_id for asset_id in request.args.getlist("asset_ids") if asset_id}
all_rows = _build_mission_board_rows(_load_assets_with_logs(), simulation_time)
filtered_rows = _filter_mission_board_rows(all_rows, asset_type_filter, location_filter, record_scope)
visible_rows = [
row for row in filtered_rows if not requested_asset_ids or str(row["asset_id"]) in requested_asset_ids
]
default_rows = visible_rows or filtered_rows or all_rows
initial_range_start, initial_range_end = _timeline_default_range(default_rows, simulation_time)
timeline_scale_mode = _resolve_timeline_scale(request.args.get("scale"), initial_range_start, initial_range_end)
default_range_start, default_range_end = _timeline_default_range_for_scale(default_rows, simulation_time, timeline_scale_mode)
timeline_range_mode = "manual" if _normalized_text(request.args.get("range_mode")) == "manual" else "auto"
if timeline_range_mode == "manual":
range_start = _resolve_timeline_boundary(request.args.get("start"), default_range_start)
range_end = _resolve_timeline_boundary(request.args.get("end"), default_range_end)
else:
range_start = default_range_start
range_end = default_range_end
if range_end < range_start:
range_end = range_start
timeline_scale = _timeline_scale_labels(range_start, range_end, timeline_scale_mode)
timeline_rows, timeline_counts = _build_timeline_rows(
visible_rows,
simulation_time,
range_start,
range_end,
timeline_scale_mode,
)
timeline_column_width = _timeline_column_width(timeline_scale_mode)
board_query_args = _mission_board_query_args(simulation_time, asset_type_filter, location_filter, record_scope)
timeline_query_args = {**board_query_args, "scale": timeline_scale_mode}
if requested_asset_ids:
timeline_query_args["asset_ids"] = sorted(requested_asset_ids)
if timeline_range_mode == "manual":
timeline_query_args["start"] = _format_datetime_input(range_start)
timeline_query_args["end"] = _format_datetime_input(range_end)
timeline_query_args["range_mode"] = timeline_range_mode
selected_asset_options = [
{
"value": str(row["asset_id"]),
"label": row["name"],
"asset_type": row["asset_type"],
"selected": str(row["asset_id"]) in requested_asset_ids,
}
for row in sorted(filtered_rows, key=lambda item: item["name"].casefold())
]
return render_template(
"mission_timeline_preview.html",
title="任务运营 · 时间线",
preview_title="时间线",
preview_eyebrow="Mission Ops",
preview_tab="timeline",
timeline_scale=timeline_scale,
timeline_rows=timeline_rows,
timeline_scale_mode=timeline_scale_mode,
timeline_scale_options=TIMELINE_SCALE_OPTIONS,
timeline_column_width=timeline_column_width,
timeline_range_start=_format_datetime_input(range_start),
timeline_range_end=_format_datetime_input(range_end),
timeline_range_mode=timeline_range_mode,
timeline_range_display=_timeline_range_display(range_start, range_end, timeline_scale_mode),
simulation_time=_format_datetime_display(simulation_time),
simulation_time_input=_format_datetime_input(simulation_time),
metrics=[
{"label": "Visible Assets", "value": str(len(timeline_rows))},
{"label": "State Segments", "value": str(timeline_counts["states"] )},
{"label": "Event Nodes", "value": str(timeline_counts["events"] )},
],
header_time_action=url_for("web.mission_timeline_preview"),
header_time_field_id="timeline-header-sim-time",
header_time_submit_label="Apply Time",
header_hidden_fields=_build_header_hidden_fields(
{
"asset_type": asset_type_filter,
"location": location_filter,
"record_scope": record_scope,
"scale": timeline_scale_mode,
"asset_ids": sorted(requested_asset_ids),
"range_mode": timeline_range_mode if timeline_range_mode == "manual" else None,
"start": _format_datetime_input(range_start) if timeline_range_mode == "manual" else None,
"end": _format_datetime_input(range_end) if timeline_range_mode == "manual" else None,
}
),
asset_type_filter=asset_type_filter,
location_filter=location_filter,
record_scope=record_scope,
selected_asset_ids=sorted(requested_asset_ids),
selected_asset_count=len(requested_asset_ids),
timeline_asset_options=selected_asset_options,
reset_url=url_for(
"web.mission_timeline_preview",
sim_time=_format_datetime_input(simulation_time),
asset_type=asset_type_filter or None,
location=location_filter or None,
record_scope=record_scope,
scale=timeline_scale_mode,
asset_ids=sorted(requested_asset_ids),
),
asset_type_options=sorted({row["asset_type"] for row in all_rows if row["asset_type"]}, key=str.casefold),
location_options=sorted({row["location"] for row in all_rows if row["location"]}, key=str.casefold),
record_scope_options=MISSION_BOARD_RECORD_SCOPE_OPTIONS,
board_nav_urls={
"status": url_for("web.mission_status_board_preview", **board_query_args),
"location": url_for("web.mission_location_board_preview", **board_query_args),
"mission": url_for("web.mission_list", sim_time=_format_datetime_input(simulation_time)),
"asset": url_for("web.asset_list", sim_time=_format_datetime_input(simulation_time)),
"timeline": url_for("web.mission_timeline_preview", **timeline_query_args),
},
)