SRETSS/TSSBOT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update for spectra changes (#1363)

Browse Source
This commit is contained in:
NotSoToothless
2026-06-29 11:05:51 -07:00
committed by GitHub
parent fa203c653b
commit 6d251be97c
4 changed files with 79 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
BOT/render_replay.py
+39 -14
View File
@@ -3575,6 +3575,10 @@ def _unit_to_model_name(unit_name: str) -> str:
internal = (unit_name or "").strip()
if not internal:
return "tankModels/unknown"
if internal.startswith(("tankModels/", "airModels/")):
return internal
if internal.startswith("aircrafts/"):
return f"airModels/{internal.split('/', 1)[1]}"
tags = _get_unit_tags(internal) or []
tag_set = set(tags)
if "type_strike_ucav" in tag_set or "ucav" in internal.lower():
@@ -3645,6 +3649,15 @@ def _position_at_time(path: list[dict[str, float]], time_ms: float) -> dict[str,
return prev
def _event_position_to_dict(pos: Any) -> dict[str, float] | None:
if not isinstance(pos, list) or len(pos) < 3:
return None
try:
return {"X": float(pos[0]), "Y": float(pos[1]), "Z": float(pos[2])}
except (TypeError, ValueError):
return None
def _find_render_entity_for_event(entities: list[dict[str, Any]],
player_id: int,
event_model: str | None,
@@ -3739,7 +3752,7 @@ def _convert_ws_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, Any]:
if not isinstance(ent, dict):
continue
uid = _to_int(ent.get("uid"), 0)
unit = str(ent.get("unit") or "")
unit = str(ent.get("model_path") or ent.get("unit") or "")
path_raw = ent.get("path") or []
if not isinstance(path_raw, list):
continue
@@ -3775,12 +3788,18 @@ def _convert_ws_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, Any]:
victim_id = _to_int(kill.get("offended_uid"), 0)
killer_id = _to_int(kill.get("offender_uid"), 0)
kill_time = float(kill.get("time") or 0.0)
victim_model = _unit_to_model_name(str(kill.get("offended_unit") or ""))
killer_model = _unit_to_model_name(str(kill.get("offender_unit") or ""))
victim_model = _unit_to_model_name(str(kill.get("offended_unit_model_path") or kill.get("offended_unit") or ""))
killer_model = _unit_to_model_name(str(kill.get("offender_unit_model_path") or kill.get("offender_unit") or ""))
victim_entity = _find_render_entity_for_event(entities_out, victim_id, victim_model, kill_time)
killer_entity = _find_render_entity_for_event(entities_out, killer_id, killer_model, kill_time)
victim_pos = _position_at_time(victim_entity.get("Path", []), kill_time) if victim_entity else None
killer_pos = _position_at_time(killer_entity.get("Path", []), kill_time) if killer_entity else None
victim_pos = (
_position_at_time(victim_entity.get("Path", []), kill_time)
if victim_entity else None
) or _event_position_to_dict(kill.get("offended_pos"))
killer_pos = (
_position_at_time(killer_entity.get("Path", []), kill_time)
if killer_entity else None
) or _event_position_to_dict(kill.get("offender_pos"))
payload: dict[str, Any] = {
"Time": kill_time,
"VictimID": victim_id,
@@ -3816,8 +3835,8 @@ def _convert_ws_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, Any]:
"Time": float(dmg.get("time") or 0.0),
"OffenderID": _to_int(dmg.get("offender_uid"), 0),
"OffendedID": _to_int(dmg.get("offended_uid"), 0),
"OffenderModel": _unit_to_model_name(str(dmg.get("offender_unit") or "")),
"OffendedModel": _unit_to_model_name(str(dmg.get("offended_unit") or "")),
"OffenderModel": _unit_to_model_name(str(dmg.get("offender_unit_model_path") or dmg.get("offender_unit") or "")),
"OffendedModel": _unit_to_model_name(str(dmg.get("offended_unit_model_path") or dmg.get("offended_unit") or "")),
"Afire": bool(dmg.get("afire", False)),
})
@@ -3905,7 +3924,7 @@ def _convert_local_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, An
if not isinstance(ent, dict):
continue
uid = _to_int(ent.get("uid"), 0)
unit = str(ent.get("unit") or "")
unit = str(ent.get("model_path") or ent.get("unit") or "")
path_raw = ent.get("path") or []
if not isinstance(path_raw, list):
continue
@@ -3941,12 +3960,18 @@ def _convert_local_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, An
victim_id = _to_int(kill.get("offended_uid"), 0)
killer_id = _to_int(kill.get("offender_uid"), 0)
kill_time = float(kill.get("time") or 0.0)
victim_model = _unit_to_model_name(str(kill.get("offended_unit") or ""))
killer_model = _unit_to_model_name(str(kill.get("offender_unit") or ""))
victim_model = _unit_to_model_name(str(kill.get("offended_unit_model_path") or kill.get("offended_unit") or ""))
killer_model = _unit_to_model_name(str(kill.get("offender_unit_model_path") or kill.get("offender_unit") or ""))
victim_entity = _find_render_entity_for_event(entities_out, victim_id, victim_model, kill_time)
killer_entity = _find_render_entity_for_event(entities_out, killer_id, killer_model, kill_time)
victim_pos = _position_at_time(victim_entity.get("Path", []), kill_time) if victim_entity else None
killer_pos = _position_at_time(killer_entity.get("Path", []), kill_time) if killer_entity else None
victim_pos = (
_position_at_time(victim_entity.get("Path", []), kill_time)
if victim_entity else None
) or _event_position_to_dict(kill.get("offended_pos"))
killer_pos = (
_position_at_time(killer_entity.get("Path", []), kill_time)
if killer_entity else None
) or _event_position_to_dict(kill.get("offender_pos"))
payload: dict[str, Any] = {
"Time": kill_time,
"VictimID": victim_id,
@@ -3982,8 +4007,8 @@ def _convert_local_replay_to_render_dict(replay: dict[str, Any]) -> dict[str, An
"Time": float(dmg.get("time") or 0.0),
"OffenderID": _to_int(dmg.get("offender_uid"), 0),
"OffendedID": _to_int(dmg.get("offended_uid"), 0),
"OffenderModel": _unit_to_model_name(str(dmg.get("offender_unit") or "")),
"OffendedModel": _unit_to_model_name(str(dmg.get("offended_unit") or "")),
"OffenderModel": _unit_to_model_name(str(dmg.get("offender_unit_model_path") or dmg.get("offender_unit") or "")),
"OffendedModel": _unit_to_model_name(str(dmg.get("offended_unit_model_path") or dmg.get("offended_unit") or "")),
"Afire": bool(dmg.get("afire", False)),
})
BOT/transform.py
+8 -2
View File
@@ -14,6 +14,8 @@ from __future__ import annotations
import logging
from typing import Any, Optional
from spectra_replay_normalize import strip_model_prefix
log = logging.getLogger("tssbot.transform")
# Imported lazily/defensively so the module is importable (and unit-testable for the
@@ -61,7 +63,7 @@ def _build_units(units: list[dict[str, Any]], translate, dead: set[str] | None =
dead = dead or set()
out: list[dict[str, Any]] = []
for u in units or []:
internal = str(u.get("unit") or "").strip()
internal = strip_model_prefix(u.get("unit"))
if not internal:
continue
flag = u.get("dead", u.get("died"))
@@ -71,6 +73,9 @@ def _build_units(units: list[dict[str, Any]], translate, dead: set[str] | None =
"name": translate(internal) or u.get("unit_normalized") or internal,
"used": bool(u.get("used")),
"dead": is_dead,
"unit_type": u.get("unit_type"),
"unit_class": u.get("unit_class"),
"unit_country": u.get("unit_country"),
})
return out
@@ -81,7 +86,7 @@ def _dead_units_by_uid(game: dict[str, Any]) -> dict[str, set[str]]:
kills = ((game.get("events") or {}).get("kills")) or []
for k in kills:
uid = str(k.get("offended_uid") or "")
unit = str(k.get("offended_unit") or "").strip()
unit = strip_model_prefix(k.get("offended_unit"))
if uid and unit:
out.setdefault(uid, set()).add(unit)
return out
@@ -143,6 +148,7 @@ def build_scoreboard_model(game: dict[str, Any], lang_column: str = "<English>")
"fake_nick": None,
"tag": str(p.get("tag") or ""),
"country_id": p.get("country_id"),
"country": p.get("country"),
"air_kills": int(p.get("air_kills") or 0),
"ground_kills": int(p.get("ground_kills") or 0),
"assists": int(p.get("assists") or 0),
tests/test_transform_model.py
+26
View File
@@ -29,3 +29,29 @@ def test_model_carries_tournament_id():
model = build_scoreboard_model(_game())
assert model is not None
assert model["tournament_id"] == 24965
def test_model_strips_unit_model_prefixes_for_dead_matching():
game = _game()
game["players"]["2"]["units"][0]["unit"] = "tankModels/pz"
game["players"]["2"]["units"][0]["unit_type"] = "tank"
game["players"]["2"]["units"][0]["unit_class"] = "medium tank"
game["players"]["2"]["country"] = "germany"
game["events"] = {
"kills": [
{
"offended_uid": "2",
"offended_unit": "tankModels/pz",
}
]
}
model = build_scoreboard_model(game)
assert model is not None
bob = model["teams"][1]["players"][0]
assert bob["country"] == "germany"
assert bob["units"][0]["internal"] == "pz"
assert bob["units"][0]["dead"] is True
assert bob["units"][0]["unit_type"] == "tank"
assert bob["units"][0]["unit_class"] == "medium tank"
tss_ws.py
+6
View File
@@ -31,6 +31,7 @@ from websockets.asyncio.client import connect as wsconnect
from BOT.storage import insert_match, insert_player_games, upsert_tss_teams
from BOT.autologging import process_game as autolog_process_game
from BOT.receiver_bridge import publish_replay_batch
from spectra_replay_normalize import normalize_spectra_replay
from spectra_ws_payload import SpectraPayloadError, decode_spectra_ws_payload
_HERE = Path(__file__).resolve().parent
@@ -73,6 +74,7 @@ def _session_id(game: Dict[str, Any]) -> str:
def _write_game(game: Dict[str, Any]) -> Path:
"""Normalize _id to hex, then write to REPLAYS/TSS/<session_id>/replay_data.json.gz."""
game.update(normalize_spectra_replay(game))
sid = _session_id(game)
game["_id"] = sid # hex from this point forward
session_dir = REPLAYS_DIR / sid
@@ -91,6 +93,10 @@ def normalize(data: Any) -> Optional[List[Dict[str, Any]]]:
if isinstance(data, dict):
if "_id" in data or "id" in data:
return [data]
if isinstance(data.get("data"), dict):
return [data["data"]]
if isinstance(data.get("data"), list):
return data["data"]
if "completed" in data:
return data["completed"]
log.warning("Unknown WS frame shape: %s", type(data))