289 lines
8.4 KiB
Python
289 lines
8.4 KiB
Python
"""Native game-log discovery, freshness validation, and parsing."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
import re
|
|
from typing import Iterable, Mapping
|
|
|
|
|
|
DEFAULT_NATIVE_LOG_PATTERNS = ("log_*.txt", "*.log")
|
|
|
|
_SCENE_RE = re.compile(
|
|
r"\[S1\]\s+after import\s+(?P<scene>[^:]+):\s*"
|
|
r"materials=(?P<materials>\d+)\s+textures=(?P<textures>\d+)",
|
|
re.IGNORECASE,
|
|
)
|
|
_FAILED_FILE_RE = re.compile(
|
|
r"Failed to (?P<operation>find|load|read) file\s*:\s*(?P<path>.*)$",
|
|
re.IGNORECASE,
|
|
)
|
|
_FAILED_IMPORT_RE = re.compile(
|
|
r"Failed to import scene:\s*(?P<scene>.*)$",
|
|
re.IGNORECASE,
|
|
)
|
|
_TIMESTAMP_RE = re.compile(
|
|
r"^\s*(?:\[)?(?P<timestamp>"
|
|
r"\d{4}-\d{2}-\d{2}[T ][0-9:.+Z-]+|\d{2}:\d{2}:\d{2}(?:\.\d+)?)"
|
|
r"(?:\])?"
|
|
)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LogFileState:
|
|
path: Path
|
|
mtime_ns: int
|
|
size: int
|
|
inode: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LogEvent:
|
|
kind: str
|
|
line_number: int
|
|
raw: str
|
|
timestamp_text: str | None = None
|
|
value: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, str | int | None]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SceneImport:
|
|
name: str
|
|
materials: int
|
|
textures: int
|
|
line_number: int
|
|
timestamp_text: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, str | int | None]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass
|
|
class NativeLogAnalysis:
|
|
scenes: dict[str, list[SceneImport]] = field(default_factory=dict)
|
|
events: list[LogEvent] = field(default_factory=list)
|
|
failed_files: list[LogEvent] = field(default_factory=list)
|
|
failed_import_scenes: list[str] = field(default_factory=list)
|
|
load_texture_to_material_failed: bool = False
|
|
import_scene_threw: bool = False
|
|
stage_without_new_stage: bool = False
|
|
tex_reclaim_count: int = 0
|
|
|
|
def latest_scene(self, name: str) -> SceneImport | None:
|
|
wanted = _scene_key(name)
|
|
for scene_name, imports in self.scenes.items():
|
|
if _scene_key(scene_name) == wanted and imports:
|
|
return imports[-1]
|
|
return None
|
|
|
|
def first_event(self, kind: str) -> LogEvent | None:
|
|
return next((event for event in self.events if event.kind == kind), None)
|
|
|
|
def to_summary(self) -> dict[str, object]:
|
|
summary: dict[str, object] = {
|
|
"scenes": {
|
|
name: [item.to_dict() for item in imports]
|
|
for name, imports in sorted(self.scenes.items())
|
|
},
|
|
"load_texture_to_material_failed": self.load_texture_to_material_failed,
|
|
"import_scene_threw": self.import_scene_threw,
|
|
"failed_import_scenes": self.failed_import_scenes,
|
|
"failed_files": [event.to_dict() for event in self.failed_files],
|
|
"stage_without_new_stage": self.stage_without_new_stage,
|
|
"tex_reclaim_count": self.tex_reclaim_count,
|
|
}
|
|
for output_key, scene_name in (
|
|
("whispering_vale", "WhisperingVale"),
|
|
("ricarten", "Ricarten"),
|
|
):
|
|
scene = self.latest_scene(scene_name)
|
|
if scene is not None:
|
|
summary[output_key] = {
|
|
"materials": scene.materials,
|
|
"textures": scene.textures,
|
|
}
|
|
if "whispering_vale" in summary:
|
|
whispering_vale = summary["whispering_vale"]
|
|
assert isinstance(whispering_vale, dict)
|
|
whispering_vale["load_texture_to_material_failed"] = (
|
|
self.load_texture_to_material_failed
|
|
)
|
|
whispering_vale["import_scene_failed"] = self.import_scene_threw or any(
|
|
"whisperingvale" in _scene_key(item)
|
|
for item in self.failed_import_scenes
|
|
)
|
|
return summary
|
|
|
|
|
|
def snapshot_native_logs(
|
|
log_directory: Path,
|
|
patterns: Iterable[str] = DEFAULT_NATIVE_LOG_PATTERNS,
|
|
) -> dict[str, LogFileState]:
|
|
states: dict[str, LogFileState] = {}
|
|
if not log_directory.is_dir():
|
|
return states
|
|
for pattern in patterns:
|
|
for path in log_directory.glob(pattern):
|
|
if not path.is_file():
|
|
continue
|
|
try:
|
|
state = _state(path)
|
|
except FileNotFoundError:
|
|
continue
|
|
states[_path_key(path)] = state
|
|
return states
|
|
|
|
|
|
def is_current_run_log(
|
|
state: LogFileState,
|
|
baseline: Mapping[str, LogFileState],
|
|
test_start_ns: int,
|
|
) -> bool:
|
|
"""Return true only for a new log or a changed post-start existing log.
|
|
|
|
A path absent from the pre-launch baseline is new for this run. An existing
|
|
path must both change and have an mtime at or after the recorded test start.
|
|
The additional change check prevents a stale future-dated file from passing.
|
|
"""
|
|
previous = baseline.get(_path_key(state.path))
|
|
if previous is None:
|
|
return True
|
|
changed = (
|
|
state.mtime_ns != previous.mtime_ns
|
|
or state.size != previous.size
|
|
or state.inode != previous.inode
|
|
)
|
|
return changed and state.mtime_ns >= test_start_ns
|
|
|
|
|
|
def discover_current_native_logs(
|
|
log_directory: Path,
|
|
baseline: Mapping[str, LogFileState],
|
|
test_start_ns: int,
|
|
patterns: Iterable[str] = DEFAULT_NATIVE_LOG_PATTERNS,
|
|
) -> list[LogFileState]:
|
|
current = snapshot_native_logs(log_directory, patterns)
|
|
eligible = [
|
|
state
|
|
for state in current.values()
|
|
if is_current_run_log(state, baseline, test_start_ns)
|
|
]
|
|
return sorted(eligible, key=lambda item: (item.mtime_ns, str(item.path)))
|
|
|
|
|
|
def newest_current_native_log(
|
|
log_directory: Path,
|
|
baseline: Mapping[str, LogFileState],
|
|
test_start_ns: int,
|
|
patterns: Iterable[str] = DEFAULT_NATIVE_LOG_PATTERNS,
|
|
) -> LogFileState | None:
|
|
logs = discover_current_native_logs(log_directory, baseline, test_start_ns, patterns)
|
|
return logs[-1] if logs else None
|
|
|
|
|
|
def parse_native_log(path: Path) -> NativeLogAnalysis:
|
|
with path.open("r", encoding="utf-8", errors="replace") as handle:
|
|
return parse_native_log_lines(handle)
|
|
|
|
|
|
def read_current_run_log_text(
|
|
state: LogFileState,
|
|
baseline: Mapping[str, LogFileState],
|
|
) -> str:
|
|
"""Read only bytes attributable to this run when a log path is reused."""
|
|
previous = baseline.get(_path_key(state.path))
|
|
offset = 0
|
|
if (
|
|
previous is not None
|
|
and state.inode == previous.inode
|
|
and state.size >= previous.size
|
|
):
|
|
offset = previous.size
|
|
with state.path.open("rb") as handle:
|
|
handle.seek(offset)
|
|
return handle.read().decode("utf-8", errors="replace")
|
|
|
|
|
|
def parse_current_run_log(
|
|
state: LogFileState,
|
|
baseline: Mapping[str, LogFileState],
|
|
) -> NativeLogAnalysis:
|
|
return parse_native_log_lines(read_current_run_log_text(state, baseline).splitlines())
|
|
|
|
|
|
def parse_native_log_lines(lines: Iterable[str]) -> NativeLogAnalysis:
|
|
analysis = NativeLogAnalysis()
|
|
for line_number, raw_line in enumerate(lines, start=1):
|
|
raw = raw_line.rstrip("\r\n")
|
|
timestamp = _timestamp(raw)
|
|
scene_match = _SCENE_RE.search(raw)
|
|
if scene_match is not None:
|
|
scene = SceneImport(
|
|
name=scene_match.group("scene").strip(),
|
|
materials=int(scene_match.group("materials")),
|
|
textures=int(scene_match.group("textures")),
|
|
line_number=line_number,
|
|
timestamp_text=timestamp,
|
|
)
|
|
analysis.scenes.setdefault(scene.name, []).append(scene)
|
|
analysis.events.append(
|
|
LogEvent("scene_import", line_number, raw, timestamp, scene.name)
|
|
)
|
|
|
|
if "GRX::SurfaceManager::LoadTextureToMaterial" in raw:
|
|
analysis.load_texture_to_material_failed = True
|
|
analysis.events.append(
|
|
LogEvent("load_texture_to_material", line_number, raw, timestamp)
|
|
)
|
|
if "ImportScene threw" in raw:
|
|
analysis.import_scene_threw = True
|
|
analysis.events.append(LogEvent("import_scene_threw", line_number, raw, timestamp))
|
|
|
|
failed_import = _FAILED_IMPORT_RE.search(raw)
|
|
if failed_import is not None:
|
|
value = failed_import.group("scene").strip()
|
|
analysis.failed_import_scenes.append(value)
|
|
analysis.events.append(
|
|
LogEvent("failed_import_scene", line_number, raw, timestamp, value)
|
|
)
|
|
|
|
failed_file = _FAILED_FILE_RE.search(raw)
|
|
if failed_file is not None:
|
|
operation = failed_file.group("operation").casefold()
|
|
value = failed_file.group("path").strip()
|
|
event = LogEvent(f"failed_{operation}_file", line_number, raw, timestamp, value)
|
|
analysis.failed_files.append(event)
|
|
analysis.events.append(event)
|
|
|
|
if "Stage without new stage" in raw:
|
|
analysis.stage_without_new_stage = True
|
|
analysis.events.append(
|
|
LogEvent("stage_without_new_stage", line_number, raw, timestamp)
|
|
)
|
|
if "TexReclaim" in raw:
|
|
analysis.tex_reclaim_count += 1
|
|
analysis.events.append(LogEvent("tex_reclaim", line_number, raw, timestamp))
|
|
return analysis
|
|
|
|
|
|
def _state(path: Path) -> LogFileState:
|
|
stat = path.stat()
|
|
return LogFileState(path.absolute(), stat.st_mtime_ns, stat.st_size, stat.st_ino)
|
|
|
|
|
|
def _path_key(path: Path) -> str:
|
|
return str(path.absolute())
|
|
|
|
|
|
def _scene_key(value: str) -> str:
|
|
return re.sub(r"[^a-z0-9]", "", value.casefold())
|
|
|
|
|
|
def _timestamp(line: str) -> str | None:
|
|
match = _TIMESTAMP_RE.match(line)
|
|
return match.group("timestamp") if match is not None else None
|