Move reverse engineering artifacts under contrib
This commit is contained in:
@@ -1,203 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Summarize GPS-relevant structure in all.traffic_persistent JSON."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import collections
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
FLAGS = {
|
||||
"FromRoadSpline": 1,
|
||||
"Bidirectional": 2,
|
||||
"PatrolRoute": 4,
|
||||
"Pavement": 8,
|
||||
"Road": 16,
|
||||
"Intersection": 32,
|
||||
"NeverDeadEnd": 64,
|
||||
"TrafficDisabled": 128,
|
||||
"CrossWalk": 256,
|
||||
"GPSOnly": 512,
|
||||
"ShowDebug": 1024,
|
||||
"Blockade": 2048,
|
||||
"Yield": 4096,
|
||||
"NoAIDriving": 8192,
|
||||
"Highway": 16384,
|
||||
"NoAutodrive": 32768,
|
||||
}
|
||||
|
||||
|
||||
def flag_value(flags: Any) -> int:
|
||||
if isinstance(flags, int):
|
||||
return flags
|
||||
if isinstance(flags, str):
|
||||
return int(flags)
|
||||
if isinstance(flags, dict):
|
||||
if "Value" in flags:
|
||||
return int(flags["Value"])
|
||||
if "$value" in flags:
|
||||
return int(flags["$value"])
|
||||
return int(flags or 0)
|
||||
|
||||
|
||||
def has(flags: int, name: str) -> bool:
|
||||
return bool(flags & FLAGS[name])
|
||||
|
||||
|
||||
def category(flags: int) -> str:
|
||||
if has(flags, "Highway"):
|
||||
return "highway"
|
||||
if has(flags, "GPSOnly"):
|
||||
return "gpsonly"
|
||||
if has(flags, "Road"):
|
||||
return "road"
|
||||
if has(flags, "Pavement"):
|
||||
return "pavement"
|
||||
return "other"
|
||||
|
||||
|
||||
def ref_index(ref: Any) -> int | None:
|
||||
if isinstance(ref, int):
|
||||
return ref
|
||||
if isinstance(ref, str):
|
||||
try:
|
||||
return int(ref)
|
||||
except ValueError:
|
||||
return None
|
||||
if isinstance(ref, dict):
|
||||
for key in ("Index", "index", "laneIndex", "LaneIndex", "value", "$value"):
|
||||
if key in ref:
|
||||
return ref_index(ref[key])
|
||||
return None
|
||||
|
||||
|
||||
def lane_refs(value: Any) -> list[int]:
|
||||
if value is None:
|
||||
return []
|
||||
if isinstance(value, list):
|
||||
out: list[int] = []
|
||||
for item in value:
|
||||
idx = ref_index(item)
|
||||
if idx is not None:
|
||||
out.append(idx)
|
||||
return out
|
||||
if isinstance(value, dict) and "Elements" in value:
|
||||
return lane_refs(value["Elements"])
|
||||
return []
|
||||
|
||||
|
||||
def gps_info(lane: dict[str, Any]) -> tuple[int | None, int | None]:
|
||||
info = lane.get("playerGPSInfo") or {}
|
||||
return info.get("subGraphId"), info.get("stronglyConnectedComponentId")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("json_file", type=Path)
|
||||
parser.add_argument("--lane-connections", type=Path)
|
||||
parser.add_argument("--samples", type=int, default=12)
|
||||
args = parser.parse_args()
|
||||
|
||||
data = json.loads(args.json_file.read_text(encoding="utf-8"))
|
||||
root = data["Data"]["RootChunk"]["data"]
|
||||
lanes = root["lanes"]
|
||||
neighbor_groups = root.get("neighborGroups", [])
|
||||
|
||||
categories = []
|
||||
flags_by_lane = []
|
||||
for lane in lanes:
|
||||
flags = flag_value(lane.get("flags", 0))
|
||||
flags_by_lane.append(flags)
|
||||
categories.append(category(flags))
|
||||
|
||||
print(f"lanes: {len(lanes)}")
|
||||
print(f"neighborGroups: {len(neighbor_groups)}")
|
||||
print("categories:", dict(collections.Counter(categories).most_common()))
|
||||
|
||||
gps_counter = collections.Counter(gps_info(lane) for lane in lanes)
|
||||
print(f"gps components: {len(gps_counter)}")
|
||||
print("top gps components:")
|
||||
for (subgraph, scc), count in gps_counter.most_common(20):
|
||||
cats = collections.Counter(
|
||||
categories[i] for i, lane in enumerate(lanes) if gps_info(lane) == (subgraph, scc)
|
||||
)
|
||||
print(f" subGraph={subgraph} scc={scc}: {count} {dict(cats.most_common())}")
|
||||
|
||||
connection_rows: list[dict[str, Any]] | None = None
|
||||
if args.lane_connections:
|
||||
connection_data = json.loads(args.lane_connections.read_text(encoding="utf-8"))
|
||||
connection_rows = connection_data["Data"]["RootChunk"]["data"]
|
||||
|
||||
transitions = collections.Counter()
|
||||
probabilities: dict[tuple[str, str], collections.Counter[int]] = collections.defaultdict(
|
||||
collections.Counter
|
||||
)
|
||||
sharp_angles = collections.Counter()
|
||||
out_degree = collections.Counter()
|
||||
missing_refs = 0
|
||||
rows = enumerate(lanes) if connection_rows is None else (
|
||||
(row["index"], row["value"]) for row in connection_rows
|
||||
)
|
||||
for i, lane_or_connections in rows:
|
||||
refs = (
|
||||
lane_refs(lane_or_connections.get("outLanes"))
|
||||
if connection_rows is None
|
||||
else lane_refs(lane_or_connections.get("outlanes"))
|
||||
)
|
||||
out_degree[len(refs)] += 1
|
||||
raw_outs = (
|
||||
lane_or_connections.get("outLanes", [])
|
||||
if connection_rows is None
|
||||
else lane_or_connections.get("outlanes", [])
|
||||
)
|
||||
for out, ref in zip(raw_outs, refs):
|
||||
if ref < 0 or ref >= len(lanes):
|
||||
missing_refs += 1
|
||||
continue
|
||||
pair = (categories[i], categories[ref])
|
||||
transitions[pair] += 1
|
||||
if isinstance(out, dict) and "exitProbabilityCompressed" in out:
|
||||
probabilities[pair][out["exitProbabilityCompressed"]] += 1
|
||||
if isinstance(out, dict) and "isSharpAngle" in out:
|
||||
sharp_angles[(pair, out["isSharpAngle"])] += 1
|
||||
print("out degree:", dict(sorted(out_degree.items())))
|
||||
print(f"missing out refs: {missing_refs}")
|
||||
print("category transitions:")
|
||||
for (src, dst), count in transitions.most_common(30):
|
||||
extra = ""
|
||||
if probabilities[(src, dst)]:
|
||||
extra = f" prob={probabilities[(src, dst)].most_common(6)}"
|
||||
sharp = [(key[1], value) for key, value in sharp_angles.items() if key[0] == (src, dst)]
|
||||
if sharp:
|
||||
extra += f" sharp={sharp}"
|
||||
print(f" {src:8s} -> {dst:8s}: {count}{extra}")
|
||||
|
||||
print("highway connector samples:")
|
||||
shown = 0
|
||||
for i, lane in enumerate(lanes):
|
||||
if categories[i] != "highway":
|
||||
continue
|
||||
if connection_rows is None:
|
||||
refs = [ref for ref in lane_refs(lane.get("outLanes")) if 0 <= ref < len(lanes)]
|
||||
else:
|
||||
row = connection_rows[i]["value"]
|
||||
refs = [ref for ref in lane_refs(row.get("outlanes")) if 0 <= ref < len(lanes)]
|
||||
ref_cats = collections.Counter(categories[ref] for ref in refs)
|
||||
if ref_cats and any(cat != "highway" for cat in ref_cats):
|
||||
subgraph, scc = gps_info(lane)
|
||||
print(
|
||||
f" lane={i} flags={flags_by_lane[i]} len={lane.get('length')} "
|
||||
f"speed={lane.get('maxSpeed')} gps=({subgraph},{scc}) out={dict(ref_cats)}"
|
||||
)
|
||||
shown += 1
|
||||
if shown >= args.samples:
|
||||
break
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,77 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from patch_traffic_lanes import FLAGS, as_int, has_flag, looks_like_lane
|
||||
|
||||
|
||||
def visit(value: Any):
|
||||
if isinstance(value, dict):
|
||||
lanes = value.get("lanes")
|
||||
if isinstance(lanes, list) and any(looks_like_lane(lane) for lane in lanes):
|
||||
for lane in lanes:
|
||||
if looks_like_lane(lane):
|
||||
yield lane
|
||||
for child in value.values():
|
||||
yield from visit(child)
|
||||
elif isinstance(value, list):
|
||||
for child in value:
|
||||
yield from visit(child)
|
||||
|
||||
|
||||
def bucket(flags: int) -> str:
|
||||
if has_flag(flags, "Highway"):
|
||||
return "highway"
|
||||
if has_flag(flags, "Pavement"):
|
||||
return "pavement"
|
||||
if has_flag(flags, "CrossWalk"):
|
||||
return "crosswalk"
|
||||
if has_flag(flags, "Road"):
|
||||
return "road"
|
||||
return "other"
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("json_file", type=Path)
|
||||
args = parser.parse_args()
|
||||
|
||||
data = json.loads(args.json_file.read_text(encoding="utf-8"))
|
||||
speeds: dict[str, Counter[int]] = {
|
||||
"all": Counter(),
|
||||
"highway": Counter(),
|
||||
"road": Counter(),
|
||||
"pavement": Counter(),
|
||||
"crosswalk": Counter(),
|
||||
"other": Counter(),
|
||||
}
|
||||
flags_counter: Counter[int] = Counter()
|
||||
|
||||
for lane in visit(data):
|
||||
flags = as_int(lane.get("flags"))
|
||||
speed = as_int(lane.get("maxSpeed"))
|
||||
if flags is None or speed is None:
|
||||
continue
|
||||
speeds["all"][speed] += 1
|
||||
speeds[bucket(flags)][speed] += 1
|
||||
flags_counter[flags] += 1
|
||||
|
||||
for name, counter in speeds.items():
|
||||
print(f"{name}: {sum(counter.values())} lanes")
|
||||
print(" " + ", ".join(f"{speed}:{count}" for speed, count in counter.most_common(20)))
|
||||
|
||||
print("top flags:")
|
||||
for flags, count in flags_counter.most_common(20):
|
||||
names = [name for name, bit in FLAGS.items() if flags & bit]
|
||||
print(f" {flags}: {count} ({', '.join(names)})")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,213 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Summarize VAND navigation-graph blobs from streamingsector JSON files."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import collections
|
||||
import dataclasses
|
||||
import json
|
||||
import math
|
||||
import struct
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterator
|
||||
|
||||
|
||||
HEADER_SIZE = 0x64
|
||||
POINT_SIZE = 0x14
|
||||
COORD_SIZE = 0x0C
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class VandBlob:
|
||||
path: Path
|
||||
ordinal: int
|
||||
size: int
|
||||
version: int
|
||||
tile_x: int
|
||||
tile_y: int
|
||||
point_count: int
|
||||
coord_count: int
|
||||
aux_count: int
|
||||
point_count_2: int
|
||||
aux_count_2: int
|
||||
point_count_3: int
|
||||
class_counts: collections.Counter[int]
|
||||
raw_class_counts: collections.Counter[int]
|
||||
mask_counts: collections.Counter[int]
|
||||
bounds: tuple[float, float, float, float, float, float] | None
|
||||
|
||||
|
||||
def walk_json(value: Any) -> Iterator[str]:
|
||||
if isinstance(value, dict):
|
||||
blob = value.get("Bytes")
|
||||
if isinstance(blob, str):
|
||||
yield blob
|
||||
for child in value.values():
|
||||
yield from walk_json(child)
|
||||
elif isinstance(value, list):
|
||||
for child in value:
|
||||
yield from walk_json(child)
|
||||
|
||||
|
||||
def finite_bounds(coords: list[tuple[float, float, float]]) -> tuple[float, float, float, float, float, float] | None:
|
||||
finite = [coord for coord in coords if all(math.isfinite(part) for part in coord)]
|
||||
if not finite:
|
||||
return None
|
||||
|
||||
xs = [coord[0] for coord in finite]
|
||||
ys = [coord[1] for coord in finite]
|
||||
zs = [coord[2] for coord in finite]
|
||||
return min(xs), min(ys), min(zs), max(xs), max(ys), max(zs)
|
||||
|
||||
|
||||
def parse_vand(path: Path, ordinal: int, data: bytes) -> VandBlob | None:
|
||||
if len(data) < HEADER_SIZE or not data.startswith(b"VAND"):
|
||||
return None
|
||||
|
||||
version = struct.unpack_from("<I", data, 0x04)[0]
|
||||
tile_x = struct.unpack_from("<i", data, 0x08)[0]
|
||||
tile_y = struct.unpack_from("<i", data, 0x0C)[0]
|
||||
point_count = struct.unpack_from("<I", data, 0x18)[0]
|
||||
coord_count = struct.unpack_from("<I", data, 0x1C)[0]
|
||||
aux_count = struct.unpack_from("<I", data, 0x20)[0]
|
||||
point_count_2 = struct.unpack_from("<I", data, 0x24)[0]
|
||||
aux_count_2 = struct.unpack_from("<I", data, 0x28)[0]
|
||||
point_count_3 = struct.unpack_from("<I", data, 0x2C)[0]
|
||||
|
||||
coord_offset = HEADER_SIZE
|
||||
point_offset = coord_offset + coord_count * COORD_SIZE
|
||||
point_end = point_offset + point_count * POINT_SIZE
|
||||
if point_end > len(data):
|
||||
return VandBlob(
|
||||
path=path,
|
||||
ordinal=ordinal,
|
||||
size=len(data),
|
||||
version=version,
|
||||
tile_x=tile_x,
|
||||
tile_y=tile_y,
|
||||
point_count=point_count,
|
||||
coord_count=coord_count,
|
||||
aux_count=aux_count,
|
||||
point_count_2=point_count_2,
|
||||
aux_count_2=aux_count_2,
|
||||
point_count_3=point_count_3,
|
||||
class_counts=collections.Counter({-1: point_count}),
|
||||
raw_class_counts=collections.Counter(),
|
||||
mask_counts=collections.Counter(),
|
||||
bounds=None,
|
||||
)
|
||||
|
||||
coords: list[tuple[float, float, float]] = []
|
||||
for index in range(coord_count):
|
||||
coords.append(struct.unpack_from("<fff", data, coord_offset + index * COORD_SIZE))
|
||||
|
||||
class_counts: collections.Counter[int] = collections.Counter()
|
||||
raw_class_counts: collections.Counter[int] = collections.Counter()
|
||||
mask_counts: collections.Counter[int] = collections.Counter()
|
||||
for index in range(point_count):
|
||||
offset = point_offset + index * POINT_SIZE
|
||||
mask = struct.unpack_from("<H", data, offset + 0x10)[0]
|
||||
raw_class = data[offset + 0x13]
|
||||
class_counts[raw_class & 0x3F] += 1
|
||||
raw_class_counts[raw_class] += 1
|
||||
mask_counts[mask] += 1
|
||||
|
||||
return VandBlob(
|
||||
path=path,
|
||||
ordinal=ordinal,
|
||||
size=len(data),
|
||||
version=version,
|
||||
tile_x=tile_x,
|
||||
tile_y=tile_y,
|
||||
point_count=point_count,
|
||||
coord_count=coord_count,
|
||||
aux_count=aux_count,
|
||||
point_count_2=point_count_2,
|
||||
aux_count_2=aux_count_2,
|
||||
point_count_3=point_count_3,
|
||||
class_counts=class_counts,
|
||||
raw_class_counts=raw_class_counts,
|
||||
mask_counts=mask_counts,
|
||||
bounds=finite_bounds(coords),
|
||||
)
|
||||
|
||||
|
||||
def iter_vand_blobs(path: Path) -> Iterator[VandBlob]:
|
||||
document = json.loads(path.read_text(encoding="utf-8"))
|
||||
ordinal = 0
|
||||
for encoded in walk_json(document):
|
||||
try:
|
||||
data = base64.b64decode(encoded)
|
||||
except ValueError:
|
||||
continue
|
||||
blob = parse_vand(path, ordinal, data)
|
||||
if blob:
|
||||
yield blob
|
||||
ordinal += 1
|
||||
|
||||
|
||||
def fmt_counter(counter: collections.Counter[int], limit: int) -> str:
|
||||
return " ".join(f"{key}:{value}" for key, value in counter.most_common(limit))
|
||||
|
||||
|
||||
def fmt_bounds(bounds: tuple[float, float, float, float, float, float] | None) -> str:
|
||||
if not bounds:
|
||||
return "<none>"
|
||||
return (
|
||||
f"({bounds[0]:.1f},{bounds[1]:.1f},{bounds[2]:.1f}).."
|
||||
f"({bounds[3]:.1f},{bounds[4]:.1f},{bounds[5]:.1f})"
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("json_files", nargs="+", type=Path)
|
||||
parser.add_argument("--detail", type=int, default=12)
|
||||
parser.add_argument("--counter-limit", type=int, default=12)
|
||||
args = parser.parse_args()
|
||||
|
||||
blobs: list[VandBlob] = []
|
||||
for path in args.json_files:
|
||||
blobs.extend(iter_vand_blobs(path))
|
||||
|
||||
total_points = sum(blob.point_count for blob in blobs)
|
||||
total_coords = sum(blob.coord_count for blob in blobs)
|
||||
class_counts: collections.Counter[int] = collections.Counter()
|
||||
raw_counts: collections.Counter[int] = collections.Counter()
|
||||
mask_counts: collections.Counter[int] = collections.Counter()
|
||||
versions: collections.Counter[int] = collections.Counter()
|
||||
files: collections.Counter[str] = collections.Counter()
|
||||
for blob in blobs:
|
||||
class_counts.update(blob.class_counts)
|
||||
raw_counts.update(blob.raw_class_counts)
|
||||
mask_counts.update(blob.mask_counts)
|
||||
versions[blob.version] += 1
|
||||
files[str(blob.path)] += 1
|
||||
|
||||
print(f"files={len(files)} blobs={len(blobs)} points={total_points} coords={total_coords}")
|
||||
print(f"versions={dict(sorted(versions.items()))}")
|
||||
print(f"classes={fmt_counter(class_counts, args.counter_limit)}")
|
||||
print(f"raw13={fmt_counter(raw_counts, args.counter_limit)}")
|
||||
print(f"masks={fmt_counter(mask_counts, args.counter_limit)}")
|
||||
print("files:")
|
||||
for path, count in files.most_common():
|
||||
print(f" {path}: {count}")
|
||||
|
||||
if args.detail:
|
||||
print("largest blobs:")
|
||||
for blob in sorted(blobs, key=lambda item: item.point_count, reverse=True)[: args.detail]:
|
||||
print(
|
||||
f" {blob.path.name}#{blob.ordinal} size={blob.size} tile=({blob.tile_x},{blob.tile_y}) "
|
||||
f"points={blob.point_count} coords={blob.coord_count} aux={blob.aux_count} "
|
||||
f"classes={fmt_counter(blob.class_counts, args.counter_limit)} "
|
||||
f"raw13={fmt_counter(blob.raw_class_counts, args.counter_limit)} "
|
||||
f"bounds={fmt_bounds(blob.bounds)}"
|
||||
)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,53 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
if [[ $# -lt 3 ]]; then
|
||||
echo "usage: $0 <game-dir> <archive-or-archive-dir> <work-dir> [resource-regex] [resource-path] [mod-name]" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
game_dir=$1
|
||||
archive_path=$2
|
||||
work_dir=$3
|
||||
resource_regex=${4:-'all\.traffic_persistent$'}
|
||||
resource_path=${5:-'base/worlds/03_night_city/sectors/_generated/traffic/all.traffic_persistent'}
|
||||
mod_name=${6:-zz_edge_weight_gps}
|
||||
|
||||
raw_dir="${work_dir}/01_raw"
|
||||
json_dir="${work_dir}/02_json"
|
||||
patched_json_dir="${work_dir}/03_json_patched"
|
||||
patched_raw_dir="${work_dir}/04_raw_patched"
|
||||
pack_root="${work_dir}/${mod_name}"
|
||||
packed_dir="${work_dir}/05_packed"
|
||||
resource_dir=$(dirname "${resource_path}")
|
||||
resource_file=$(basename "${resource_path}")
|
||||
resource_hash=3419764573789342681
|
||||
|
||||
mkdir -p "${raw_dir}/${resource_dir}" "${json_dir}" "${patched_json_dir}" "${patched_raw_dir}" "${pack_root}/${resource_dir}" "${packed_dir}"
|
||||
|
||||
if [[ -f "${archive_path}" ]]; then
|
||||
kark_path="${raw_dir}/${resource_file}.kark"
|
||||
decompressed_base="${raw_dir}/${resource_file}.raw"
|
||||
python3 tools/extract_archive_segment.py "${archive_path}" "${resource_hash}" "${kark_path}"
|
||||
tools/cp77_toolbox.sh oodle decompress "${kark_path}" "${decompressed_base}" || true
|
||||
if [[ ! -f "${decompressed_base}.bin" ]]; then
|
||||
echo "failed to decompress ${kark_path}" >&2
|
||||
exit 1
|
||||
fi
|
||||
mv "${decompressed_base}.bin" "${raw_dir}/${resource_path}"
|
||||
else
|
||||
echo "raw segment extraction needs a single archive file, got: ${archive_path}" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
tools/cp77_toolbox.sh convert serialize "${raw_dir}" --outpath "${json_dir}"
|
||||
|
||||
python3 tools/patch_traffic_lanes.py "${json_dir}" "${patched_json_dir}" --copy-unchanged
|
||||
|
||||
tools/cp77_toolbox.sh convert deserialize "${patched_json_dir}" --outpath "${patched_raw_dir}"
|
||||
|
||||
cp "${patched_raw_dir}/${resource_file}" "${pack_root}/${resource_path}"
|
||||
|
||||
tools/cp77_toolbox.sh pack "${pack_root}" --outpath "${packed_dir}"
|
||||
|
||||
echo "packed archive output: ${packed_dir}"
|
||||
@@ -1,7 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
container="${CP77_TOOLBOX_CONTAINER:-2077}"
|
||||
tool="${HOME}/.dotnet/tools/cp77tools"
|
||||
|
||||
exec toolbox run --container "${container}" "${tool}" "$@"
|
||||
@@ -1,101 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Disassemble a PE .text range by RVA and annotate direct branch targets."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import math
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
from capstone import Cs, CS_ARCH_X86, CS_MODE_64
|
||||
from capstone.x86 import X86_OP_IMM, X86_OP_MEM, X86_REG_RIP
|
||||
|
||||
from find_pe_string_xrefs import Section, parse_pe
|
||||
|
||||
|
||||
def find_section(sections: list[Section], rva: int) -> Section:
|
||||
for section in sections:
|
||||
if section.contains_rva(rva):
|
||||
return section
|
||||
raise ValueError(f"RVA 0x{rva:x} is not inside any PE section")
|
||||
|
||||
|
||||
def find_section_or_none(sections: list[Section], rva: int) -> Section | None:
|
||||
for section in sections:
|
||||
if section.contains_rva(rva):
|
||||
return section
|
||||
return None
|
||||
|
||||
|
||||
def parse_rva(text: str) -> int:
|
||||
return int(text, 16 if text.lower().startswith("0x") else 16)
|
||||
|
||||
|
||||
def format_memory_annotation(data: bytes, sections: list[Section], target_rva: int) -> str:
|
||||
section = find_section_or_none(sections, target_rva)
|
||||
if section is None:
|
||||
return f"rip_rva=0x{target_rva:x}"
|
||||
|
||||
offset = section.rva_to_offset(target_rva)
|
||||
if offset < 0 or offset >= len(data):
|
||||
return f"rip_rva=0x{target_rva:x}"
|
||||
|
||||
available = data[offset : min(len(data), offset + 8)]
|
||||
fields = [f"rip_rva=0x{target_rva:x}"]
|
||||
if len(available) >= 4:
|
||||
u32 = struct.unpack_from("<I", available)[0]
|
||||
f32 = struct.unpack_from("<f", available)[0]
|
||||
fields.append(f"u32=0x{u32:08x}")
|
||||
if math.isfinite(f32) and abs(f32) < 1.0e12:
|
||||
fields.append(f"f32={f32:.9g}")
|
||||
if len(available) >= 8:
|
||||
u64 = struct.unpack_from("<Q", available)[0]
|
||||
fields.append(f"u64=0x{u64:016x}")
|
||||
return " ".join(fields)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("pe", type=Path)
|
||||
parser.add_argument("rva", type=parse_rva)
|
||||
parser.add_argument("--size", type=lambda value: int(value, 0), default=0x200)
|
||||
parser.add_argument("--before", type=lambda value: int(value, 0), default=0)
|
||||
args = parser.parse_args()
|
||||
|
||||
data = args.pe.read_bytes()
|
||||
image_base, sections = parse_pe(data)
|
||||
start_rva = max(0, args.rva - args.before)
|
||||
section = find_section(sections, start_rva)
|
||||
end_rva = min(section.virtual_address + section.raw_size, start_rva + args.size)
|
||||
start_offset = section.rva_to_offset(start_rva)
|
||||
code = data[start_offset : start_offset + (end_rva - start_rva)]
|
||||
|
||||
disassembler = Cs(CS_ARCH_X86, CS_MODE_64)
|
||||
disassembler.detail = True
|
||||
|
||||
print(f"image_base=0x{image_base:x} section={section.name} range=0x{start_rva:x}..0x{end_rva:x}")
|
||||
for insn in disassembler.disasm(code, image_base + start_rva):
|
||||
rva = insn.address - image_base
|
||||
annotation = ""
|
||||
if insn.group(1) or insn.group(2): # jump/call
|
||||
for operand in insn.operands:
|
||||
if operand.type == X86_OP_IMM:
|
||||
annotation = f" ; target_rva=0x{operand.imm - image_base:x}"
|
||||
break
|
||||
elif insn.operands:
|
||||
memory_annotations = []
|
||||
for operand in insn.operands:
|
||||
if operand.type == X86_OP_MEM and operand.mem.base == X86_REG_RIP:
|
||||
target_rva = (insn.address + insn.size + operand.mem.disp) - image_base
|
||||
memory_annotations.append(format_memory_annotation(data, sections, target_rva))
|
||||
if memory_annotations:
|
||||
annotation = " ; " + " ; ".join(memory_annotations)
|
||||
marker = ">>" if rva == args.rva else " "
|
||||
print(f"{marker} 0x{rva:08x}: {insn.mnemonic:<8} {insn.op_str}{annotation}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,17 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
if [[ $# -lt 1 ]]; then
|
||||
echo "usage: $0 <archive-file-or-dir> [output-list]" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
archive_path=$1
|
||||
output=${2:-traffic-resources.txt}
|
||||
|
||||
tools/cp77_toolbox.sh archive --list \
|
||||
--regex 'traffic.*persistent|persistent.*traffic|traffic.*lane|world.*traffic' \
|
||||
"${archive_path}" \
|
||||
> "${output}"
|
||||
|
||||
echo "wrote ${output}"
|
||||
@@ -1,77 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Extract a single compressed RED4 archive segment by resource hash."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
FILE_ENTRY_SIZE = 56
|
||||
FILE_SEGMENT_SIZE = 16
|
||||
|
||||
|
||||
def parse_hash(value: str) -> int:
|
||||
return int(value, 0)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("archive", type=Path)
|
||||
parser.add_argument("resource_hash", type=parse_hash)
|
||||
parser.add_argument("outpath", type=Path)
|
||||
args = parser.parse_args()
|
||||
|
||||
with args.archive.open("rb") as archive:
|
||||
header = archive.read(40)
|
||||
if len(header) != 40:
|
||||
raise SystemExit("archive header is too short")
|
||||
magic, _version, index_pos, _index_size, *_rest = struct.unpack("<IIQIQIQ", header)
|
||||
if magic != 1380009042:
|
||||
raise SystemExit(f"unexpected archive magic: {magic}")
|
||||
|
||||
archive.seek(index_pos + 16)
|
||||
file_count, segment_count, _dependency_count = struct.unpack("<III", archive.read(12))
|
||||
|
||||
found: tuple[int, int] | None = None
|
||||
for _ in range(file_count):
|
||||
entry = archive.read(FILE_ENTRY_SIZE)
|
||||
(
|
||||
key,
|
||||
_timestamp,
|
||||
_flags,
|
||||
segments_start,
|
||||
segments_end,
|
||||
_dependencies_start,
|
||||
_dependencies_end,
|
||||
_sha1,
|
||||
) = struct.unpack("<QqIIIII20s", entry)
|
||||
if key == args.resource_hash:
|
||||
if segments_end - segments_start != 1:
|
||||
raise SystemExit(
|
||||
f"resource has {segments_end - segments_start} segments; expected 1"
|
||||
)
|
||||
found = (segments_start, segments_end)
|
||||
|
||||
if found is None:
|
||||
raise SystemExit(f"resource hash {args.resource_hash} not found")
|
||||
|
||||
segments = []
|
||||
for _ in range(segment_count):
|
||||
segments.append(struct.unpack("<QII", archive.read(FILE_SEGMENT_SIZE)))
|
||||
|
||||
offset, z_size, size = segments[found[0]]
|
||||
archive.seek(offset)
|
||||
data = archive.read(z_size)
|
||||
if len(data) != z_size:
|
||||
raise SystemExit("archive ended before segment data was fully read")
|
||||
|
||||
args.outpath.parent.mkdir(parents=True, exist_ok=True)
|
||||
args.outpath.write_bytes(data)
|
||||
print(f"wrote {args.outpath} z_size={z_size} size={size}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,79 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Find simple direct x64 call/jump xrefs to code RVAs in a PE file."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
from find_pe_string_xrefs import Section, parse_pe
|
||||
|
||||
|
||||
def iter_rel32_xrefs(data: bytes, section: Section, image_base: int, target_vas: set[int]) -> dict[int, list[tuple[int, str]]]:
|
||||
hits: dict[int, list[tuple[int, str]]] = {target_va: [] for target_va in target_vas}
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
|
||||
opcodes = {
|
||||
0xE8: "call",
|
||||
0xE9: "jmp",
|
||||
}
|
||||
for offset in range(start, max(start, end - 5)):
|
||||
opcode = data[offset]
|
||||
mnemonic = opcodes.get(opcode)
|
||||
if mnemonic is None:
|
||||
continue
|
||||
|
||||
disp = struct.unpack_from("<i", data, offset + 1)[0]
|
||||
instr_rva = section.virtual_address + (offset - section.raw_offset)
|
||||
target_va = image_base + instr_rva + 5 + disp
|
||||
if target_va in target_vas:
|
||||
hits[target_va].append((instr_rva, mnemonic))
|
||||
|
||||
# Common tail-call form: ff 25 <riprel32> imports/thunks are not resolved
|
||||
# here. This helper intentionally stays small and deterministic.
|
||||
return hits
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("pe", type=Path)
|
||||
parser.add_argument("targets", nargs="+", help="name=rva_hex pairs")
|
||||
args = parser.parse_args()
|
||||
|
||||
data = args.pe.read_bytes()
|
||||
image_base, sections = parse_pe(data)
|
||||
code_sections = [section for section in sections if section.is_code]
|
||||
|
||||
targets: list[tuple[str, int, int]] = []
|
||||
target_vas: set[int] = set()
|
||||
for item in args.targets:
|
||||
name, _, rva_text = item.partition("=")
|
||||
if not name or not rva_text:
|
||||
raise ValueError(f"target must be name=rva_hex: {item}")
|
||||
target_rva = int(rva_text, 16)
|
||||
target_va = image_base + target_rva
|
||||
targets.append((name, target_rva, target_va))
|
||||
target_vas.add(target_va)
|
||||
|
||||
xrefs: dict[int, list[tuple[int, str]]] = {target_va: [] for target_va in target_vas}
|
||||
for section in code_sections:
|
||||
section_hits = iter_rel32_xrefs(data, section, image_base, target_vas)
|
||||
for target_va, hits in section_hits.items():
|
||||
xrefs[target_va].extend(hits)
|
||||
|
||||
print(f"image_base=0x{image_base:x}")
|
||||
for name, target_rva, target_va in targets:
|
||||
all_hits = sorted(xrefs[target_va])
|
||||
print(f"{name}: target_rva=0x{target_rva:x} direct_xrefs={len(all_hits)}")
|
||||
for instr_rva, mnemonic in all_hits[:64]:
|
||||
print(f" {mnemonic}_rva=0x{instr_rva:x}")
|
||||
if len(all_hits) > 64:
|
||||
print(f" ... {len(all_hits) - 64} more")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,136 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Find likely x64 RIP-relative references to known string RVAs in a PE file."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import dataclasses
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Section:
|
||||
name: str
|
||||
virtual_address: int
|
||||
virtual_size: int
|
||||
raw_offset: int
|
||||
raw_size: int
|
||||
characteristics: int
|
||||
|
||||
@property
|
||||
def is_code(self) -> bool:
|
||||
image_scn_cnt_code = 0x00000020
|
||||
image_scn_mem_execute = 0x20000000
|
||||
return bool(self.characteristics & image_scn_cnt_code) and bool(self.characteristics & image_scn_mem_execute)
|
||||
|
||||
def contains_rva(self, rva: int) -> bool:
|
||||
size = max(self.virtual_size, self.raw_size)
|
||||
return self.virtual_address <= rva < self.virtual_address + size
|
||||
|
||||
def rva_to_offset(self, rva: int) -> int:
|
||||
return self.raw_offset + (rva - self.virtual_address)
|
||||
|
||||
|
||||
def parse_pe(data: bytes) -> tuple[int, list[Section]]:
|
||||
if data[:2] != b"MZ":
|
||||
raise ValueError("not an MZ executable")
|
||||
|
||||
pe_offset = struct.unpack_from("<I", data, 0x3C)[0]
|
||||
if data[pe_offset : pe_offset + 4] != b"PE\0\0":
|
||||
raise ValueError("not a PE executable")
|
||||
|
||||
coff_offset = pe_offset + 4
|
||||
section_count = struct.unpack_from("<H", data, coff_offset + 2)[0]
|
||||
optional_size = struct.unpack_from("<H", data, coff_offset + 16)[0]
|
||||
optional_offset = coff_offset + 20
|
||||
magic = struct.unpack_from("<H", data, optional_offset)[0]
|
||||
if magic != 0x20B:
|
||||
raise ValueError("expected PE32+ executable")
|
||||
|
||||
image_base = struct.unpack_from("<Q", data, optional_offset + 24)[0]
|
||||
section_offset = optional_offset + optional_size
|
||||
|
||||
sections: list[Section] = []
|
||||
for index in range(section_count):
|
||||
offset = section_offset + index * 40
|
||||
name = data[offset : offset + 8].split(b"\0", 1)[0].decode("ascii", "replace")
|
||||
virtual_size, virtual_address, raw_size, raw_offset = struct.unpack_from("<IIII", data, offset + 8)
|
||||
characteristics = struct.unpack_from("<I", data, offset + 36)[0]
|
||||
sections.append(Section(name, virtual_address, virtual_size, raw_offset, raw_size, characteristics))
|
||||
|
||||
return image_base, sections
|
||||
|
||||
|
||||
def section_for_rva(sections: list[Section], rva: int) -> Section | None:
|
||||
for section in sections:
|
||||
if section.contains_rva(rva):
|
||||
return section
|
||||
return None
|
||||
|
||||
|
||||
def iter_rip_xrefs(data: bytes, section: Section, image_base: int, target_vas: set[int]) -> dict[int, list[int]]:
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
hits: dict[int, list[int]] = {target_va: [] for target_va in target_vas}
|
||||
|
||||
# On x64, most string references are encoded as a signed 32-bit displacement
|
||||
# relative to the address immediately after the displacement.
|
||||
for offset in range(start, max(start, end - 4)):
|
||||
disp = struct.unpack_from("<i", data, offset)[0]
|
||||
instr_end_va = image_base + section.virtual_address + (offset - section.raw_offset) + 4
|
||||
target_va = instr_end_va + disp
|
||||
if target_va in target_vas:
|
||||
hits[target_va].append(section.virtual_address + (offset - section.raw_offset))
|
||||
|
||||
return hits
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("pe", type=Path)
|
||||
parser.add_argument("targets", nargs="+", help="name=rva_hex pairs")
|
||||
args = parser.parse_args()
|
||||
|
||||
data = args.pe.read_bytes()
|
||||
image_base, sections = parse_pe(data)
|
||||
code_sections = [section for section in sections if section.is_code]
|
||||
|
||||
targets: list[tuple[str, int, int]] = []
|
||||
target_vas: set[int] = set()
|
||||
for item in args.targets:
|
||||
name, _, rva_text = item.partition("=")
|
||||
if not name or not rva_text:
|
||||
raise ValueError(f"target must be name=rva_hex: {item}")
|
||||
target_rva = int(rva_text, 16)
|
||||
target_va = image_base + target_rva
|
||||
targets.append((name, target_rva, target_va))
|
||||
target_vas.add(target_va)
|
||||
|
||||
xrefs: dict[int, list[int]] = {target_va: [] for target_va in target_vas}
|
||||
for section in code_sections:
|
||||
section_hits = iter_rip_xrefs(data, section, image_base, target_vas)
|
||||
for target_va, hits in section_hits.items():
|
||||
xrefs[target_va].extend(hits)
|
||||
|
||||
print(f"image_base=0x{image_base:x}")
|
||||
for name, target_rva, target_va in targets:
|
||||
target_section = section_for_rva(sections, target_rva)
|
||||
section_name = target_section.name if target_section else "<none>"
|
||||
print(f"{name}: target_rva=0x{target_rva:x} target_section={section_name}")
|
||||
|
||||
all_hits = xrefs[target_va]
|
||||
if not all_hits:
|
||||
print(" no code xrefs found")
|
||||
continue
|
||||
|
||||
for hit in all_hits[:32]:
|
||||
print(f" code_xref_disp_rva=0x{hit:x} probable_instr_rva=0x{max(0, hit - 7):x}..0x{hit:x}")
|
||||
if len(all_hits) > 32:
|
||||
print(f" ... {len(all_hits) - 32} more")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,168 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Find PE code blocks containing selected immediate byte values.
|
||||
|
||||
This is a lightweight triage helper, not a real disassembler. It groups .text
|
||||
bytes into contiguous blocks split by int3 padding and reports blocks whose raw
|
||||
bytes contain values of interest. It is useful when looking for native code that
|
||||
touches compact flags such as worldTrafficLanePersistentFlags.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import dataclasses
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from find_pe_string_xrefs import Section, parse_pe
|
||||
|
||||
|
||||
DEFAULT_PATTERNS = {
|
||||
"flag_road_u16": (0x0010, 2),
|
||||
"flag_intersection_u16": (0x0020, 2),
|
||||
"flag_traffic_disabled_u16": (0x0080, 2),
|
||||
"flag_gpsonly_u16": (0x0200, 2),
|
||||
"flag_no_ai_driving_u16": (0x2000, 2),
|
||||
"flag_highway_u16": (0x4000, 2),
|
||||
"flag_no_autodrive_u16": (0x8000, 2),
|
||||
"flag_road_u32": (0x0010, 4),
|
||||
"flag_intersection_u32": (0x0020, 4),
|
||||
"flag_traffic_disabled_u32": (0x0080, 4),
|
||||
"flag_gpsonly_u32": (0x0200, 4),
|
||||
"flag_no_ai_driving_u32": (0x2000, 4),
|
||||
"flag_highway_u32": (0x4000, 4),
|
||||
"flag_no_autodrive_u32": (0x8000, 4),
|
||||
"lane_size_u32": (0x00A0, 4),
|
||||
"traffic_data_size_u32": (0x0110, 4),
|
||||
}
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Pattern:
|
||||
name: str
|
||||
needle: bytes
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Hit:
|
||||
rva: int
|
||||
pattern: str
|
||||
|
||||
|
||||
def parse_pattern(text: str) -> Pattern:
|
||||
name, sep, spec = text.partition("=")
|
||||
if not sep or not name:
|
||||
raise argparse.ArgumentTypeError("patterns must be name=hex[:size]")
|
||||
|
||||
value_text, _, size_text = spec.partition(":")
|
||||
value = int(value_text, 0)
|
||||
size = int(size_text, 0) if size_text else 4
|
||||
if size not in (1, 2, 4, 8):
|
||||
raise argparse.ArgumentTypeError("pattern size must be 1, 2, 4, or 8")
|
||||
if value < 0 or value >= (1 << (size * 8)):
|
||||
raise argparse.ArgumentTypeError("pattern value does not fit size")
|
||||
|
||||
return Pattern(name=name, needle=value.to_bytes(size, "little"))
|
||||
|
||||
|
||||
def default_patterns() -> list[Pattern]:
|
||||
return [Pattern(name, value.to_bytes(size, "little")) for name, (value, size) in DEFAULT_PATTERNS.items()]
|
||||
|
||||
|
||||
def iter_code_blocks(data: bytes, section: Section, min_size: int) -> list[tuple[int, bytes]]:
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
blob = data[start:end]
|
||||
blocks: list[tuple[int, bytes]] = []
|
||||
block_start = 0
|
||||
|
||||
for match in re.finditer(rb"\xcc{2,}", blob):
|
||||
if match.start() - block_start >= min_size:
|
||||
blocks.append((section.virtual_address + block_start, blob[block_start : match.start()]))
|
||||
block_start = match.end()
|
||||
|
||||
if len(blob) - block_start >= min_size:
|
||||
blocks.append((section.virtual_address + block_start, blob[block_start:]))
|
||||
|
||||
return blocks
|
||||
|
||||
|
||||
def find_hits(block_rva: int, block: bytes, patterns: list[Pattern], max_offsets: int) -> tuple[dict[str, int], list[Hit]]:
|
||||
counts: dict[str, int] = {}
|
||||
hits: list[Hit] = []
|
||||
for pattern in patterns:
|
||||
offset = block.find(pattern.needle)
|
||||
while offset != -1:
|
||||
counts[pattern.name] = counts.get(pattern.name, 0) + 1
|
||||
if len([hit for hit in hits if hit.pattern == pattern.name]) < max_offsets:
|
||||
hits.append(Hit(block_rva + offset, pattern.name))
|
||||
offset = block.find(pattern.needle, offset + 1)
|
||||
return counts, hits
|
||||
|
||||
|
||||
def score_counts(counts: dict[str, int]) -> int:
|
||||
score = 0
|
||||
for name, count in counts.items():
|
||||
weight = 1
|
||||
if name.endswith("_u32"):
|
||||
weight = 2
|
||||
if name in {"flag_highway_u16", "flag_highway_u32", "flag_gpsonly_u16", "flag_gpsonly_u32"}:
|
||||
weight += 2
|
||||
if name in {"lane_size_u32", "traffic_data_size_u32"}:
|
||||
weight += 3
|
||||
score += min(count, 8) * weight
|
||||
|
||||
names = set(counts)
|
||||
if names & {"flag_highway_u16", "flag_highway_u32"} and names & {"flag_gpsonly_u16", "flag_gpsonly_u32"}:
|
||||
score += 8
|
||||
if names & {"flag_road_u16", "flag_road_u32"} and names & {"flag_highway_u16", "flag_highway_u32"}:
|
||||
score += 4
|
||||
return score
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("pe", type=Path)
|
||||
parser.add_argument("--pattern", action="append", type=parse_pattern, default=[])
|
||||
parser.add_argument("--no-defaults", action="store_true")
|
||||
parser.add_argument("--require", action="append", default=[], help="Require a pattern name; repeatable")
|
||||
parser.add_argument("--min-score", type=int, default=10)
|
||||
parser.add_argument("--min-block-size", type=int, default=16)
|
||||
parser.add_argument("--max-results", type=int, default=80)
|
||||
parser.add_argument("--max-offsets", type=int, default=4)
|
||||
args = parser.parse_args()
|
||||
|
||||
patterns = ([] if args.no_defaults else default_patterns()) + args.pattern
|
||||
data = args.pe.read_bytes()
|
||||
image_base, sections = parse_pe(data)
|
||||
|
||||
results: list[tuple[int, int, int, dict[str, int], list[Hit]]] = []
|
||||
for section in sections:
|
||||
if not section.is_code:
|
||||
continue
|
||||
for block_rva, block in iter_code_blocks(data, section, args.min_block_size):
|
||||
counts, hits = find_hits(block_rva, block, patterns, args.max_offsets)
|
||||
if not counts:
|
||||
continue
|
||||
if any(required not in counts for required in args.require):
|
||||
continue
|
||||
score = score_counts(counts)
|
||||
if score < args.min_score:
|
||||
continue
|
||||
results.append((score, block_rva, block_rva + len(block), counts, hits))
|
||||
|
||||
results.sort(key=lambda item: (-item[0], item[1]))
|
||||
print(f"image_base=0x{image_base:x}")
|
||||
print(f"blocks={len(results)} min_score={args.min_score} require={args.require}")
|
||||
for score, start_rva, end_rva, counts, hits in results[: args.max_results]:
|
||||
count_text = ", ".join(f"{name}:{counts[name]}" for name in sorted(counts))
|
||||
print(f"block rva=0x{start_rva:x}..0x{end_rva:x} size=0x{end_rva - start_rva:x} score={score}")
|
||||
print(f" counts {count_text}")
|
||||
for hit in sorted(hits, key=lambda item: item.rva):
|
||||
print(f" hit rva=0x{hit.rva:x} pattern={hit.pattern}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,174 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Map REDscript/native registration strings to nearby code pointer candidates."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
from find_pe_string_xrefs import Section, parse_pe
|
||||
|
||||
|
||||
def section_for_rva(sections: list[Section], rva: int) -> Section | None:
|
||||
for section in sections:
|
||||
if section.contains_rva(rva):
|
||||
return section
|
||||
return None
|
||||
|
||||
|
||||
def is_readable_data(section: Section) -> bool:
|
||||
return bool(section.characteristics & 0x40000000) and not section.is_code
|
||||
|
||||
|
||||
def iter_ascii_strings(data: bytes, section: Section, min_len: int) -> list[tuple[int, str]]:
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
pattern = re.compile(rb"[\x20-\x7e]{%d,}" % min_len)
|
||||
return [
|
||||
(section.virtual_address + match.start(), match.group(0).decode("ascii", "replace"))
|
||||
for match in pattern.finditer(data[start:end])
|
||||
]
|
||||
|
||||
|
||||
def read_i32(data: bytes, offset: int) -> int | None:
|
||||
if offset < 0 or offset + 4 > len(data):
|
||||
return None
|
||||
return struct.unpack_from("<i", data, offset)[0]
|
||||
|
||||
|
||||
def find_rip_disp_xrefs(data: bytes, sections: list[Section], image_base: int, target_rvas: set[int]) -> dict[int, list[int]]:
|
||||
target_vas = {image_base + rva: rva for rva in target_rvas}
|
||||
hits = {rva: [] for rva in target_rvas}
|
||||
for section in sections:
|
||||
if not section.is_code:
|
||||
continue
|
||||
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
for offset in range(start, max(start, end - 4)):
|
||||
disp = read_i32(data, offset)
|
||||
if disp is None:
|
||||
continue
|
||||
instr_end_va = image_base + section.virtual_address + (offset - section.raw_offset) + 4
|
||||
target_rva = target_vas.get(instr_end_va + disp)
|
||||
if target_rva is not None:
|
||||
hits[target_rva].append(section.virtual_address + (offset - section.raw_offset))
|
||||
return hits
|
||||
|
||||
|
||||
def iter_nearby_rip_code_targets(
|
||||
data: bytes,
|
||||
sections: list[Section],
|
||||
image_base: int,
|
||||
code_section: Section,
|
||||
center_disp_rva: int,
|
||||
radius: int,
|
||||
) -> list[tuple[int, int, str]]:
|
||||
# Common x64 RIP-relative LEA/MOV forms used by the generated registration
|
||||
# thunks to pass function pointers and type descriptors around.
|
||||
patterns: tuple[tuple[bytes, int, str], ...] = (
|
||||
(b"\x48\x8d\x05", 3, "lea rax"),
|
||||
(b"\x48\x8d\x0d", 3, "lea rcx"),
|
||||
(b"\x48\x8d\x15", 3, "lea rdx"),
|
||||
(b"\x48\x8d\x1d", 3, "lea rbx"),
|
||||
(b"\x48\x8d\x35", 3, "lea rsi"),
|
||||
(b"\x48\x8d\x3d", 3, "lea rdi"),
|
||||
(b"\x4c\x8d\x05", 3, "lea r8"),
|
||||
(b"\x4c\x8d\x0d", 3, "lea r9"),
|
||||
(b"\x4c\x8d\x15", 3, "lea r10"),
|
||||
(b"\x4c\x8d\x1d", 3, "lea r11"),
|
||||
(b"\x4c\x8d\x25", 3, "lea r12"),
|
||||
(b"\x4c\x8d\x2d", 3, "lea r13"),
|
||||
(b"\x4c\x8d\x35", 3, "lea r14"),
|
||||
(b"\x4c\x8d\x3d", 3, "lea r15"),
|
||||
)
|
||||
|
||||
center_off = code_section.rva_to_offset(center_disp_rva)
|
||||
start = max(code_section.raw_offset, center_off - radius)
|
||||
end = min(len(data), code_section.raw_offset + code_section.raw_size, center_off + radius)
|
||||
out: list[tuple[int, int, str]] = []
|
||||
|
||||
for offset in range(start, end):
|
||||
for prefix, disp_start, label in patterns:
|
||||
if data[offset : offset + len(prefix)] != prefix:
|
||||
continue
|
||||
|
||||
disp = read_i32(data, offset + disp_start)
|
||||
if disp is None:
|
||||
continue
|
||||
|
||||
instr_rva = code_section.virtual_address + (offset - code_section.raw_offset)
|
||||
instr_end_va = image_base + instr_rva + disp_start + 4
|
||||
target_rva = instr_end_va + disp - image_base
|
||||
target_section = section_for_rva(sections, target_rva)
|
||||
if target_section and target_section.is_code:
|
||||
out.append((instr_rva, target_rva, label))
|
||||
|
||||
return sorted(set(out))
|
||||
|
||||
|
||||
def likely_registered_code(ref_rva: int, candidates: list[tuple[int, int, str]]) -> tuple[int, int, str] | None:
|
||||
for candidate in candidates:
|
||||
instr_rva, _target_rva, label = candidate
|
||||
if instr_rva > ref_rva and label == "lea rax":
|
||||
return candidate
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("pe", type=Path)
|
||||
parser.add_argument("--pattern", required=True, help="case-insensitive regex for registration strings")
|
||||
parser.add_argument("--min-len", type=int, default=5)
|
||||
parser.add_argument("--radius", type=int, default=192)
|
||||
parser.add_argument("--max-strings", type=int, default=200)
|
||||
args = parser.parse_args()
|
||||
|
||||
data = args.pe.read_bytes()
|
||||
image_base, sections = parse_pe(data)
|
||||
regex = re.compile(args.pattern, re.IGNORECASE)
|
||||
|
||||
strings: list[tuple[int, str]] = []
|
||||
for section in sections:
|
||||
if is_readable_data(section):
|
||||
strings.extend((rva, text) for rva, text in iter_ascii_strings(data, section, args.min_len) if regex.search(text))
|
||||
strings = sorted(strings, key=lambda item: (item[1].lower(), item[0]))
|
||||
|
||||
xrefs = find_rip_disp_xrefs(data, sections, image_base, {rva for rva, _ in strings})
|
||||
print(f"image_base=0x{image_base:x}")
|
||||
print(f"matches={len(strings)} pattern={args.pattern!r}")
|
||||
|
||||
emitted = 0
|
||||
for string_rva, text in strings:
|
||||
refs = xrefs.get(string_rva, [])
|
||||
if not refs:
|
||||
continue
|
||||
|
||||
print(f"name={text!r} string_rva=0x{string_rva:x} xrefs={len(refs)}")
|
||||
for ref in refs[:12]:
|
||||
code_section = section_for_rva(sections, ref)
|
||||
if not code_section:
|
||||
continue
|
||||
candidates = iter_nearby_rip_code_targets(data, sections, image_base, code_section, ref, args.radius)
|
||||
print(f" xref_disp_rva=0x{ref:x} nearby_code_ptrs={len(candidates)}")
|
||||
likely = likely_registered_code(ref, candidates)
|
||||
if likely:
|
||||
instr_rva, target_rva, label = likely
|
||||
print(f" likely_registered_code {label} instr_rva=0x{instr_rva:x} target_rva=0x{target_rva:x}")
|
||||
for instr_rva, target_rva, label in candidates[:16]:
|
||||
print(f" {label} instr_rva=0x{instr_rva:x} target_rva=0x{target_rva:x}")
|
||||
if len(candidates) > 16:
|
||||
print(f" ... +{len(candidates) - 16}")
|
||||
|
||||
emitted += 1
|
||||
if emitted >= args.max_strings:
|
||||
break
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,169 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Find static GPS/map route candidates in a Cyberpunk 2077 executable."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import dataclasses
|
||||
import json
|
||||
import re
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@dataclasses.dataclass(frozen=True)
|
||||
class Section:
|
||||
name: str
|
||||
virtual_address: int
|
||||
virtual_size: int
|
||||
raw_offset: int
|
||||
raw_size: int
|
||||
characteristics: int
|
||||
|
||||
@property
|
||||
def is_code(self) -> bool:
|
||||
return bool(self.characteristics & 0x20) and bool(self.characteristics & 0x20000000)
|
||||
|
||||
@property
|
||||
def is_readable_data(self) -> bool:
|
||||
return bool(self.characteristics & 0x40000000) and not self.is_code
|
||||
|
||||
def contains_rva(self, rva: int) -> bool:
|
||||
return self.virtual_address <= rva < self.virtual_address + max(self.virtual_size, self.raw_size)
|
||||
|
||||
def rva_to_offset(self, rva: int) -> int:
|
||||
return self.raw_offset + (rva - self.virtual_address)
|
||||
|
||||
|
||||
def parse_pe(data: bytes) -> tuple[int, list[Section]]:
|
||||
if data[:2] != b"MZ":
|
||||
raise ValueError("not an MZ executable")
|
||||
|
||||
pe_offset = struct.unpack_from("<I", data, 0x3C)[0]
|
||||
if data[pe_offset : pe_offset + 4] != b"PE\0\0":
|
||||
raise ValueError("not a PE executable")
|
||||
|
||||
coff_offset = pe_offset + 4
|
||||
section_count = struct.unpack_from("<H", data, coff_offset + 2)[0]
|
||||
optional_size = struct.unpack_from("<H", data, coff_offset + 16)[0]
|
||||
optional_offset = coff_offset + 20
|
||||
if struct.unpack_from("<H", data, optional_offset)[0] != 0x20B:
|
||||
raise ValueError("expected PE32+ executable")
|
||||
|
||||
image_base = struct.unpack_from("<Q", data, optional_offset + 24)[0]
|
||||
section_offset = optional_offset + optional_size
|
||||
sections: list[Section] = []
|
||||
for index in range(section_count):
|
||||
offset = section_offset + index * 40
|
||||
name = data[offset : offset + 8].split(b"\0", 1)[0].decode("ascii", "replace")
|
||||
virtual_size, virtual_address, raw_size, raw_offset = struct.unpack_from("<IIII", data, offset + 8)
|
||||
characteristics = struct.unpack_from("<I", data, offset + 36)[0]
|
||||
sections.append(Section(name, virtual_address, virtual_size, raw_offset, raw_size, characteristics))
|
||||
|
||||
return image_base, sections
|
||||
|
||||
|
||||
def iter_strings(data: bytes, section: Section, min_len: int) -> list[tuple[int, str, str]]:
|
||||
results: list[tuple[int, str, str]] = []
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
blob = data[start:end]
|
||||
|
||||
ascii_re = re.compile(rb"[\x20-\x7e]{%d,}" % min_len)
|
||||
for match in ascii_re.finditer(blob):
|
||||
text = match.group(0).decode("ascii", "replace")
|
||||
results.append((section.virtual_address + match.start(), "ascii", text))
|
||||
|
||||
utf16_re = re.compile((rb"(?:[\x20-\x7e]\x00){%d,}" % min_len))
|
||||
for match in utf16_re.finditer(blob):
|
||||
raw = match.group(0)
|
||||
text = raw.decode("utf-16le", "replace")
|
||||
results.append((section.virtual_address + match.start(), "utf16", text))
|
||||
|
||||
return results
|
||||
|
||||
|
||||
def find_xrefs(data: bytes, sections: list[Section], image_base: int, target_rvas: set[int]) -> dict[int, list[int]]:
|
||||
target_vas = {image_base + rva: rva for rva in target_rvas}
|
||||
hits = {rva: [] for rva in target_rvas}
|
||||
for section in sections:
|
||||
if not section.is_code:
|
||||
continue
|
||||
|
||||
start = section.raw_offset
|
||||
end = min(len(data), section.raw_offset + section.raw_size)
|
||||
for offset in range(start, max(start, end - 4)):
|
||||
disp = struct.unpack_from("<i", data, offset)[0]
|
||||
instr_end_va = image_base + section.virtual_address + (offset - section.raw_offset) + 4
|
||||
target_va = instr_end_va + disp
|
||||
target_rva = target_vas.get(target_va)
|
||||
if target_rva is not None:
|
||||
hits[target_rva].append(section.virtual_address + (offset - section.raw_offset))
|
||||
return hits
|
||||
|
||||
|
||||
def load_address_symbols(path: Path, pattern: re.Pattern[str]) -> list[tuple[int, str]]:
|
||||
payload = json.loads(path.read_text())
|
||||
symbols: list[tuple[int, str]] = []
|
||||
for item in payload.get("Addresses", []):
|
||||
symbol = item.get("symbol")
|
||||
offset = item.get("offset")
|
||||
if not symbol or not offset or not pattern.search(symbol):
|
||||
continue
|
||||
|
||||
section, _, value = offset.partition(":")
|
||||
if section != "0001":
|
||||
continue
|
||||
|
||||
symbols.append((int(value, 16), symbol))
|
||||
return sorted(symbols)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("exe", type=Path)
|
||||
parser.add_argument("--addresses", type=Path)
|
||||
parser.add_argument(
|
||||
"--pattern",
|
||||
default=r"gps|route|path|mappin|track|quest|journal|navigation|map",
|
||||
help="case-insensitive regex for string and symbol filtering",
|
||||
)
|
||||
parser.add_argument("--min-len", type=int, default=5)
|
||||
parser.add_argument("--max-strings", type=int, default=300)
|
||||
args = parser.parse_args()
|
||||
|
||||
data = args.exe.read_bytes()
|
||||
image_base, sections = parse_pe(data)
|
||||
pattern = re.compile(args.pattern, re.IGNORECASE)
|
||||
|
||||
matches: list[tuple[int, str, str]] = []
|
||||
for section in sections:
|
||||
if section.is_readable_data:
|
||||
matches.extend(item for item in iter_strings(data, section, args.min_len) if pattern.search(item[2]))
|
||||
|
||||
matches.sort(key=lambda item: (item[0], item[1], item[2]))
|
||||
target_rvas = {rva for rva, _, _ in matches}
|
||||
xrefs = find_xrefs(data, sections, image_base, target_rvas)
|
||||
|
||||
print(f"image_base=0x{image_base:x}")
|
||||
print(f"string_matches={len(matches)} pattern={args.pattern!r}")
|
||||
for rva, encoding, text in matches[: args.max_strings]:
|
||||
hit_text = ", ".join(f"0x{hit:x}" for hit in xrefs.get(rva, [])[:12])
|
||||
if len(xrefs.get(rva, [])) > 12:
|
||||
hit_text += f", ... +{len(xrefs[rva]) - 12}"
|
||||
print(f"string rva=0x{rva:x} enc={encoding} xrefs={len(xrefs.get(rva, []))} text={text!r}")
|
||||
if hit_text:
|
||||
print(f" xref_disp_rvas={hit_text}")
|
||||
|
||||
if args.addresses:
|
||||
print()
|
||||
symbols = load_address_symbols(args.addresses, pattern)
|
||||
print(f"address_symbols={len(symbols)}")
|
||||
for rva, symbol in symbols[: args.max_strings]:
|
||||
print(f"symbol rva=0x{rva:x} name={symbol}")
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,327 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Generate a compact road-class grid for the RED4ext GPS edge-cost shim."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable
|
||||
|
||||
|
||||
FLAGS = {
|
||||
"Pavement": 0x0008,
|
||||
"Road": 0x0010,
|
||||
"GPSOnly": 0x0200,
|
||||
"Highway": 0x4000,
|
||||
}
|
||||
|
||||
CATEGORY_CODES = {
|
||||
"unknown": ".",
|
||||
"pavement": "P",
|
||||
"gpsonly": "G",
|
||||
"road": "R",
|
||||
"highway": "H",
|
||||
}
|
||||
|
||||
CATEGORY_PRIORITY = {
|
||||
".": 0,
|
||||
"P": 1,
|
||||
"G": 2,
|
||||
"R": 3,
|
||||
"H": 4,
|
||||
}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class GridSpec:
|
||||
min_x: float
|
||||
min_y: float
|
||||
width: int
|
||||
height: int
|
||||
cell_size: float
|
||||
|
||||
|
||||
def flag_value(value: Any) -> int:
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
return int(value)
|
||||
if isinstance(value, dict):
|
||||
for key in ("Value", "$value", "value"):
|
||||
if key in value:
|
||||
return flag_value(value[key])
|
||||
return int(value or 0)
|
||||
|
||||
|
||||
def lane_category(flags: int) -> str:
|
||||
if flags & FLAGS["Highway"]:
|
||||
return "highway"
|
||||
if flags & FLAGS["GPSOnly"]:
|
||||
return "gpsonly"
|
||||
if flags & FLAGS["Road"]:
|
||||
return "road"
|
||||
if flags & FLAGS["Pavement"]:
|
||||
return "pavement"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def load_lanes(path: Path) -> list[dict[str, Any]]:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
return data["Data"]["RootChunk"]["data"]["lanes"]
|
||||
|
||||
|
||||
def load_lane_polygons(path: Path) -> list[list[tuple[float, float]]]:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
rows = data["Data"]["RootChunk"]["data"]
|
||||
polygons: list[list[tuple[float, float]]] = []
|
||||
for row in rows:
|
||||
polygon = row.get("value", {}).get("polygon") or []
|
||||
polygons.append([(float(point["X"]), float(point["Y"])) for point in polygon])
|
||||
return polygons
|
||||
|
||||
|
||||
def point_in_polygon(x: float, y: float, polygon: list[tuple[float, float]]) -> bool:
|
||||
inside = False
|
||||
count = len(polygon)
|
||||
if count < 3:
|
||||
return False
|
||||
|
||||
previous_x, previous_y = polygon[-1]
|
||||
for current_x, current_y in polygon:
|
||||
if (current_y > y) != (previous_y > y):
|
||||
edge_x = (previous_x - current_x) * (y - current_y) / (previous_y - current_y) + current_x
|
||||
if x < edge_x:
|
||||
inside = not inside
|
||||
previous_x, previous_y = current_x, current_y
|
||||
|
||||
return inside
|
||||
|
||||
|
||||
def orientation(ax: float, ay: float, bx: float, by: float, cx: float, cy: float) -> float:
|
||||
return (by - ay) * (cx - bx) - (bx - ax) * (cy - by)
|
||||
|
||||
|
||||
def on_segment(ax: float, ay: float, bx: float, by: float, cx: float, cy: float) -> bool:
|
||||
return min(ax, cx) <= bx <= max(ax, cx) and min(ay, cy) <= by <= max(ay, cy)
|
||||
|
||||
|
||||
def segments_intersect(
|
||||
ax: float,
|
||||
ay: float,
|
||||
bx: float,
|
||||
by: float,
|
||||
cx: float,
|
||||
cy: float,
|
||||
dx: float,
|
||||
dy: float,
|
||||
) -> bool:
|
||||
o1 = orientation(ax, ay, bx, by, cx, cy)
|
||||
o2 = orientation(ax, ay, bx, by, dx, dy)
|
||||
o3 = orientation(cx, cy, dx, dy, ax, ay)
|
||||
o4 = orientation(cx, cy, dx, dy, bx, by)
|
||||
|
||||
if (o1 > 0) != (o2 > 0) and (o3 > 0) != (o4 > 0):
|
||||
return True
|
||||
|
||||
eps = 1e-5
|
||||
if abs(o1) <= eps and on_segment(ax, ay, cx, cy, bx, by):
|
||||
return True
|
||||
if abs(o2) <= eps and on_segment(ax, ay, dx, dy, bx, by):
|
||||
return True
|
||||
if abs(o3) <= eps and on_segment(cx, cy, ax, ay, dx, dy):
|
||||
return True
|
||||
if abs(o4) <= eps and on_segment(cx, cy, bx, by, dx, dy):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def polygon_intersects_cell(
|
||||
polygon: list[tuple[float, float]],
|
||||
min_x: float,
|
||||
min_y: float,
|
||||
max_x: float,
|
||||
max_y: float,
|
||||
) -> bool:
|
||||
center_x = (min_x + max_x) * 0.5
|
||||
center_y = (min_y + max_y) * 0.5
|
||||
if point_in_polygon(center_x, center_y, polygon):
|
||||
return True
|
||||
|
||||
corners = ((min_x, min_y), (max_x, min_y), (max_x, max_y), (min_x, max_y))
|
||||
if any(point_in_polygon(x, y, polygon) for x, y in corners):
|
||||
return True
|
||||
|
||||
if any(min_x <= x <= max_x and min_y <= y <= max_y for x, y in polygon):
|
||||
return True
|
||||
|
||||
cell_edges = (
|
||||
(min_x, min_y, max_x, min_y),
|
||||
(max_x, min_y, max_x, max_y),
|
||||
(max_x, max_y, min_x, max_y),
|
||||
(min_x, max_y, min_x, min_y),
|
||||
)
|
||||
previous_x, previous_y = polygon[-1]
|
||||
for current_x, current_y in polygon:
|
||||
for edge in cell_edges:
|
||||
if segments_intersect(previous_x, previous_y, current_x, current_y, *edge):
|
||||
return True
|
||||
previous_x, previous_y = current_x, current_y
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def grid_index(spec: GridSpec, x: float, y: float) -> tuple[int, int]:
|
||||
column = math.floor((x - spec.min_x) / spec.cell_size)
|
||||
row = math.floor((y - spec.min_y) / spec.cell_size)
|
||||
return int(column), int(row)
|
||||
|
||||
|
||||
def mark_cell(grid: list[list[str]], column: int, row: int, code: str) -> bool:
|
||||
if row < 0 or row >= len(grid) or column < 0 or column >= len(grid[row]):
|
||||
return False
|
||||
if CATEGORY_PRIORITY[code] <= CATEGORY_PRIORITY[grid[row][column]]:
|
||||
return False
|
||||
grid[row][column] = code
|
||||
return True
|
||||
|
||||
|
||||
def iter_cells_for_polygon(spec: GridSpec, polygon: list[tuple[float, float]]) -> Iterable[tuple[int, int]]:
|
||||
xs = [point[0] for point in polygon]
|
||||
ys = [point[1] for point in polygon]
|
||||
min_col, min_row = grid_index(spec, min(xs), min(ys))
|
||||
max_col, max_row = grid_index(spec, max(xs), max(ys))
|
||||
|
||||
for row in range(max(0, min_row), min(spec.height - 1, max_row) + 1):
|
||||
for column in range(max(0, min_col), min(spec.width - 1, max_col) + 1):
|
||||
cell_min_x = spec.min_x + column * spec.cell_size
|
||||
cell_min_y = spec.min_y + row * spec.cell_size
|
||||
if polygon_intersects_cell(
|
||||
polygon,
|
||||
cell_min_x,
|
||||
cell_min_y,
|
||||
cell_min_x + spec.cell_size,
|
||||
cell_min_y + spec.cell_size,
|
||||
):
|
||||
yield column, row
|
||||
|
||||
|
||||
def build_grid(
|
||||
lanes: list[dict[str, Any]],
|
||||
polygons: list[list[tuple[float, float]]],
|
||||
cell_size: float,
|
||||
inflate_cells: int,
|
||||
) -> tuple[GridSpec, list[list[str]], dict[str, int]]:
|
||||
points = [point for polygon in polygons for point in polygon]
|
||||
if not points:
|
||||
raise ValueError("no polygon points found")
|
||||
|
||||
min_x = math.floor(min(point[0] for point in points) / cell_size) * cell_size
|
||||
min_y = math.floor(min(point[1] for point in points) / cell_size) * cell_size
|
||||
max_x = math.ceil(max(point[0] for point in points) / cell_size) * cell_size
|
||||
max_y = math.ceil(max(point[1] for point in points) / cell_size) * cell_size
|
||||
spec = GridSpec(
|
||||
min_x=min_x,
|
||||
min_y=min_y,
|
||||
width=int(round((max_x - min_x) / cell_size)),
|
||||
height=int(round((max_y - min_y) / cell_size)),
|
||||
cell_size=cell_size,
|
||||
)
|
||||
|
||||
grid = [[CATEGORY_CODES["unknown"] for _ in range(spec.width)] for _ in range(spec.height)]
|
||||
for index, (lane, polygon) in enumerate(zip(lanes, polygons, strict=True)):
|
||||
if len(polygon) < 3:
|
||||
continue
|
||||
code = CATEGORY_CODES[lane_category(flag_value(lane.get("flags", 0)))]
|
||||
if code == CATEGORY_CODES["unknown"]:
|
||||
continue
|
||||
|
||||
changed_cells: set[tuple[int, int]] = set()
|
||||
for column, row in iter_cells_for_polygon(spec, polygon):
|
||||
for dy in range(-inflate_cells, inflate_cells + 1):
|
||||
for dx in range(-inflate_cells, inflate_cells + 1):
|
||||
changed_cells.add((column + dx, row + dy))
|
||||
|
||||
for column, row in changed_cells:
|
||||
mark_cell(grid, column, row, code)
|
||||
|
||||
counts = {code: sum(row.count(code) for row in grid) for code in CATEGORY_PRIORITY}
|
||||
return spec, grid, counts
|
||||
|
||||
|
||||
def write_header(path: Path, spec: GridSpec, grid: list[list[str]], counts: dict[str, int]) -> None:
|
||||
lines = [
|
||||
"#pragma once",
|
||||
"",
|
||||
"#include <array>",
|
||||
"#include <cstdint>",
|
||||
"#include <string_view>",
|
||||
"",
|
||||
"namespace EdgeWeightGPS::Generated",
|
||||
"{",
|
||||
f"constexpr float kSpatialRoadGridMinX = {spec.min_x:.1f}f;",
|
||||
f"constexpr float kSpatialRoadGridMinY = {spec.min_y:.1f}f;",
|
||||
f"constexpr float kSpatialRoadGridCellSize = {spec.cell_size:.1f}f;",
|
||||
f"constexpr uint32_t kSpatialRoadGridWidth = {spec.width};",
|
||||
f"constexpr uint32_t kSpatialRoadGridHeight = {spec.height};",
|
||||
"",
|
||||
"// Cell codes: H=highway, R=road, G=GPS-only, P=pavement, .=unknown.",
|
||||
"// Generated by tools/generate_spatial_edge_grid.py from all.traffic_persistent and all.lane_polygons.",
|
||||
f"// Counts: H={counts['H']} R={counts['R']} G={counts['G']} P={counts['P']} unknown={counts['.']}.",
|
||||
"constexpr std::array<std::string_view, kSpatialRoadGridHeight> kSpatialRoadGridRows = {{",
|
||||
]
|
||||
|
||||
for row in grid:
|
||||
lines.append(f' "{("").join(row)}",')
|
||||
|
||||
lines.extend(
|
||||
[
|
||||
"}};",
|
||||
"",
|
||||
"} // namespace EdgeWeightGPS::Generated",
|
||||
"",
|
||||
]
|
||||
)
|
||||
path.write_text("\n".join(lines), encoding="utf-8")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--traffic-json", type=Path, default=Path("work/raw-segment-json/all.traffic_persistent.json"))
|
||||
parser.add_argument(
|
||||
"--lane-polygons-json",
|
||||
type=Path,
|
||||
default=Path("work/traffic-companions/json/all.lane_polygons.json"),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
type=Path,
|
||||
default=Path("work/generated/GeneratedSpatialRoadGrid.hpp"),
|
||||
)
|
||||
parser.add_argument("--cell-size", type=float, default=16.0)
|
||||
parser.add_argument("--inflate-cells", type=int, default=0)
|
||||
args = parser.parse_args()
|
||||
|
||||
lanes = load_lanes(args.traffic_json)
|
||||
polygons = load_lane_polygons(args.lane_polygons_json)
|
||||
if len(lanes) != len(polygons):
|
||||
raise ValueError(f"lane/polygon count mismatch: {len(lanes)} != {len(polygons)}")
|
||||
|
||||
spec, grid, counts = build_grid(lanes, polygons, args.cell_size, args.inflate_cells)
|
||||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||||
write_header(args.output, spec, grid, counts)
|
||||
|
||||
total = spec.width * spec.height
|
||||
covered = total - sum(row.count(".") for row in grid)
|
||||
print(
|
||||
f"wrote {args.output} cells={total} covered={covered} "
|
||||
f"size={spec.width}x{spec.height} origin=({spec.min_x},{spec.min_y}) cell={spec.cell_size}"
|
||||
)
|
||||
print("counts " + " ".join(f"{code}={counts[code]}" for code in ("H", "R", "G", "P", ".")))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,11 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
game_dir="${CYBERPUNK2077_GAME_DIR:-/var/home/salt/.var/app/com.valvesoftware.Steam/.local/share/Steam/steamapps/common/Cyberpunk 2077}"
|
||||
script_out="$game_dir/r6/scripts/EdgeWeightGPS"
|
||||
|
||||
mkdir -p "$script_out"
|
||||
install -m 0644 "$repo_root/redscript/EdgeWeightGPS/EdgeWeightGPS.reds" "$script_out/EdgeWeightGPS.reds"
|
||||
|
||||
printf 'Installed %s\n' "$script_out/EdgeWeightGPS.reds"
|
||||
@@ -1,132 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Patch traffic lane connection probabilities toward highway routing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
FLAGS = {
|
||||
"Road": 16,
|
||||
"Pavement": 8,
|
||||
"TrafficDisabled": 128,
|
||||
"CrossWalk": 256,
|
||||
"GPSOnly": 512,
|
||||
"Blockade": 2048,
|
||||
"Highway": 16384,
|
||||
}
|
||||
|
||||
|
||||
def load_json(path: Path) -> Any:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def flag_value(flags: Any) -> int:
|
||||
return int(flags or 0)
|
||||
|
||||
|
||||
def has(flags: int, name: str) -> bool:
|
||||
return bool(flags & FLAGS[name])
|
||||
|
||||
|
||||
def category(flags: int) -> str:
|
||||
if has(flags, "Highway"):
|
||||
return "highway"
|
||||
if has(flags, "GPSOnly"):
|
||||
return "gpsonly"
|
||||
if has(flags, "Road"):
|
||||
return "road"
|
||||
if has(flags, "Pavement"):
|
||||
return "pavement"
|
||||
return "other"
|
||||
|
||||
|
||||
def write_json(path: Path, data: Any) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("traffic_persistent_json", type=Path)
|
||||
parser.add_argument("lane_connections_json", type=Path)
|
||||
parser.add_argument("out_json", type=Path)
|
||||
parser.add_argument("--high", type=int, default=255)
|
||||
parser.add_argument("--low", type=int, default=1)
|
||||
args = parser.parse_args()
|
||||
|
||||
traffic = load_json(args.traffic_persistent_json)
|
||||
lanes = traffic["Data"]["RootChunk"]["data"]["lanes"]
|
||||
lane_categories = [category(flag_value(lane.get("flags", 0))) for lane in lanes]
|
||||
|
||||
connections = load_json(args.lane_connections_json)
|
||||
rows = connections["Data"]["RootChunk"]["data"]
|
||||
|
||||
changed = 0
|
||||
high_set = 0
|
||||
low_set = 0
|
||||
touched_rows = 0
|
||||
|
||||
for row in rows:
|
||||
source_index = row["index"]
|
||||
source_category = lane_categories[source_index]
|
||||
outlanes = row["value"]["outlanes"]
|
||||
if not outlanes:
|
||||
continue
|
||||
|
||||
target_categories = [
|
||||
lane_categories[out["laneIndex"]]
|
||||
for out in outlanes
|
||||
if 0 <= out["laneIndex"] < len(lane_categories)
|
||||
]
|
||||
has_highway_target = "highway" in target_categories
|
||||
has_non_highway_target = any(cat != "highway" for cat in target_categories)
|
||||
row_changed = False
|
||||
|
||||
for out in outlanes:
|
||||
target_category = lane_categories[out["laneIndex"]]
|
||||
old = out["exitProbabilityCompressed"]
|
||||
new = old
|
||||
|
||||
if target_category == "highway" and source_category in {"road", "gpsonly", "highway"}:
|
||||
new = args.high
|
||||
elif (
|
||||
source_category == "highway"
|
||||
and has_highway_target
|
||||
and target_category != "highway"
|
||||
):
|
||||
new = args.low
|
||||
elif (
|
||||
source_category in {"road", "gpsonly"}
|
||||
and has_highway_target
|
||||
and has_non_highway_target
|
||||
and target_category != "highway"
|
||||
):
|
||||
new = args.low
|
||||
|
||||
if new != old:
|
||||
out["exitProbabilityCompressed"] = new
|
||||
changed += 1
|
||||
row_changed = True
|
||||
if new == args.high:
|
||||
high_set += 1
|
||||
elif new == args.low:
|
||||
low_set += 1
|
||||
|
||||
if row_changed:
|
||||
touched_rows += 1
|
||||
|
||||
write_json(args.out_json, connections)
|
||||
print(f"rows seen: {len(rows)}")
|
||||
print(f"rows touched: {touched_rows}")
|
||||
print(f"edges changed: {changed}")
|
||||
print(f"edges set high: {high_set}")
|
||||
print(f"edges set low: {low_set}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,240 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Patch WolvenKit JSON exports of Cyberpunk traffic lane resources.
|
||||
|
||||
The game GPS planner appears to be native and data-driven over traffic lanes.
|
||||
This tool changes lane metadata that is likely to feed edge-cost selection:
|
||||
`maxSpeed` and, optionally, non-highway speed caps.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import copy
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
FLAGS = {
|
||||
"FromRoadSpline": 1,
|
||||
"Bidirectional": 2,
|
||||
"PatrolRoute": 4,
|
||||
"Pavement": 8,
|
||||
"Road": 16,
|
||||
"Intersection": 32,
|
||||
"NeverDeadEnd": 64,
|
||||
"TrafficDisabled": 128,
|
||||
"CrossWalk": 256,
|
||||
"GPSOnly": 512,
|
||||
"ShowDebug": 1024,
|
||||
"Blockade": 2048,
|
||||
"Yield": 4096,
|
||||
"NoAIDriving": 8192,
|
||||
"Highway": 16384,
|
||||
"NoAutodrive": 32768,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class PatchStats:
|
||||
files_seen: int = 0
|
||||
files_changed: int = 0
|
||||
lane_sets_seen: int = 0
|
||||
lanes_seen: int = 0
|
||||
lanes_changed: int = 0
|
||||
highway_lanes_changed: int = 0
|
||||
road_lanes_changed: int = 0
|
||||
capped_lanes_changed: int = 0
|
||||
|
||||
|
||||
def unwrap(value: Any) -> Any:
|
||||
if isinstance(value, dict):
|
||||
for key in ("value", "Value", "$value", "_value"):
|
||||
if key in value and len(value) <= 3:
|
||||
return value[key]
|
||||
return value
|
||||
|
||||
|
||||
def set_wrapped(container: dict[str, Any], key: str, value: Any) -> None:
|
||||
current = container[key]
|
||||
if isinstance(current, dict):
|
||||
for value_key in ("value", "Value", "$value", "_value"):
|
||||
if value_key in current:
|
||||
current[value_key] = value
|
||||
return
|
||||
container[key] = value
|
||||
|
||||
|
||||
def as_int(value: Any) -> int | None:
|
||||
value = unwrap(value)
|
||||
if isinstance(value, bool):
|
||||
return None
|
||||
if isinstance(value, int):
|
||||
return value
|
||||
if isinstance(value, float):
|
||||
return int(value)
|
||||
if isinstance(value, str):
|
||||
try:
|
||||
return int(value, 0)
|
||||
except ValueError:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def looks_like_lane(value: Any) -> bool:
|
||||
if not isinstance(value, dict):
|
||||
return False
|
||||
return "flags" in value and "maxSpeed" in value and (
|
||||
"length" in value or "playerGPSInfo" in value or "laneNumber" in value
|
||||
)
|
||||
|
||||
|
||||
def has_flag(flags: int, flag: str) -> bool:
|
||||
return (flags & FLAGS[flag]) == FLAGS[flag]
|
||||
|
||||
|
||||
def patch_lane(
|
||||
lane: dict[str, Any],
|
||||
*,
|
||||
highway_speed: int,
|
||||
road_speed_floor: int,
|
||||
non_highway_cap: int | None,
|
||||
) -> tuple[bool, str | None]:
|
||||
flags = as_int(lane.get("flags"))
|
||||
speed = as_int(lane.get("maxSpeed"))
|
||||
if flags is None or speed is None:
|
||||
return False, None
|
||||
|
||||
if has_flag(flags, "TrafficDisabled") or has_flag(flags, "Blockade"):
|
||||
return False, None
|
||||
if has_flag(flags, "Pavement") or has_flag(flags, "CrossWalk"):
|
||||
return False, None
|
||||
|
||||
is_highway = has_flag(flags, "Highway")
|
||||
is_road = has_flag(flags, "Road") or has_flag(flags, "GPSOnly")
|
||||
|
||||
target = speed
|
||||
reason: str | None = None
|
||||
|
||||
if is_highway and speed < highway_speed:
|
||||
target = highway_speed
|
||||
reason = "highway"
|
||||
elif is_road and speed < road_speed_floor:
|
||||
target = road_speed_floor
|
||||
reason = "road"
|
||||
|
||||
if non_highway_cap is not None and not is_highway and target > non_highway_cap:
|
||||
target = non_highway_cap
|
||||
reason = "cap"
|
||||
|
||||
if target == speed:
|
||||
return False, None
|
||||
|
||||
set_wrapped(lane, "maxSpeed", target)
|
||||
return True, reason
|
||||
|
||||
|
||||
def walk_and_patch(value: Any, stats: PatchStats, args: argparse.Namespace) -> None:
|
||||
if isinstance(value, dict):
|
||||
lanes = value.get("lanes")
|
||||
if isinstance(lanes, list) and any(looks_like_lane(lane) for lane in lanes):
|
||||
stats.lane_sets_seen += 1
|
||||
for lane in lanes:
|
||||
if not looks_like_lane(lane):
|
||||
continue
|
||||
stats.lanes_seen += 1
|
||||
changed, reason = patch_lane(
|
||||
lane,
|
||||
highway_speed=args.highway_speed,
|
||||
road_speed_floor=args.road_speed_floor,
|
||||
non_highway_cap=args.non_highway_cap,
|
||||
)
|
||||
if changed:
|
||||
stats.lanes_changed += 1
|
||||
if reason == "highway":
|
||||
stats.highway_lanes_changed += 1
|
||||
elif reason == "road":
|
||||
stats.road_lanes_changed += 1
|
||||
elif reason == "cap":
|
||||
stats.capped_lanes_changed += 1
|
||||
|
||||
for child in value.values():
|
||||
walk_and_patch(child, stats, args)
|
||||
elif isinstance(value, list):
|
||||
for child in value:
|
||||
walk_and_patch(child, stats, args)
|
||||
|
||||
|
||||
def iter_json_files(path: Path) -> list[Path]:
|
||||
if path.is_file():
|
||||
return [path]
|
||||
return sorted(path.rglob("*.json"))
|
||||
|
||||
|
||||
def output_path(input_file: Path, input_root: Path, output_root: Path) -> Path:
|
||||
if input_root.is_file():
|
||||
return output_root
|
||||
return output_root / input_file.relative_to(input_root)
|
||||
|
||||
|
||||
def patch_file(input_file: Path, input_root: Path, output_root: Path, args: argparse.Namespace, stats: PatchStats) -> None:
|
||||
stats.files_seen += 1
|
||||
original = input_file.read_text(encoding="utf-8")
|
||||
data = json.loads(original)
|
||||
patched = copy.deepcopy(data)
|
||||
|
||||
before_changed = stats.lanes_changed
|
||||
walk_and_patch(patched, stats, args)
|
||||
changed = stats.lanes_changed != before_changed
|
||||
if changed:
|
||||
stats.files_changed += 1
|
||||
|
||||
destination = output_path(input_file, input_root, output_root)
|
||||
if args.dry_run:
|
||||
return
|
||||
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
if changed or args.copy_unchanged:
|
||||
destination.write_text(
|
||||
json.dumps(patched, indent=2, ensure_ascii=False) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
elif input_root.is_file():
|
||||
destination.write_text(original, encoding="utf-8")
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("input", type=Path, help="WolvenKit JSON file or export directory")
|
||||
parser.add_argument("output", type=Path, help="Patched JSON file or output directory")
|
||||
parser.add_argument("--highway-speed", type=int, default=25)
|
||||
parser.add_argument("--road-speed-floor", type=int, default=15)
|
||||
parser.add_argument("--non-highway-cap", type=int, default=None)
|
||||
parser.add_argument("--copy-unchanged", action="store_true")
|
||||
parser.add_argument("--dry-run", action="store_true")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
if not args.input.exists():
|
||||
raise SystemExit(f"Input does not exist: {args.input}")
|
||||
|
||||
stats = PatchStats()
|
||||
for input_file in iter_json_files(args.input):
|
||||
patch_file(input_file, args.input, args.output, args, stats)
|
||||
|
||||
print(f"files seen: {stats.files_seen}")
|
||||
print(f"files changed: {stats.files_changed}")
|
||||
print(f"lane sets seen: {stats.lane_sets_seen}")
|
||||
print(f"lanes seen: {stats.lanes_seen}")
|
||||
print(f"lanes changed: {stats.lanes_changed}")
|
||||
print(f"highway boosts: {stats.highway_lanes_changed}")
|
||||
print(f"road boosts: {stats.road_lanes_changed}")
|
||||
print(f"non-highway caps: {stats.capped_lanes_changed}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,375 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Summarize EdgeWeightGPS route-result records from RED4ext logs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import collections
|
||||
import dataclasses
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
RE_TIMESTAMP = re.compile(r"^(\d{4}-\d\d-\d\d \d\d:\d\d:\d\d\.\d{3})")
|
||||
RE_ELAPSED_MS = re.compile(r" \+(\d+)ms ")
|
||||
RE_RECORD_BLOCK = re.compile(r"gpsResult_ptr28_rec40=\[(.*?)\]")
|
||||
RE_RECORD = re.compile(r"(\d+):([0-9a-f]+),0x([0-9a-f]{8}),([^;\]]+)")
|
||||
RE_JOB_COUNT = re.compile(r"GPSRouteJobBuild result call=(\d+).*?job_count50=(\d+)/0x([0-9a-f]+)")
|
||||
RE_FIELD_DEC_HEX = re.compile(r"([A-Za-z0-9_]+)=(-?\d+)/0x([0-9a-fA-F]+)")
|
||||
RE_FIELD_HEX = re.compile(r"([A-Za-z0-9_]+)=0x([0-9a-fA-F]+)")
|
||||
RE_FIELD_DEC = re.compile(r"([A-Za-z0-9_]+)=(-?\d+)(?= |$)")
|
||||
|
||||
FLAGS = {
|
||||
"FromRoadSpline": 1,
|
||||
"Bidirectional": 2,
|
||||
"PatrolRoute": 4,
|
||||
"Pavement": 8,
|
||||
"Road": 16,
|
||||
"Intersection": 32,
|
||||
"NeverDeadEnd": 64,
|
||||
"TrafficDisabled": 128,
|
||||
"CrossWalk": 256,
|
||||
"GPSOnly": 512,
|
||||
"ShowDebug": 1024,
|
||||
"Blockade": 2048,
|
||||
"Yield": 4096,
|
||||
"NoAIDriving": 8192,
|
||||
"Highway": 16384,
|
||||
"NoAutodrive": 32768,
|
||||
}
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class SubmitInfo:
|
||||
line_no: int
|
||||
timestamp: str
|
||||
elapsed_ms: int | None
|
||||
call: int
|
||||
fields: dict[str, int]
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class RouteRecord:
|
||||
index: int
|
||||
handle: int
|
||||
packed: int
|
||||
parts: list[str]
|
||||
|
||||
@property
|
||||
def low_class(self) -> int:
|
||||
return self.packed & 0xFF
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class RouteResult:
|
||||
line_no: int
|
||||
timestamp: str
|
||||
elapsed_ms: int | None
|
||||
query_id: int | None
|
||||
submit_call: int | None
|
||||
submit_ret_rva: int | None
|
||||
fields: dict[str, int]
|
||||
records: list[RouteRecord]
|
||||
|
||||
|
||||
def bytes_le(value: int) -> tuple[int, int, int, int]:
|
||||
return tuple((value >> (8 * index)) & 0xFF for index in range(4))
|
||||
|
||||
|
||||
def flag_value(flags: Any) -> int:
|
||||
if isinstance(flags, dict):
|
||||
return int(flags.get("Value", flags.get("$value", 0)) or 0)
|
||||
return int(flags or 0)
|
||||
|
||||
|
||||
def lane_category(flags: int) -> str:
|
||||
if flags & FLAGS["Highway"]:
|
||||
return "highway"
|
||||
if flags & FLAGS["GPSOnly"]:
|
||||
return "gpsonly"
|
||||
if flags & FLAGS["Road"]:
|
||||
return "road"
|
||||
if flags & FLAGS["Pavement"]:
|
||||
return "pavement"
|
||||
return "other"
|
||||
|
||||
|
||||
def parse_timestamp(line: str) -> str:
|
||||
match = RE_TIMESTAMP.search(line)
|
||||
return match.group(1) if match else "<unknown>"
|
||||
|
||||
|
||||
def parse_elapsed_ms(line: str) -> int | None:
|
||||
match = RE_ELAPSED_MS.search(line)
|
||||
return int(match.group(1)) if match else None
|
||||
|
||||
|
||||
def parse_fields(line: str) -> dict[str, int]:
|
||||
fields: dict[str, int] = {}
|
||||
for match in RE_FIELD_DEC_HEX.finditer(line):
|
||||
fields[match.group(1)] = int(match.group(2))
|
||||
for match in RE_FIELD_HEX.finditer(line):
|
||||
fields.setdefault(match.group(1), int(match.group(2), 16))
|
||||
for match in RE_FIELD_DEC.finditer(line):
|
||||
fields.setdefault(match.group(1), int(match.group(2)))
|
||||
return fields
|
||||
|
||||
|
||||
def load_lane_lookup(path: Path | None) -> dict[int, dict[str, Any]]:
|
||||
if not path:
|
||||
return {}
|
||||
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
lanes = data["Data"]["RootChunk"]["data"]["lanes"]
|
||||
lookup: dict[int, dict[str, Any]] = {}
|
||||
for index, lane in enumerate(lanes):
|
||||
node_hash = lane.get("nodeRefHash")
|
||||
if node_hash is None:
|
||||
continue
|
||||
flags = flag_value(lane.get("flags"))
|
||||
lookup[int(node_hash)] = {
|
||||
"index": index,
|
||||
"flags": flags,
|
||||
"category": lane_category(flags),
|
||||
"speed": flag_value(lane.get("maxSpeed")),
|
||||
"length": lane.get("length"),
|
||||
}
|
||||
return lookup
|
||||
|
||||
|
||||
def parse_route_records(block: str) -> list[RouteRecord]:
|
||||
records: list[RouteRecord] = []
|
||||
for record_match in RE_RECORD.finditer(block):
|
||||
pieces = record_match.group(4).split(",")
|
||||
if len(pieces) < 4:
|
||||
continue
|
||||
records.append(
|
||||
RouteRecord(
|
||||
index=int(record_match.group(1)),
|
||||
handle=int(record_match.group(2), 16),
|
||||
packed=int(record_match.group(3), 16),
|
||||
parts=pieces,
|
||||
)
|
||||
)
|
||||
return records
|
||||
|
||||
|
||||
def iter_route_results(path: Path) -> tuple[dict[int, SubmitInfo], list[RouteResult], list[int]]:
|
||||
submits: dict[int, SubmitInfo] = {}
|
||||
results: list[RouteResult] = []
|
||||
job_counts: list[int] = []
|
||||
|
||||
for line_no, line in enumerate(path.read_text(encoding="utf-8", errors="replace").splitlines(), start=1):
|
||||
fields = parse_fields(line)
|
||||
if "hook GPSQuerySubmit 0x70a42c" in line:
|
||||
call = fields.get("call")
|
||||
if call is not None:
|
||||
submits[call] = SubmitInfo(
|
||||
line_no=line_no,
|
||||
timestamp=parse_timestamp(line),
|
||||
elapsed_ms=parse_elapsed_ms(line),
|
||||
call=call,
|
||||
fields=fields,
|
||||
)
|
||||
|
||||
job_match = RE_JOB_COUNT.search(line)
|
||||
if job_match:
|
||||
job_counts.append(int(job_match.group(2)))
|
||||
|
||||
block_match = RE_RECORD_BLOCK.search(line)
|
||||
if not block_match:
|
||||
continue
|
||||
|
||||
records = parse_route_records(block_match.group(1))
|
||||
if not records:
|
||||
continue
|
||||
|
||||
results.append(
|
||||
RouteResult(
|
||||
line_no=line_no,
|
||||
timestamp=parse_timestamp(line),
|
||||
elapsed_ms=parse_elapsed_ms(line),
|
||||
query_id=fields.get("queryId"),
|
||||
submit_call=fields.get("submitCall"),
|
||||
submit_ret_rva=fields.get("submitRetRva"),
|
||||
fields=fields,
|
||||
records=records,
|
||||
)
|
||||
)
|
||||
|
||||
return submits, results, job_counts
|
||||
|
||||
|
||||
def summarize_route_result(route: RouteResult, lane_lookup: dict[int, dict[str, Any]]) -> tuple[dict[str, Any], int]:
|
||||
class_counts = collections.Counter(record.low_class for record in route.records)
|
||||
category_counts: collections.Counter[str] = collections.Counter()
|
||||
category_lengths: collections.Counter[str] = collections.Counter()
|
||||
speed_lengths: collections.Counter[int] = collections.Counter()
|
||||
unmatched = 0
|
||||
|
||||
for record in route.records:
|
||||
lane = lane_lookup.get(record.handle)
|
||||
if not lane:
|
||||
unmatched += 1
|
||||
continue
|
||||
category = lane["category"]
|
||||
length = float(lane.get("length") or 0.0)
|
||||
category_counts[category] += 1
|
||||
category_lengths[category] += length
|
||||
speed_lengths[int(lane["speed"])] += length
|
||||
|
||||
total_length = sum(category_lengths.values())
|
||||
summary = {
|
||||
"class_counts": class_counts,
|
||||
"category_counts": category_counts,
|
||||
"category_lengths": category_lengths,
|
||||
"speed_lengths": speed_lengths,
|
||||
"total_length": total_length,
|
||||
"unmatched": unmatched,
|
||||
}
|
||||
return summary, len(route.records) - unmatched
|
||||
|
||||
|
||||
def format_counter(counter: collections.Counter[Any], limit: int | None = None, digits: int | None = None) -> str:
|
||||
items = counter.most_common(limit)
|
||||
parts = []
|
||||
for key, value in items:
|
||||
if digits is None:
|
||||
parts.append(f"{key}:{value}")
|
||||
else:
|
||||
parts.append(f"{key}:{value:.{digits}f}")
|
||||
return "{" + ", ".join(parts) + "}"
|
||||
|
||||
|
||||
def print_per_route(
|
||||
path: Path,
|
||||
submits: dict[int, SubmitInfo],
|
||||
route_results: list[RouteResult],
|
||||
lane_lookup: dict[int, dict[str, Any]],
|
||||
) -> None:
|
||||
print(" per-route results:")
|
||||
for ordinal, route in enumerate(route_results, start=1):
|
||||
submit = submits.get(route.submit_call) if route.submit_call is not None else None
|
||||
summary, matched = summarize_route_result(route, lane_lookup)
|
||||
span = route.fields.get("gpsResult_u20")
|
||||
submit_fields = submit.fields if submit else {}
|
||||
ret_rva = route.submit_ret_rva
|
||||
ret_text = f"0x{ret_rva:x}" if ret_rva is not None else "?"
|
||||
submit_ms = submit.elapsed_ms if submit else None
|
||||
result_ms = route.elapsed_ms
|
||||
duration_ms = result_ms - submit_ms if submit_ms is not None and result_ms is not None else None
|
||||
submit_marker = f"+{submit_ms}ms" if submit_ms is not None else "?"
|
||||
result_marker = f"+{result_ms}ms" if result_ms is not None else "?"
|
||||
duration_marker = f"{duration_ms}ms" if duration_ms is not None else "?"
|
||||
print(
|
||||
f" route#{ordinal} qid={route.query_id} submitCall={route.submit_call} "
|
||||
f"submit={submit_marker} result={result_marker} duration={duration_marker} line={route.line_no} "
|
||||
f"ret={ret_text} records={len(route.records)} spanU20={span} matched={matched}"
|
||||
)
|
||||
if submit_fields:
|
||||
print(
|
||||
" query "
|
||||
f"f08=0x{submit_fields.get('query_f08', 0):x} "
|
||||
f"f0c={submit_fields.get('query_f0c')} "
|
||||
f"fcc={submit_fields.get('query_fcc')}"
|
||||
)
|
||||
if lane_lookup:
|
||||
category_lengths = summary["category_lengths"]
|
||||
total = summary["total_length"]
|
||||
print(
|
||||
" resource "
|
||||
f"segments={format_counter(summary['category_counts'])} "
|
||||
f"length={format_counter(category_lengths, digits=1)} "
|
||||
f"total={total:.1f} "
|
||||
f"speedLength={format_counter(summary['speed_lengths'], limit=6, digits=1)} "
|
||||
f"unmatched={summary['unmatched']}"
|
||||
)
|
||||
print(f" routeClasses={format_counter(summary['class_counts'])}")
|
||||
|
||||
|
||||
def summarize(path: Path, lane_lookup: dict[int, dict[str, Any]]) -> None:
|
||||
submits, route_results, job_counts = iter_route_results(path)
|
||||
routes = 0
|
||||
records = 0
|
||||
first_byte = collections.Counter()
|
||||
byte_positions: list[collections.Counter[int]] = [collections.Counter() for _ in range(4)]
|
||||
patterns = collections.Counter()
|
||||
segments = collections.Counter()
|
||||
class_categories: dict[int, collections.Counter[str]] = collections.defaultdict(collections.Counter)
|
||||
class_flags: dict[int, collections.Counter[int]] = collections.defaultdict(collections.Counter)
|
||||
class_speed: dict[int, collections.Counter[int]] = collections.defaultdict(collections.Counter)
|
||||
matched_handles = 0
|
||||
unmatched_handles = 0
|
||||
|
||||
for route in route_results:
|
||||
route_records = 0
|
||||
for record in route.records:
|
||||
route_records += 1
|
||||
records += 1
|
||||
bytes_tuple = bytes_le(record.packed)
|
||||
class_id = bytes_tuple[0]
|
||||
patterns[bytes_tuple] += 1
|
||||
first_byte[class_id] += 1
|
||||
for index, byte in enumerate(bytes_tuple):
|
||||
byte_positions[index][byte] += 1
|
||||
segments[(record.handle, class_id)] += 1
|
||||
|
||||
if lane_lookup:
|
||||
lane = lane_lookup.get(record.handle)
|
||||
if lane:
|
||||
matched_handles += 1
|
||||
class_categories[class_id][lane["category"]] += 1
|
||||
class_flags[class_id][lane["flags"]] += 1
|
||||
class_speed[class_id][lane["speed"]] += 1
|
||||
else:
|
||||
unmatched_handles += 1
|
||||
|
||||
if route_records:
|
||||
routes += 1
|
||||
|
||||
print(f"{path}:")
|
||||
print(f" routes={routes} records={records}")
|
||||
if job_counts:
|
||||
sorted_counts = sorted(job_counts)
|
||||
p50 = sorted_counts[len(sorted_counts) // 2]
|
||||
p90 = sorted_counts[min(len(sorted_counts) - 1, int(len(sorted_counts) * 0.9))]
|
||||
avg = sum(sorted_counts) / len(sorted_counts)
|
||||
print(
|
||||
" job_count50 "
|
||||
f"n={len(sorted_counts)} min={sorted_counts[0]} p50={p50} "
|
||||
f"p90={p90} max={sorted_counts[-1]} avg={avg:.1f}"
|
||||
)
|
||||
|
||||
print(" low-byte class counts:", dict(first_byte.most_common()))
|
||||
if lane_lookup:
|
||||
print(f" matched route records to lane hashes: {matched_handles}/{matched_handles + unmatched_handles}")
|
||||
print(" class -> resource categories:")
|
||||
for class_id, count in first_byte.most_common():
|
||||
categories = dict(class_categories[class_id].most_common())
|
||||
speeds = dict(class_speed[class_id].most_common(8))
|
||||
flags = dict(class_flags[class_id].most_common(8))
|
||||
print(f" {class_id}: categories={categories} speeds={speeds} flags={flags}")
|
||||
print_per_route(path, submits, route_results, lane_lookup)
|
||||
for index, counter in enumerate(byte_positions):
|
||||
print(f" byte{index} counts:", dict(counter.most_common(12)))
|
||||
print(" packed metadata patterns:")
|
||||
for pattern, count in patterns.most_common(24):
|
||||
print(f" {pattern}: {count}")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--traffic-json", type=Path)
|
||||
parser.add_argument("logs", nargs="+", type=Path)
|
||||
args = parser.parse_args()
|
||||
|
||||
lane_lookup = load_lane_lookup(args.traffic_json)
|
||||
for path in args.logs:
|
||||
summarize(path, lane_lookup)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,62 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Write EdgeWeightGPS momentum penalty presets as one raw float32 value."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
DEFAULT_INSTALL_PATH = Path(
|
||||
"/var/home/salt/.var/app/com.valvesoftware.Steam/.local/share/Steam/steamapps/common/"
|
||||
"Cyberpunk 2077/red4ext/plugins/EdgeWeightGPS/momentum_weights.bin"
|
||||
)
|
||||
|
||||
PRESETS: dict[str, float] = {
|
||||
"vanilla": 0.0,
|
||||
"legacy-default": 8.0,
|
||||
"mild": 4.0,
|
||||
"strong": 16.0,
|
||||
"default": 80.0,
|
||||
"silly": 80.0,
|
||||
"overstrong": 160.0,
|
||||
"pathological": 250.0,
|
||||
}
|
||||
|
||||
|
||||
def parse_penalty(text: str) -> float:
|
||||
value = float(text)
|
||||
if not 0.0 <= value <= 1000.0:
|
||||
raise argparse.ArgumentTypeError("penalty must be between 0 and 1000")
|
||||
return value
|
||||
|
||||
|
||||
def write_weight(path: Path, value: float) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(struct.pack("<f", value))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("preset", nargs="?", choices=sorted(PRESETS), default="default")
|
||||
parser.add_argument("--value", type=parse_penalty, help="explicit fixed edge penalty")
|
||||
parser.add_argument("--output", type=Path, default=DEFAULT_INSTALL_PATH)
|
||||
parser.add_argument("--preset-dir", type=Path, help="write every named preset into this directory")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.preset_dir:
|
||||
for name, value in PRESETS.items():
|
||||
path = args.preset_dir / f"{name}.bin"
|
||||
write_weight(path, value)
|
||||
print(f"{path}: {value}")
|
||||
return 0
|
||||
|
||||
value = args.value if args.value is not None else PRESETS[args.preset]
|
||||
write_weight(args.output, value)
|
||||
print(f"{args.output}: {value}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,60 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Write EdgeWeightGPS solver highway presets as one raw float32 value."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
DEFAULT_INSTALL_PATH = Path(
|
||||
"/var/home/salt/.var/app/com.valvesoftware.Steam/.local/share/Steam/steamapps/common/"
|
||||
"Cyberpunk 2077/red4ext/plugins/EdgeWeightGPS/solver_weights.bin"
|
||||
)
|
||||
|
||||
PRESETS: dict[str, float] = {
|
||||
"vanilla": 1.00,
|
||||
"default": 0.80,
|
||||
"mild": 0.90,
|
||||
"strong": 0.70,
|
||||
"proof-highway-cheap": 0.35,
|
||||
"highway-expensive": 3.00,
|
||||
}
|
||||
|
||||
|
||||
def parse_multiplier(text: str) -> float:
|
||||
value = float(text)
|
||||
if not 0.0 < value < 20.0:
|
||||
raise argparse.ArgumentTypeError("multiplier must be greater than 0 and less than 20")
|
||||
return value
|
||||
|
||||
|
||||
def write_weight(path: Path, value: float) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(struct.pack("<f", value))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("preset", nargs="?", choices=sorted(PRESETS), default="default")
|
||||
parser.add_argument("--value", type=parse_multiplier, help="explicit highway multiplier")
|
||||
parser.add_argument("--output", type=Path, default=DEFAULT_INSTALL_PATH)
|
||||
parser.add_argument("--preset-dir", type=Path, help="write every named preset into this directory")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.preset_dir:
|
||||
for name, value in PRESETS.items():
|
||||
path = args.preset_dir / f"{name}.bin"
|
||||
write_weight(path, value)
|
||||
print(f"{path}: {value}")
|
||||
return 0
|
||||
|
||||
value = args.value if args.value is not None else PRESETS[args.preset]
|
||||
write_weight(args.output, value)
|
||||
print(f"{args.output}: {value}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -1,58 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Write EdgeWeightGPS spatial weight presets as five raw float32 values."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import struct
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
DEFAULT_INSTALL_PATH = Path(
|
||||
"/var/home/salt/.var/app/com.valvesoftware.Steam/.local/share/Steam/steamapps/common/"
|
||||
"Cyberpunk 2077/red4ext/plugins/EdgeWeightGPS/spatial_weights.bin"
|
||||
)
|
||||
|
||||
PRESETS: dict[str, tuple[float, float, float, float, float]] = {
|
||||
"vanilla": (1.0, 1.0, 1.0, 1.0, 1.0),
|
||||
"current": (0.62, 1.0, 1.20, 1.35, 1.05),
|
||||
"highway-free": (0.25, 1.10, 1.35, 1.60, 1.15),
|
||||
"highway-expensive": (3.0, 1.0, 1.0, 1.0, 1.0),
|
||||
"surface-extreme": (0.45, 1.20, 1.80, 2.50, 1.30),
|
||||
}
|
||||
|
||||
|
||||
def parse_weights(text: str) -> tuple[float, float, float, float, float]:
|
||||
values = tuple(float(part) for part in text.split(","))
|
||||
if len(values) != 5:
|
||||
raise argparse.ArgumentTypeError("expected five comma-separated floats")
|
||||
return values # type: ignore[return-value]
|
||||
|
||||
|
||||
def write_weights(path: Path, values: tuple[float, float, float, float, float]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_bytes(struct.pack("<5f", *values))
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("preset", nargs="?", choices=sorted(PRESETS), default="current")
|
||||
parser.add_argument("--weights", type=parse_weights, help="five comma-separated floats: H,R,G,P,U")
|
||||
parser.add_argument("--output", type=Path, default=DEFAULT_INSTALL_PATH)
|
||||
parser.add_argument("--preset-dir", type=Path, help="write every named preset into this directory")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.preset_dir:
|
||||
for name, values in PRESETS.items():
|
||||
write_weights(args.preset_dir / f"{name}.bin", values)
|
||||
print(f"{args.preset_dir / f'{name}.bin'}: {values}")
|
||||
return 0
|
||||
|
||||
values = args.weights if args.weights is not None else PRESETS[args.preset]
|
||||
write_weights(args.output, values)
|
||||
print(f"{args.output}: {values}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user