Files
KSP_project/topology_api/queries/snapshot_queries.py
T
2026-08-11 15:02:47 +08:00

434 lines
17 KiB
Python

# topology_api/queries/snapshot_queries.py
"""Snapshot resolution and data fetching logic."""
from topology_api.db import read_cursor
def _require_presentation(condition, message):
if not condition:
raise RuntimeError(f"presentation_config_error: {message}")
def fetch_missions(base_code):
"""Return missions for a base with their latest recorded event."""
with read_cursor() as cur:
cur.execute("""
SELECT
m.code,
m.title,
m.occurrence_status,
MAX(me.effective_at) AS last_event_at
FROM missions m
JOIN bases b ON b.id = m.base_id
LEFT JOIN mission_events me ON me.mission_id = m.id
WHERE b.code = %(base_code)s
GROUP BY m.id, m.code, m.title, m.occurrence_status
ORDER BY MIN(me.effective_at) NULLS LAST, m.code
""", {"base_code": base_code})
return [
{
"code": row["code"],
"title": row["title"],
"occurrenceStatus": row["occurrence_status"],
"lastEventAt": (
row["last_event_at"].isoformat()
if row["last_event_at"] is not None
else None
),
}
for row in cur.fetchall()
]
def resolve_snapshot_time(requested_mission=None, requested_at=None):
"""Determine the snapshot timestamp per spec.
Returns: (snapshot_at: str | None, anchor_source: str)
"""
with read_cursor() as cur:
if requested_at is not None:
return requested_at, "explicit_time"
if requested_mission is not None:
cur.execute("""
SELECT id, occurrence_status
FROM missions
WHERE code = %(code)s
""", {"code": requested_mission})
mission = cur.fetchone()
if mission is None:
return None, "mission_not_found"
cur.execute("""
SELECT effective_at
FROM mission_events
WHERE mission_id = %(mid)s
ORDER BY effective_at DESC, event_order DESC
LIMIT 1
""", {"mid": mission["id"]})
event = cur.fetchone()
if event is None:
return None, "mission_has_no_timed_event"
return event["effective_at"].isoformat(), "mission_latest_event"
# Neither provided: latest event across all occurred missions
cur.execute("""
SELECT me.effective_at
FROM mission_events me
JOIN missions m ON m.id = me.mission_id
WHERE m.occurrence_status = 'occurred'
ORDER BY me.effective_at DESC, me.event_order DESC
LIMIT 1
""")
event = cur.fetchone()
if event is None:
return None, "no_events"
return event["effective_at"].isoformat(), "latest_recorded_state"
def fetch_snapshot(snapshot_at):
"""Fetch complete base topology at a given point in time.
Returns dict with keys: base, components, connections
"""
result = {"base": None, "components": [], "connections": []}
params = {"at": snapshot_at}
with read_cursor() as cur:
# --- base ---
cur.execute("SELECT code, name FROM bases WHERE code = 'guanghan'")
base = cur.fetchone()
if base is None:
return result
result["base"] = dict(base)
# --- components present at snapshot_at ---
cur.execute("""
SELECT
c.id,
c.component_key,
c.component_type,
cv.display_name,
cv.summary,
cv.description,
cv.length_m,
cv.width_m,
cv.height_m,
cv.diameter_m,
cv.mass_t,
cv.detail_url,
cv.properties,
intro_mission.code AS introduced_mission,
e_intro.effective_at AS introduced_at
FROM components c
JOIN mission_events e_intro
ON e_intro.id = c.introduced_by_event_id
JOIN missions intro_mission
ON intro_mission.id = e_intro.mission_id
LEFT JOIN mission_events e_retire
ON e_retire.id = c.retired_by_event_id
LEFT JOIN LATERAL (
SELECT cv_sub.*
FROM component_versions cv_sub
WHERE cv_sub.component_id = c.id
AND cv_sub.valid_from <= %(at)s::timestamptz
AND (cv_sub.valid_to IS NULL OR %(at)s::timestamptz < cv_sub.valid_to)
ORDER BY cv_sub.valid_from DESC
LIMIT 1
) cv ON true
WHERE c.base_id = (SELECT id FROM bases WHERE code = 'guanghan')
AND e_intro.effective_at <= %(at)s::timestamptz
AND (c.retired_by_event_id IS NULL
OR (SELECT effective_at FROM mission_events WHERE id = c.retired_by_event_id) > %(at)s::timestamptz)
ORDER BY c.component_key
""", params)
component_rows = cur.fetchall()
component_map = {}
for row in component_rows:
comp = {
"key": row["component_key"],
"type": row["component_type"],
"name": row["display_name"],
"summary": row["summary"],
"description": row["description"],
"detailUrl": row["detail_url"],
"introducedMission": row["introduced_mission"],
"introducedAt": row["introduced_at"].isoformat(),
"dimensions": {},
"properties": row["properties"] or {},
"ports": [],
"states": {},
}
for dim in ["length_m", "width_m", "height_m", "diameter_m"]:
val = row[dim]
if val is not None:
comp["dimensions"][dim.replace("_m", "M")] = float(val)
if row["mass_t"] is not None:
comp["dimensions"]["massT"] = float(row["mass_t"])
result["components"].append(comp)
component_map[row["id"]] = comp
if not component_map:
return result
# --- ports ---
component_ids = list(component_map.keys())
cur.execute("""
SELECT id, component_id, port_key, display_name, direction, interface_type
FROM ports
WHERE component_id = ANY(%(ids)s)
ORDER BY component_id, port_key
""", {"ids": component_ids})
port_map = {}
for row in cur.fetchall():
port = {
"key": row["port_key"],
"displayName": row["display_name"],
"direction": row["direction"],
"interfaceType": row["interface_type"],
}
cid = row["component_id"]
if cid in component_map:
component_map[cid]["ports"].append(port)
port_map[row["id"]] = {
"component_key": component_map.get(cid, {}).get("key"),
**port,
}
# --- states ---
cur.execute("""
SELECT component_id, state_kind, state_value, details
FROM component_states
WHERE component_id = ANY(%(ids)s)
AND valid_from <= %(at)s::timestamptz
AND (valid_to IS NULL OR %(at)s::timestamptz < valid_to)
ORDER BY component_id, state_kind
""", {"ids": component_ids, "at": snapshot_at})
for row in cur.fetchall():
comp = component_map.get(row["component_id"])
if comp:
comp["states"][row["state_kind"]] = row["state_value"]
if row["details"]:
comp["states"][row["state_kind"] + "Details"] = row["details"]
# --- connections ---
cur.execute("""
SELECT
tc.connection_type,
tc.properties,
pa.component_id AS comp_a_id,
pa.port_key AS port_a_key,
pb.component_id AS comp_b_id,
pb.port_key AS port_b_key
FROM topology_connections tc
JOIN ports pa ON pa.id = tc.port_a_id
JOIN ports pb ON pb.id = tc.port_b_id
WHERE tc.base_id = (SELECT id FROM bases WHERE code = 'guanghan')
AND tc.valid_from <= %(at)s::timestamptz
AND (tc.valid_to IS NULL OR %(at)s::timestamptz < tc.valid_to)
AND pa.component_id = ANY(%(ids)s)
AND pb.component_id = ANY(%(ids)s)
ORDER BY tc.connection_type, pa.port_key
""", {"ids": component_ids, "at": snapshot_at})
for row in cur.fetchall():
comp_a = component_map.get(row["comp_a_id"])
comp_b = component_map.get(row["comp_b_id"])
if comp_a and comp_b:
result["connections"].append({
"type": row["connection_type"],
"fromPort": f"{comp_a['key']}.{row['port_a_key']}",
"toPort": f"{comp_b['key']}.{row['port_b_key']}",
"properties": row["properties"],
})
return result
def fetch_layout(layout_key=None):
"""Fetch a named diagram layout, falling back to the default layout."""
with read_cursor() as cur:
cur.execute("""
SELECT
dl.layout_key,
dl.name AS layout_name,
dl.canvas_width,
dl.canvas_height,
COALESCE(layout_theme.id, default_theme.id) AS theme_id,
COALESCE(layout_theme.theme_key, default_theme.theme_key) AS theme_key,
COALESCE(layout_theme.name, default_theme.name) AS theme_name,
COALESCE(layout_theme.tokens, default_theme.tokens) AS theme_tokens,
dn.node_key,
dn.component_id,
c.component_key,
dn.x, dn.y, dn.width, dn.height,
dn.style_key, dn.label_override, dn.z_index,
dn.display_meta, dn.display_hint, dn.decorations,
members.member_keys
FROM diagram_layouts dl
JOIN diagram_nodes dn ON dn.layout_id = dl.id
LEFT JOIN components c ON c.id = dn.component_id
LEFT JOIN LATERAL (
SELECT ARRAY_AGG(
member_component.component_key
ORDER BY member.display_order
) AS member_keys
FROM diagram_node_members member
JOIN components member_component
ON member_component.id = member.component_id
WHERE member.diagram_node_id = dn.id
) members ON true
LEFT JOIN diagram_themes layout_theme
ON layout_theme.id = dl.theme_id
LEFT JOIN diagram_themes default_theme
ON default_theme.base_id = dl.base_id
AND default_theme.is_default = true
WHERE dl.base_id = (SELECT id FROM bases WHERE code = 'guanghan')
AND (
(%(layout_key)s::text IS NOT NULL AND dl.layout_key = %(layout_key)s)
OR (%(layout_key)s::text IS NULL AND dl.is_default = true)
)
ORDER BY dn.z_index, dn.node_key
""", {"layout_key": layout_key})
nodes = []
layout_info = None
theme_info = None
for row in cur.fetchall():
if layout_info is None:
layout_info = {
"key": row["layout_key"],
"name": row["layout_name"],
"canvasWidth": float(row["canvas_width"]),
"canvasHeight": float(row["canvas_height"]),
"themeKey": row["theme_key"],
}
theme_info = {
"id": row["theme_id"],
"key": row["theme_key"],
"name": row["theme_name"],
"tokens": row["theme_tokens"] or {},
}
nodes.append({
"nodeKey": row["node_key"],
"componentKey": row["component_key"],
"x": float(row["x"]),
"y": float(row["y"]),
"width": float(row["width"]),
"height": float(row["height"]),
"styleKey": row["style_key"],
"labelOverride": row["label_override"],
"zIndex": row["z_index"],
"displayMeta": row["display_meta"],
"displayHint": row["display_hint"],
"decorations": row["decorations"] or {},
"memberKeys": row["member_keys"] or [],
})
if layout_info is None:
return None
_require_presentation(theme_info and theme_info["id"], "default theme missing")
theme_id_val = theme_info["id"]
cur.execute("""
SELECT style_key, display_name, fill_color, stroke_color,
stroke_width, corner_radius, title_color, meta_color,
hint_color
FROM diagram_node_styles
WHERE theme_id = %(theme)s
ORDER BY style_key
""", {"theme": theme_id_val})
node_styles = {
row["style_key"]: {
"displayName": row["display_name"],
"fillColor": row["fill_color"],
"strokeColor": row["stroke_color"],
"strokeWidth": float(row["stroke_width"]),
"cornerRadius": float(row["corner_radius"]),
"titleColor": row["title_color"],
"metaColor": row["meta_color"],
"hintColor": row["hint_color"],
}
for row in cur.fetchall()
}
cur.execute("""
SELECT connection_type, display_name, stroke_color, stroke_width,
dash_array, line_cap, legend_group
FROM diagram_edge_styles
WHERE theme_id = %(theme)s
ORDER BY connection_type
""", {"theme": theme_id_val})
edge_styles = {
row["connection_type"]: {
"displayName": row["display_name"],
"strokeColor": row["stroke_color"],
"strokeWidth": float(row["stroke_width"]),
"dashArray": row["dash_array"],
"lineCap": row["line_cap"],
"legendGroup": row["legend_group"],
}
for row in cur.fetchall()
}
cur.execute("""
SELECT decoration_key, display_name, fill_color, stroke_color,
stroke_width, label, label_color, render_params
FROM diagram_decoration_styles
WHERE theme_id = %(theme)s
ORDER BY decoration_key
""", {"theme": theme_id_val})
decoration_styles = {
row["decoration_key"]: {
"displayName": row["display_name"],
"fillColor": row["fill_color"],
"strokeColor": row["stroke_color"],
"strokeWidth": float(row["stroke_width"]),
"label": row["label"],
"labelColor": row["label_color"],
"renderParams": row["render_params"] or {},
}
for row in cur.fetchall()
}
cur.execute("""
SELECT legend_key, label, item_type, style_ref, display_order
FROM diagram_legend_items
WHERE theme_id = %(theme)s
AND is_visible = true
ORDER BY display_order, legend_key
""", {"theme": theme_id_val})
legend_items = [
{
"key": row["legend_key"],
"label": row["label"],
"itemType": row["item_type"],
"styleRef": row["style_ref"],
"displayOrder": row["display_order"],
}
for row in cur.fetchall()
]
missing_node_styles = {
node["styleKey"] for node in nodes
} - set(node_styles)
_require_presentation(
not missing_node_styles,
f"missing node styles: {sorted(missing_node_styles)}",
)
return {
"layout": layout_info,
"nodes": nodes,
"presentation": {
"theme": {
"key": theme_info["key"],
"name": theme_info["name"],
"tokens": theme_info["tokens"],
},
"nodeStyles": node_styles,
"edgeStyles": edge_styles,
"decorationStyles": decoration_styles,
"legendItems": legend_items,
},
}