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 import uuid from flask import Blueprint, abort, flash, has_request_context, redirect, render_template, request, session, url_for from sqlalchemy import select, text, distinct from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import selectinload from app.extensions import db from app.models import Asset, AssetLogEntry, AssetStateNode, CommunicationPart, DockingEvent, 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) HARDCODED_SIMULATION_TIME = datetime(2060, 3, 12, 9, 0, tzinfo=timezone.utc) def _db_read_simulation_time() -> datetime | None: try: row = db.session.execute( text("SELECT value FROM simulation_settings WHERE key = 'current_time'") ).fetchone() if row is None: return None return _parse_datetime(row[0], "DB Simulation Time") except Exception: return None def _db_write_simulation_time(value: datetime) -> None: formatted = _format_datetime_input(value) try: db.session.execute( text( "INSERT INTO simulation_settings (key, value, updated_at) " "VALUES ('current_time', :val, NOW()) " "ON CONFLICT (key) DO UPDATE SET value = :val2, updated_at = NOW()" ), {"val": formatted, "val2": formatted}, ) db.session.commit() except Exception: db.session.rollback() def _default_simulation_time() -> datetime: db_time = _db_read_simulation_time() if db_time is not None: return db_time return HARDCODED_SIMULATION_TIME def _all_state_labels() -> list[str]: """Return merged state labels: defaults + all unique labels from DB.""" defaults = ['start-transit','transit-complete','Exploration','Maintenance','Docked', 'Construction','Testing','Autonomous Standby','Crew Handoff','Cargo Delivery', 'Surface Expedition','Planetary Operations','Base Construction','Outpost Construction', 'Base Expansion','Outpost Expansion'] db_labels = ( db.session.execute( select(distinct(AssetStateNode.state_label)).where(AssetStateNode.state_label.isnot(None)) ).scalars().all() ) merged = {*defaults, *db_labels} return sorted(merged, key=str.casefold) 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 = None) -> datetime: db_time = _db_read_simulation_time() if db_time is not None: _remember_simulation_time(db_time) return db_time 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 = HARDCODED_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) else: state_node_rows = _event_location_state_node_rows(target, form.get("location"), title, start_at) target.entry_kind = entry_kind target.title = title target.mission_label = _normalize_multi_value_text(form.get("mission_label")) target.start_at = start_at target.end_at = end_at target.summary = _normalized_text(form.get("summary")) target.note = _normalized_text(form.get("note")) _sync_state_nodes(target, state_node_rows) def _apply_asset_entry_json_fields(target: AssetLogEntry, data: object) -> None: if not isinstance(data, dict): message = "Payload invalid" raise FormValidationError(message, field_errors={"payload": message}) entry_kind = _normalized_text(data.get("entry_kind")) or target.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 invalid" raise FormValidationError(message, field_errors={"entry_kind": message}) title = _normalized_text(data.get("title")) or target.title if title is None: message = "Title required" raise FormValidationError(message, field_errors={"title": message}) try: start_at = _parse_datetime(data.get("start_at") or data.get("sim_time"), "Start At") except ValueError as exc: raise FormValidationError(str(exc), field_errors={"start_at": str(exc)}) from exc if start_at is None: start_at = target.start_at if start_at is None: message = "Start At required" raise FormValidationError(message, field_errors={"start_at": message}) if "end_at" in data: try: end_at = _parse_datetime(data.get("end_at"), "End At") except ValueError as exc: raise FormValidationError(str(exc), field_errors={"end_at": str(exc)}) from exc else: end_at = target.end_at if entry_kind == "event": end_at = None elif end_at is not None and end_at <= start_at: message = "End At must be later than Start At" raise FormValidationError(message, field_errors={"start_at": message, "end_at": message}) state_node_rows = [] if entry_kind == "state": state_node_rows = _parse_json_state_node_rows(data, start_at, end_at) else: state_node_rows = _event_location_state_node_rows(target, data.get("location"), title, start_at) target.entry_kind = entry_kind target.title = title target.mission_label = _normalize_multi_value_text(data.get("mission_label")) target.start_at = start_at target.end_at = end_at target.summary = _normalized_text(data.get("summary")) target.note = _normalized_text(data.get("note")) _sync_state_nodes(target, state_node_rows) def _asset_entry_form_values(entry: AssetLogEntry) -> dict[str, object]: return { "entry_kind": entry.entry_kind or "event", "title": entry.title or "", "mission_label": entry.mission_label or "", "location": _entry_location(entry) 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), "target_location": node.target_location or "", "previous_location": node.previous_location or "", "transit_location": node.transit_location or "", "state_label": node.state_label or "", } 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") node_locations = form.getlist("state_node_target_location") node_prev_locations = form.getlist("state_node_previous_location") node_transit_locations = form.getlist("state_node_transit_location") node_state_labels = form.getlist("state_node_state_label") 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 "", "target_location": node_locations[index] if index < len(node_locations) else "", "previous_location": node_prev_locations[index] if index < len(node_prev_locations) else "", "transit_location": node_transit_locations[index] if index < len(node_transit_locations) else "", "state_label": node_state_labels[index] if index < len(node_state_labels) else "", } ) node_locations_2 = form.getlist("location") return { "entry_kind": form.get("entry_kind") or "event", "title": form.get("title") or "", "mission_label": form.get("mission_label") or "", "location": node_locations_2[0] if node_locations_2 else "", "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") node_locations = form.getlist("state_node_target_location") node_prev_locations = form.getlist("state_node_previous_location") node_transit_locations = form.getlist("state_node_transit_location") node_state_labels = form.getlist("state_node_state_label") 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) ): node_id = _normalized_text(raw_id) title = _normalized_text(raw_title) detail = _normalized_text(raw_detail) display_index = index + 1 try: at_time = _parse_datetime(raw_at_time, f"State Node {display_index} At") except ValueError as exc: raise FormValidationError( str(exc), field_errors={f"state_node_at:{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 {display_index} 需要同时填写 Title 和 At" field_errors: dict[str, str] = {} if title is None: field_errors[f"state_node_title:{index}"] = message if at_time is None: field_errors[f"state_node_at:{index}"] = message raise FormValidationError(message, field_errors=field_errors) if at_time < start_at: message = f"State Node {display_index} 时间不能早于 Start At" raise FormValidationError( message, field_errors={"start_at": message, f"state_node_at:{index}": message}, ) if end_at is not None and at_time > end_at: message = f"State Node {display_index} 时间不能晚于 End At" raise FormValidationError( message, field_errors={"end_at": message, f"state_node_at:{index}": message}, ) location = _normalized_text(node_locations[index]) if index < len(node_locations) else None prev_loc = _normalized_text(node_prev_locations[index]) if index < len(node_prev_locations) else None transit_loc = _normalized_text(node_transit_locations[index]) if index < len(node_transit_locations) else None state_label = _normalize_multi_value_text(node_state_labels[index]) if index < len(node_state_labels) else None state_nodes.append( { "id": node_id, "title": title, "detail": detail, "at_time": at_time, "target_location": location, "previous_location": prev_loc, "transit_location": transit_loc, "state_label": state_label, } ) return state_nodes def _parse_json_state_node_rows( data: dict[str, object], start_at: datetime, end_at: datetime | None, ) -> list[dict[str, object]]: raw_nodes = data.get("state_nodes") or [] if not isinstance(raw_nodes, list): message = "State Nodes invalid" raise FormValidationError(message, field_errors={"state_nodes": message}) state_nodes = [] for index, raw_node in enumerate(raw_nodes): if not isinstance(raw_node, dict): message = f"State Node {index + 1} invalid" raise FormValidationError(message, field_errors={"state_nodes": message}) title = _normalized_text(raw_node.get("title")) detail = _normalized_text(raw_node.get("detail")) try: at_time = _parse_datetime(raw_node.get("at_time") or raw_node.get("at"), f"State Node {index + 1} At") except ValueError as exc: raise FormValidationError( str(exc), field_errors={f"state_node_at:{index}": str(exc)}, ) from exc if at_time is None: at_time = start_at if title is None and detail is None: continue if title is None: message = f"State Node {index + 1} Title required" raise FormValidationError(message, field_errors={f"state_node_title:{index}": message}) if at_time < start_at: message = f"State Node {index + 1} cannot be earlier than Start At" raise FormValidationError( message, field_errors={"start_at": message, f"state_node_at:{index}": message}, ) if end_at is not None and at_time > end_at: message = f"State Node {index + 1} cannot be later than End At" raise FormValidationError( message, field_errors={"end_at": message, f"state_node_at:{index}": message}, ) target_location = _normalized_text( raw_node.get("target_location") or raw_node.get("location") or data.get("location") ) state_label = _normalize_multi_value_text(raw_node.get("state_label") or data.get("state_label")) state_nodes.append( { "id": _normalized_text(raw_node.get("id")), "title": title, "detail": detail, "at_time": at_time, "target_location": target_location, "previous_location": _normalized_text(raw_node.get("previous_location")), "transit_location": _normalized_text(raw_node.get("transit_location")), "state_label": state_label, } ) return state_nodes def _event_location_state_node_rows( target: AssetLogEntry, raw_location: object, title: str, start_at: datetime, ) -> list[dict[str, object]]: location = _normalized_text(raw_location) if location is None: return [] existing_node_id = "" if target.state_nodes: existing_node_id = str(sorted(target.state_nodes, key=lambda item: _as_utc(item.at_time) or start_at)[0].id) return [ { "id": existing_node_id, "title": title, "detail": None, "at_time": start_at, "target_location": location, "previous_location": None, "transit_location": None, "state_label": None, } ] def _append_docking_log( asset: Asset, child_name: str, parent_name: str, target_location: str, docked_at: datetime, note: str | None, ) -> None: title = f"{child_name} docked to {parent_name}" # Check if an existing (non-dock) state interval covers the dock time existing = AssetLogEntry.query.filter( AssetLogEntry.asset_id == asset.id, AssetLogEntry.entry_kind == "state", AssetLogEntry.start_at <= docked_at, ).filter( (AssetLogEntry.end_at.is_(None)) | (AssetLogEntry.end_at > docked_at) ).filter( ~AssetLogEntry.title.ilike("%docked to%") ).order_by(AssetLogEntry.start_at.asc()).first() if existing is not None: existing.state_nodes.append( AssetStateNode( title=title, detail=note, at_time=docked_at, target_location=target_location, state_label="Docked", ) ) else: entry = AssetLogEntry( entry_kind="state", title=title, start_at=docked_at, summary=note, asset=asset, ) entry.state_nodes.append( AssetStateNode( title=title, detail=note, at_time=docked_at, target_location=target_location, state_label="Docked", ) ) db.session.add(entry) def _append_undocking_log( asset: Asset, child_name: str, parent_name: str, docked_at: datetime, undocked_at: datetime, note: str | None, ) -> None: """Record an undocking event in the asset's log timeline.""" dock_title = f"{child_name} docked to {parent_name}" # Find the entry that contains the dock node (may not be titled 'docked to') entry = AssetLogEntry.query.filter( AssetLogEntry.asset_id == asset.id, AssetLogEntry.entry_kind == "state", AssetLogEntry.state_nodes.any( (AssetStateNode.title == dock_title) & (AssetStateNode.at_time == docked_at) ) ).first() if entry is not None: entry.state_nodes.append(AssetStateNode( title=f"{child_name} undocked from {parent_name}", detail=note, at_time=undocked_at, state_label="Docked", )) if entry.end_at is None: entry.end_at = undocked_at else: undock_title = f"{child_name} undocked from {parent_name}" entry = AssetLogEntry( entry_kind="event", title=undock_title, start_at=undocked_at, summary=note, asset=asset, ) db.session.add(entry) def _remove_undocking_log( asset: Asset, child_name: str, parent_name: str, ) -> None: """Remove undock state nodes and reopen the dock entry.""" dock_title = f"{child_name} docked to {parent_name}" undock_title = f"{child_name} undocked from {parent_name}" entry = AssetLogEntry.query.filter_by( asset_id=asset.id, entry_kind="state", title=dock_title ).order_by(AssetLogEntry.start_at.desc()).first() if entry is not None: nodes_to_remove = [n for n in entry.state_nodes if n.title == undock_title] for node in nodes_to_remove: db.session.delete(node) entry.end_at = None def _parse_uuid_value(raw_value: object) -> uuid.UUID | None: if isinstance(raw_value, uuid.UUID): return raw_value cleaned = _normalized_text(str(raw_value)) if raw_value is not None else None if cleaned is None: return None try: return uuid.UUID(cleaned) except ValueError: return None 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"] node.target_location = row.get("target_location") node.previous_location = row.get("previous_location") node.transit_location = row.get("transit_location") node.state_label = row.get("state_label") 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(_entry_state_label(active_state, simulation_time)), "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(entry, simulation_time) 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(entry, simulation_time)), "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 _entry_location(entry: AssetLogEntry, simulation_time: datetime | None = None) -> str | None: """Derive location from the most recent state node at or before simulation_time.""" sim = simulation_time or _default_simulation_time() if entry.state_nodes: sorted_nodes = sorted(entry.state_nodes, key=lambda n: _as_utc(n.at_time) or _default_simulation_time()) for node in reversed(sorted_nodes): node_at = _as_utc(node.at_time) if node_at and node_at <= sim and node.target_location: return node.target_location return None def _entry_state_label(entry: AssetLogEntry, simulation_time: datetime | None = None) -> str | None: """Derive state_label from the most recent state node at or before simulation_time.""" sim = simulation_time or _default_simulation_time() if entry.state_nodes: sorted_nodes = sorted(entry.state_nodes, key=lambda n: _as_utc(n.at_time) or _default_simulation_time()) for node in reversed(sorted_nodes): node_at = _as_utc(node.at_time) if node_at and node_at <= sim and node.state_label: return node.state_label return None def _is_docking_event_active(event: DockingEvent, simulation_time: datetime) -> bool: docked_at = _as_utc(event.docked_at) undocked_at = _as_utc(event.undocked_at) return bool(docked_at and docked_at <= simulation_time and (undocked_at is None or simulation_time < undocked_at)) def _asset_summary_link(asset: Asset, simulation_time: datetime) -> dict[str, object]: snapshot = _build_asset_snapshot(asset, simulation_time, include_docking=False) return { "id": asset.id, "name": asset.name, "current_location": snapshot["current_location"], } def _build_docking_summary(asset: Asset, simulation_time: datetime) -> dict[str, object]: target_events = db.session.execute( select(DockingEvent) .options(selectinload(DockingEvent.parent_asset)) .where(DockingEvent.child_asset_id == asset.id) .order_by(DockingEvent.docked_at.desc()) ).scalars().all() target_event = next( (event for event in target_events if _is_docking_event_active(event, simulation_time)), None, ) child_events = [ event for event in db.session.execute( select(DockingEvent) .options(selectinload(DockingEvent.child_asset)) .where(DockingEvent.parent_asset_id == asset.id) .order_by(DockingEvent.docked_at.desc()) ).scalars().all() if _is_docking_event_active(event, simulation_time) ] docking_target = None if target_event is not None and target_event.parent_asset is not None: docking_target = { **_asset_summary_link(target_event.parent_asset, simulation_time), "docked_at": _format_datetime_display(target_event.docked_at), "event_id": str(target_event.id), "undocked_at": _format_datetime_input(target_event.undocked_at) if target_event.undocked_at else None, } docked_vehicles = [] for event in child_events: if event.child_asset is not None: item = _asset_summary_link(event.child_asset, simulation_time) else: item = {"id": None, "name": event.child_label or "External vehicle", "current_location": "-"} item["docked_at"] = _format_datetime_display(event.docked_at) item["event_id"] = str(event.id) item["undocked_at"] = _format_datetime_input(event.undocked_at) if event.undocked_at else None docked_vehicles.append(item) return { "docking_target": docking_target, "docked_vehicles": docked_vehicles, } def _build_asset_snapshot(asset: Asset, simulation_time: datetime, *, include_docking: bool = True) -> dict[str, object]: simulation_time = _as_utc(simulation_time) or _default_simulation_time() 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(_entry_state_label(active_state, simulation_time)) or active_state.title current_location = _entry_location(active_state, simulation_time) 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 = ( (_entry_location(last_entry, simulation_time) 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']}" snapshot = { "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], } if include_docking: snapshot.update(_build_docking_summary(asset, simulation_time)) else: snapshot.update({"docking_target": None, "docked_vehicles": []}) return snapshot 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(entry) 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 ), "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" ), "asset": url_for("web.asset_list" ), "timeline": url_for("web.mission_timeline_preview" ), }, } 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" ), "location": url_for("web.mission_location_board_preview" ), "mission": url_for("web.mission_list" ), "asset": url_for("web.asset_list" ), "timeline": url_for("web.mission_timeline_preview" ), } 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(entry, simulation_time) 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, ) 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 # Map location to celestial body color loc = (_entry_location(entry, simulation_time) or "").lower() if "jupiter" in loc or "europa" in loc or "ganymede" in loc or "callisto" in loc: body_color = "#c9b07c" # Jupiter gold elif "saturn" in loc or "titan" in loc: body_color = "#e0c97f" # Saturn beige elif "mars" in loc: body_color = "#c1440e" # Mars red elif "earth" in loc or "leo" in loc or "geo" in loc: body_color = "#4da6ff" # Earth blue elif "neptune" in loc or "triton" in loc: body_color = "#3f54ba" # Neptune blue elif "sun" in loc: body_color = "#FDB813" # Sun yellow elif "transfer" in loc: body_color = "#94A3B8" # Gray else: body_color = "#94A3B8" # Default gray 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, "start_at": _format_datetime_display(entry.start_at), "end_at": _format_datetime_display(entry.end_at) if entry.end_at else "ongoing", "state_node_count": len([n for n in entry.state_nodes if _as_utc(n.at_time) and range_start <= _as_utc(n.at_time) <= range_end]), "body_color": body_color, "start_iso": _format_datetime_input(entry.start_at), "end_iso": _format_datetime_input(entry.end_at) if entry.end_at else "", } ) 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, "at_iso": _format_datetime_input(state_node.at_time), "is_state_node": True, "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, "at_iso": _format_datetime_input(entry.start_at), "is_state_node": False, "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" 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: simulation_time = _resolve_simulation_time(request.args.get("sim_time")) simulation_time_input = _format_datetime_input(simulation_time) # Load all assets with logs assets = db.session.execute( select(Asset).options(selectinload(Asset.log_entries)).order_by(Asset.name) ).scalars().all() # Build snapshots asset_rows = [] for asset in assets: snapshot = _build_asset_snapshot(asset, simulation_time) asset_rows.append({ "id": asset.id, "name": asset.name, "asset_type": asset.asset_type, "is_retired": asset.is_retired, "location": snapshot.get("current_location", "—"), "mission": snapshot.get("current_mission", "—"), "state": snapshot.get("current_state", "—"), }) # Metrics active = [a for a in asset_rows if not a["is_retired"]] motherships = len([a for a in active if a["asset_type"] == "Exploration Mothership"]) stations = len([a for a in active if a["asset_type"] == "Station"]) total_assets = len(active) # Status groups state_groups = { "Transfer": {"label": "In Transit", "css_class": "tf", "assets": []}, "Exploration": {"label": "Exploration", "css_class": "ex", "assets": []}, "Maintenance": {"label": "Maintenance", "css_class": "mt", "assets": []}, "Docked": {"label": "Docked / Surface", "css_class": "dk", "assets": []}, } for a in active: state = a["state"] if state in state_groups: state_groups[state]["assets"].append(a) else: state_groups["Docked"]["assets"].append(a) status_groups = [g for g in state_groups.values() if g["assets"]] # Upcoming events (30 days) now = simulation_time upcoming_events = [] for a in active: snapshot = _build_asset_snapshot(a if hasattr(a, "log_entries") else assets[0], simulation_time) for ev in snapshot.get("future_events", []): ev_date = ev.get("date", "") if ev_date and ev_date >= now.strftime("%Y-%m-%d"): upcoming_events.append({ "date": ev_date, "title": f"{a['name']}: {ev.get('title', '')}", "within_30d": True, }) upcoming_events.sort(key=lambda e: e["date"]) upcoming_events = upcoming_events[:5] active_missions = len([a for a in active if a["state"] in ("Transfer", "Exploration")]) upcoming_count = len(upcoming_events) upcoming_summary = f"{upcoming_count} events" if upcoming_count else "None imminent" return render_template( "dashboard.html", title="任务总览", active_page="overview", simulation_time_display=_format_datetime_display(simulation_time), simulation_time_input=simulation_time_input, total_assets=total_assets, mothership_count=motherships, station_count=stations, active_missions=active_missions, upcoming_count=upcoming_count, upcoming_summary=upcoming_summary, status_groups=status_groups, upcoming_events=upcoming_events, ) @web_bp.get("/fuel-converter") def fuel_converter() -> str: return render_template( "fuel_converter.html", title="燃料换算", active_page="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="引擎目录", active_page="engines", 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/") 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//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//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//variants//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//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="通信部件", active_page="comms", 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//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//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="燃料箱规格", active_page="tanks", 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//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//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", active_page="vehicles", 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//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//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="任务资产", active_page="assets", 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(entry, simulation_time)) 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(entry, simulation_time) 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 )) 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" ), 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//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 )) 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 ), 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//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/") 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"], ), } 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, "source_entry_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(entry, simulation_time) or "-", "summary": entry.summary or "未填写摘要。", "mission_label": _display_multi_value_text(entry.mission_label), "state_label": _display_multi_value_text(_entry_state_label(entry, simulation_time)), "mission_labels": _split_multi_value_text(entry.mission_label), "state_labels": _split_multi_value_text(_entry_state_label(entry, simulation_time)), "edit_url": url_for("web.asset_entry_edit", asset_id=asset.id, entry_id=entry.id ), "state_nodes": [ {"title": n.title, "at": _format_datetime_display(n.at_time), "detail": n.detail or ""} for n in sorted(entry.state_nodes, key=lambda x: _as_utc(x.at_time) or _default_simulation_time()) ] if entry.entry_kind == "state" and entry.state_nodes else [], } ) 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, active_page="assets", preview_title=asset.name, preview_eyebrow="Mission Asset", asset=asset, asset_summary=snapshot, all_assets=db.session.execute(select(Asset).where(Asset.id != asset.id).order_by(Asset.name)).scalars().all(), 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//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": # Support both form and JSON if request.content_type and "application/json" in request.content_type: data = request.get_json(force=True, silent=True) or {} try: _apply_asset_entry_json_fields(entry, data) entry.asset = asset db.session.add(entry) db.session.commit() return {"success": True, "entry_id": str(entry.id)} entry.entry_kind = data.get("entry_kind", "event") entry.title = data.get("title", "") entry.summary = data.get("summary") if data.get("start_at"): entry.start_at = _parse_datetime(data["start_at"], "开始时间") if data.get("end_at"): entry.end_at = _parse_datetime(data["end_at"], "结束时间") entry.asset = asset db.session.add(entry) # Handle state nodes for node_data in data.get("state_nodes", []): if node_data.get("title"): node = AssetStateNode( title=node_data["title"], detail=node_data.get("detail"), at_time=_parse_datetime(node_data.get("at_time"), "节点时间") or entry.start_at, entry=entry, ) db.session.add(node) db.session.commit() return {"success": True, "entry_id": str(entry.id)} except Exception as e: db.session.rollback() return {"error": str(e)}, 400 else: 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 )) 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, all_states=_all_state_labels(), cancel_url=url_for("web.asset_detail", asset_id=asset.id ), 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//entries//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": if request.content_type and "application/json" in request.content_type: data = request.get_json(force=True, silent=True) or {} try: _apply_asset_entry_json_fields(entry, data) db.session.commit() return {"success": True} entry.entry_kind = data.get("entry_kind", entry.entry_kind) entry.title = data.get("title", entry.title) entry.summary = data.get("summary") if data.get("start_at"): entry.start_at = _parse_datetime(data["start_at"], "开始时间") if "end_at" in data: entry.end_at = _parse_datetime(data["end_at"], "结束时间") if data["end_at"] else None # Handle state nodes for node in entry.state_nodes: db.session.delete(node) for node_data in data.get("state_nodes", []): if node_data.get("title"): node = AssetStateNode( title=node_data["title"], detail=node_data.get("detail"), at_time=_parse_datetime(node_data.get("at_time"), "节点时间") or entry.start_at, entry=entry, ) db.session.add(node) db.session.commit() return {"success": True} except Exception as e: db.session.rollback() return {"error": str(e)}, 400 else: 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 )) 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, all_states=_all_state_labels(), cancel_url=url_for("web.asset_detail", asset_id=asset.id ), 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("/api/v1/sim-time") def api_set_sim_time() -> object: data = request.get_json(force=True, silent=True) or {} time_str = data.get("sim_time", "").strip() if time_str: parsed = _parse_datetime(time_str, "sim_time") if parsed is not None: _db_write_simulation_time(parsed) return {"success": True} return {"error": "Invalid time"}, 400 @web_bp.post("/api/v1/assets//dock") def api_asset_dock(asset_id: object) -> object: asset = _load_asset_or_404(asset_id) data = request.get_json(force=True, silent=True) or {} mode = _normalized_text(data.get("mode")) or "dock_vehicle" vehicle_id = data.get("vehicle_id") target_asset_id = data.get("target_asset_id") or vehicle_id vehicle_name = _normalized_text(data.get("vehicle_name")) note = _normalized_text(data.get("note")) sim_time = _resolve_simulation_time(data.get("sim_time")) dock_time = _parse_datetime(data.get("dock_time"), "对接时间") or sim_time try: if mode == "dock_to_target": if not target_asset_id: raise FormValidationError("Docking target required") parent_asset = db.session.get(Asset, _parse_uuid_value(target_asset_id)) if parent_asset is None or parent_asset.id == asset.id: raise FormValidationError("Docking target invalid") child_asset = asset child_label = None elif mode == "dock_vehicle": parent_asset = asset if vehicle_id and vehicle_id != "_custom": child_asset = db.session.get(Asset, _parse_uuid_value(vehicle_id)) if child_asset is None or child_asset.id == asset.id: raise FormValidationError("Docked vehicle invalid") child_label = None else: child_asset = None child_label = vehicle_name if child_label is None: raise FormValidationError("Vehicle name required") else: raise FormValidationError("Docking mode invalid") child_name = child_asset.name if child_asset is not None else str(child_label) parent_location = str(_build_asset_snapshot(parent_asset, dock_time, include_docking=False)["current_location"]) event = DockingEvent( parent_asset=parent_asset, child_asset=child_asset, child_label=child_label, docked_at=dock_time, dock_note=note, ) db.session.add(event) _append_docking_log(parent_asset, child_name, parent_asset.name, parent_location, dock_time, note) if child_asset is not None: _append_docking_log(child_asset, child_name, parent_asset.name, parent_location, dock_time, note) db.session.commit() except FormValidationError as exc: db.session.rollback() return {"error": str(exc)}, 400 if request.content_type and "application/json" in request.content_type: return {"success": True} flash("对接已记录。", "success") return redirect(url_for("web.asset_detail", asset_id=asset.id )) @web_bp.post("/api/v1/assets//undock") def api_asset_undock(asset_id: object) -> object: asset = _load_asset_or_404(asset_id) data = request.get_json(force=True, silent=True) or {} event_id = data.get("event_id") action = _normalized_text(data.get("action")) or "undock" note = _normalized_text(data.get("note")) sim_time = _resolve_simulation_time(data.get("sim_time")) undock_time = _parse_datetime(data.get("undock_time"), "解对接时间") or sim_time try: if not event_id: raise FormValidationError("Docking event required") docking_event = db.session.get(DockingEvent, _parse_uuid_value(event_id)) if docking_event is None: raise FormValidationError("Docking event not found") if docking_event.parent_asset_id != asset.id and docking_event.child_asset_id != asset.id: raise FormValidationError("Docking event does not belong to this asset") parent_asset = db.session.get(Asset, docking_event.parent_asset_id) child_asset = db.session.get(Asset, docking_event.child_asset_id) if docking_event.child_asset_id else None child_name = child_asset.name if child_asset is not None else (docking_event.child_label or "External vehicle") parent_name = parent_asset.name if parent_asset is not None else "Unknown" if action == "delete_future": if docking_event.undocked_at is None: raise FormValidationError("No future undock to delete") docking_event.undocked_at = None docking_event.undock_note = None _remove_undocking_log(parent_asset, child_name, parent_name) if child_asset is not None: _remove_undocking_log(child_asset, child_name, parent_name) elif action == "edit_future": if docking_event.undocked_at is None: raise FormValidationError("No future undock to edit") _remove_undocking_log(parent_asset, child_name, parent_name) if child_asset is not None: _remove_undocking_log(child_asset, child_name, parent_name) docking_event.undocked_at = undock_time docking_event.undock_note = note _append_undocking_log(parent_asset, child_name, parent_name, docking_event.docked_at, undock_time, note) if child_asset is not None: _append_undocking_log(child_asset, child_name, parent_name, docking_event.docked_at, undock_time, note) else: if docking_event.undocked_at is not None: raise FormValidationError("Already undocked") docking_event.undocked_at = undock_time docking_event.undock_note = note if parent_asset is not None: _append_undocking_log(parent_asset, child_name, parent_name, docking_event.docked_at, undock_time, note) if child_asset is not None: _append_undocking_log(child_asset, child_name, parent_name, docking_event.docked_at, undock_time, note) db.session.commit() except FormValidationError as exc: db.session.rollback() return {"error": str(exc)}, 400 if request.content_type and "application/json" in request.content_type: return {"success": True} flash("对接已解除。", "success") return redirect(url_for("web.asset_detail", asset_id=asset.id )) @web_bp.get("/docking-logs") def docking_log_list() -> str: simulation_time = _resolve_simulation_time() simulation_time_input = _format_datetime_input(simulation_time) events = ( db.session.execute( select(DockingEvent) .options(selectinload(DockingEvent.parent_asset), selectinload(DockingEvent.child_asset)) .order_by(DockingEvent.docked_at.desc()) ).scalars().all() ) return render_template( "docking_logs.html", title="对接日志管理", active_page="docking_logs", events=events, simulation_time_display=_format_datetime_display(simulation_time), simulation_time_input=simulation_time_input, ) @web_bp.post("/docking-logs//delete") def docking_log_delete(event_id: object) -> object: event = db.session.get(DockingEvent, _parse_uuid_value(event_id)) if event is None: flash("对接记录未找到。", "error") return redirect(url_for("web.docking_log_list")) parent = db.session.get(Asset, event.parent_asset_id) if event.parent_asset_id else None child = db.session.get(Asset, event.child_asset_id) if event.child_asset_id else None child_name = child.name if child else (event.child_label or "External vehicle") parent_name = parent.name if parent else "Unknown" # Remove docking log entries from both assets dock_title = f"{child_name} docked to {parent_name}" for asset in [parent, child]: if asset is None: continue entry = AssetLogEntry.query.filter( AssetLogEntry.asset_id == asset.id, AssetLogEntry.entry_kind == "state", AssetLogEntry.state_nodes.any( (AssetStateNode.title == dock_title) & (AssetStateNode.at_time == event.docked_at) ) ).first() if entry: nodes_to_remove = [n for n in entry.state_nodes if n.title == dock_title or 'undocked' in (n.title or '')] for n in nodes_to_remove: db.session.delete(n) if not entry.state_nodes: db.session.delete(entry) db.session.delete(event) db.session.commit() flash("对接记录已删除。", "success") return redirect(url_for("web.docking_log_list")) @web_bp.post("/assets//entries//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")) simulation_time_input = _format_datetime_input(simulation_time) # Load all assets with logs assets = _load_assets_with_logs() # Build asset position data for Canvas # Map location strings to approximate orbital bodies and distances LOCATION_BODY_MAP = { "LEO": "earth", "GEO": "earth", "MEO": "earth", "Earth System": "earth", "Earth Surface": "earth", "Lunar Orbit": "moon", "Lunar Surface": "moon", "Mars Orbit": "mars", "Mars Surface": "mars", "Phobos Orbit": "mars", "Deimos Orbit": "mars", "Jupiter Orbit": "jupiter", "Europa Orbit": "jupiter", "Europa Surface": "jupiter", "Ganymede Orbit": "jupiter", "Callisto Orbit": "jupiter", "Saturn Orbit": "saturn", "Titan Orbit": "saturn", "Titan Surface": "saturn", "Uranus Orbit": "uranus", "Neptune Orbit": "neptune", "Triton Orbit": "neptune", "Mercury Orbit": "mercury", "Mercury Surface": "mercury", "Venus Orbit": "venus", "Venus Surface": "venus", "Solar Orbit": "sun", "Transfer": "transfer", "Ceres Orbit": "ceres", "Ceres Surface": "ceres", "Vesta Orbit": "vesta", "Vesta Surface": "vesta", "Pluto Orbit": "pluto", "Pluto Surface": "pluto", } asset_positions = [] for asset in assets: if asset.is_retired: continue snapshot = _build_asset_snapshot(asset, simulation_time) location = snapshot.get("current_location", "Transfer") body = LOCATION_BODY_MAP.get(location, "transfer") asset_positions.append({ "id": str(asset.id), "name": asset.name, "asset_type": asset.asset_type, "location": location, "body": body, "mission": snapshot.get("current_mission", ""), "state": snapshot.get("current_state", ""), }) # Count ongoing missions and upcoming events active_assets = [a for a in asset_positions if a["state"] in ("Transfer", "Exploration")] upcoming = [] return render_template( "mission_location_board_preview.html", title="地点板", active_page="location", simulation_time_display=_format_datetime_display(simulation_time), simulation_time_input=simulation_time_input, asset_positions_json=__import__("json").dumps(asset_positions), asset_count=len(asset_positions), ongoing_count=len(active_assets), upcoming_events=[], ) @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" )) @web_bp.get("/mission-preview/timeline") def mission_timeline_preview() -> str: simulation_time = _resolve_simulation_time(request.args.get("sim_time")) search_term = request.args.get("q", "").strip() asset_type_filter = request.args.get("asset_type", "").strip() location_filter = request.args.get("location", "").strip() state_filter = request.args.get("state", "").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) # Apply search filter if search_term: all_rows = [r for r in all_rows if search_term.lower() in (r.get("name", "") or "").lower()] 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_range_mode = "manual" if _normalized_text(request.args.get("range_mode")) == "manual" else "auto" raw_scale = request.args.get("scale") if timeline_range_mode == "manual": range_start = _resolve_timeline_boundary(request.args.get("start"), initial_range_start) range_end = _resolve_timeline_boundary(request.args.get("end"), initial_range_end) else: range_start = initial_range_start range_end = initial_range_end if range_end < range_start: range_end = range_start # Determine scale from the ACTUAL range, not the initial default timeline_scale_mode = _resolve_timeline_scale(raw_scale, range_start, 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, ) for r in timeline_rows: r["asset_url"] = url_for("web.asset_detail", asset_id=r["asset_id"] ) 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()) ] # Build JSON-LD like data for client-side rendering import json as _json timeline_json_data = { "range_start": _format_datetime_input(range_start), "range_end": _format_datetime_input(range_end), "bucket_origin": _format_datetime_input(_timeline_bucket_start(range_start, timeline_scale_mode)), "scale": timeline_scale_mode, "bucket_count": len(timeline_scale), "rows": [ { "asset_id": str(r["asset_id"]), "asset_name": str(r["name"]), "asset_type": str(r["asset_type"]), "asset_url": url_for("web.asset_detail", asset_id=r["asset_id"] ), "segments": [ { "entry_id": str(s["entry_id"]), "label": str(s["label"]), "tone": str(s["tone"]), "url": str(s["url"]), "start_position": float(s["start_position"]), "body_end_position": float(s["body_end_position"]), "display_end_position": float(s["display_end_position"]), "track": int(s.get("track", 0)), "start_at": str(s["start_at"]), "end_at": str(s["end_at"]), "body_color": str(s["body_color"]), "start_iso": str(s.get("start_iso", "")), "end_iso": str(s.get("end_iso", "")), } for s in r.get("segments", []) ], "events": [ { "label": str(e["label"]), "url": str(e["url"]), "position": float(e["position"]), "at_iso": str(e.get("at_iso", "")), "full_label": str(e.get("full_label", e["label"])), "is_state_node": bool(e.get("is_state_node", False)), "track": int(e.get("track", 0)), "parent_entry_id": str(e.get("parent_entry_id", "")), "parent_track": next( ( int(segment.get("track", 0)) for segment in r.get("segments", []) if segment.get("entry_id") == e.get("parent_entry_id") ), None, ), } for e in r.get("events", []) ], } for r in timeline_rows ], } timeline_json = _json.dumps(timeline_json_data, ensure_ascii=False) is_partial = request.args.get("_partial") == "1" template = "mission_timeline_preview.html" if is_partial: return timeline_json, 200, {"Content-Type": "application/json"} return render_template( template, title="时间线", active_page="timeline", search_term=search_term, asset_type_filter=asset_type_filter, location_filter=location_filter, state_filter=state_filter, preview_title="时间线", preview_eyebrow="Mission Ops", preview_tab="timeline", timeline_scale=timeline_scale, timeline_rows=timeline_rows, timeline_json=timeline_json, 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, } ), 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", 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" ), "asset": url_for("web.asset_list" ), "timeline": url_for("web.mission_timeline_preview", **timeline_query_args), }, )