diff --git a/fle/cluster/scenarios/open_world/control.lua b/fle/cluster/scenarios/open_world/control.lua index 6cf383aab..ed2221a7e 100644 --- a/fle/cluster/scenarios/open_world/control.lua +++ b/fle/cluster/scenarios/open_world/control.lua @@ -1 +1,3 @@ util = require("util") + +require("observation_diff") diff --git a/fle/cluster/scenarios/open_world/observation_diff.lua b/fle/cluster/scenarios/open_world/observation_diff.lua new file mode 100644 index 000000000..8a5704bdd --- /dev/null +++ b/fle/cluster/scenarios/open_world/observation_diff.lua @@ -0,0 +1,474 @@ +-- Tiered event-driven observation diffs. +-- +-- Maintains two change buffers in storage so a client can poll game-state +-- deltas in O(changes) instead of rescanning the world: +-- +-- Entity buffer (obs_diff_drain) - player-force structures as quantized +-- "rich rows". Build/mine/die/rotate events append records immediately; a +-- round-robin slice scan reconciles eventless mutations (script-set fields, +-- inventory/fuel/progress bucket transitions, entities created without +-- raise_built). +-- +-- Terrain buffer (obs_terrain_drain) - static-ish world state, streamed per +-- chunk from on_chunk_generated (new chunks generate continually, so this +-- tier is event-driven too): water bitmasks, ore tiles with quantized +-- amounts, trees, rocks/cliffs, enemy structures. Ore amounts drift with no +-- events, so the reconciler also rescans one resource chunk per pass. +-- +-- Hot per-tick fields use quantized change signatures (energy MJ, progress +-- deciles, log2 item counts, fluid buckets): an active factory only emits +-- records on bucket transitions, but each emitted row carries EXACT values, +-- so client snapshots are precise with staleness bounded by bucket width. +-- +-- RCON API (via /sc): +-- obs_diff_full_sync() -- entity snapshot; resets entity-tier state +-- obs_diff_drain() -- header + entity records +-- obs_terrain_full_sync() -- all generated chunks (expensive; mid-episode attach) +-- obs_terrain_drain() -- terrain records +-- +-- Record types (";"-joined): +-- entity buffer: h:::: u, r q +-- terrain buffer: c:: o::: d: +-- t: k: x: n,,, m +-- either: !overflow (client must full_sync) +-- +-- Row: name,x,y,direction,status followed by tagged optional fields: +-- E P R H W: D: +-- K: N T +-- I.:... F:... +-- Rows carry exact values; emission triggers on quantized-signature change. + +local RECONCILE_INTERVAL = 47 -- ticks; avoids tools' on_nth_tick(5/15/60) slots +local ENTITY_SLICE = 50 -- entities re-rowed per reconcile pass +local RES_BUCKET = 100 -- ore units per amount bucket +local RES_CHUNKS_PER_PASS = 4 -- resource chunks rescanned per reconcile pass +local DISCOVERY_EVERY = 32 -- reconcile passes between untracked-entity sweeps +local MAX_BUF = 50000 -- records; overflow => client full_sync + +local WATER_TILES = { "water", "deepwater", "water-green", "deepwater-green" } + +local function state() + local s = storage.obs_diff + if not s then + s = { + cache = {}, ents = {}, keys = {}, cursor = 0, + ebuf = {}, ebuf_n = 0, tbuf = {}, tbuf_n = 0, + res_chunks = {}, res_cursor = 0, + } + storage.obs_diff = s + end + return s +end + +local function qlog(n) + if n <= 0 then return 0 end + return math.floor(math.log(n) / math.log(2)) + 1 +end + +local function push_e(s, rec) + if s.ebuf_n >= MAX_BUF then s.ebuf[MAX_BUF] = "!overflow" return end + s.ebuf_n = s.ebuf_n + 1 + s.ebuf[s.ebuf_n] = rec +end + +local function push_t(s, rec) + if s.tbuf_n >= MAX_BUF then s.tbuf[MAX_BUF] = "!overflow" return end + s.tbuf_n = s.tbuf_n + 1 + s.tbuf[s.tbuf_n] = rec +end + +local function rich_row(e) + -- Returns (row, sig). The row carries EXACT values (energy in joules, + -- progress in percent, raw item/fluid counts) so the client snapshot is + -- precise; the sig quantizes hot fields, and only a sig change triggers + -- emission, so drains stay sparse while transmitted values stay exact. + local base = e.name .. "," .. e.position.x .. "," .. e.position.y .. "," + .. (e.direction or 0) .. "," .. (e.status or 0) + local row = { base } + local sig = { base } + local ok, energy = pcall(function() return e.energy end) + if ok and energy and energy > 0 then + row[#row + 1] = "E" .. math.floor(energy) + sig[#sig + 1] = "E" .. math.floor(energy / 1e6) + end + local okp, progress = pcall(function() return e.crafting_progress end) + if okp and progress and progress > 0 then + row[#row + 1] = "P" .. math.floor(progress * 100) + sig[#sig + 1] = "P" .. math.floor(progress * 10) + end + local okr, recipe = pcall(function() return e.get_recipe() end) + if okr and recipe then + row[#row + 1] = "R" .. recipe.name + sig[#sig + 1] = "R" .. recipe.name + end + local okh, health = pcall(function() return e.health end) + if okh and health then + row[#row + 1] = "H" .. math.floor(health) + sig[#sig + 1] = "H" .. math.floor(health / 50) + end + local okw, tw, th = pcall(function() + return e.prototype.tile_width, e.prototype.tile_height + end) + if okw and tw then + local dims = "W" .. tw .. ":" .. th + row[#row + 1] = dims + sig[#sig + 1] = dims + end + local okd, drop = pcall(function() return e.drop_position end) + if okd and drop then + local rec = "D" .. drop.x .. ":" .. drop.y + row[#row + 1] = rec + sig[#sig + 1] = rec + end + local okk, pickup = pcall(function() return e.pickup_position end) + if okk and pickup then + local rec = "K" .. pickup.x .. ":" .. pickup.y + row[#row + 1] = rec + sig[#sig + 1] = rec + end + local okn, netid = pcall(function() return e.electric_network_id end) + if okn and netid then + local rec = "N" .. netid + row[#row + 1] = rec + sig[#sig + 1] = rec + end + local okt, temp = pcall(function() return e.temperature end) + if okt and temp then + row[#row + 1] = "T" .. math.floor(temp) + sig[#sig + 1] = "T" .. math.floor(temp / 25) + end + for idx = 1, 4 do + local inv = e.get_inventory(idx) + if inv then + for _, item in ipairs(inv.get_contents()) do + row[#row + 1] = "I" .. idx .. "." .. item.name .. ":" .. item.count + sig[#sig + 1] = "I" .. idx .. "." .. item.name .. ":" .. qlog(item.count) + end + end + end + local okf, fb = pcall(function() return e.fluidbox end) + if okf and fb and #fb > 0 then + for j = 1, #fb do + local f = fb[j] + if f then + row[#row + 1] = "F" .. f.name .. ":" .. math.floor(f.amount) + sig[#sig + 1] = "F" .. f.name .. ":" .. qlog(f.amount) + end + end + end + return table.concat(row, ","), table.concat(sig, ",") +end + +local function upsert(e) + if not (e and e.valid and e.unit_number) then return end + local s = state() + local key = e.unit_number + local row, sig = rich_row(e) + if s.cache[key] ~= sig then + s.cache[key] = sig + s.ents[key] = e + push_e(s, "u" .. key .. "," .. row) + end +end + +local function remove_key(s, key) + if s.cache[key] == nil then return end + s.cache[key] = nil + s.ents[key] = nil + push_e(s, "r" .. key) +end + +-- Terrain helpers ------------------------------------------------------------ + +local function pos_key(p) + return math.floor(p.x) .. ":" .. math.floor(p.y) +end + +local function encode_chunk(surface, cx, cy, area, emit) + local s = state() + -- Water bitmask: 32 rows of 32 bits, hex-encoded (empty string = all land) + local water_hex = "" + local okw, tiles = pcall(function() + return surface.find_tiles_filtered{ area = area, name = WATER_TILES } + end) + if okw and tiles and #tiles > 0 then + local rows = {} + for i = 0, 31 do rows[i] = 0 end + local x0, y0 = area.left_top.x, area.left_top.y + for i = 1, #tiles do + local p = tiles[i].position + local dx = math.floor(p.x) - x0 + local dy = math.floor(p.y) - y0 + if dx >= 0 and dx < 32 and dy >= 0 and dy < 32 then + rows[dy] = bit32.bor(rows[dy], bit32.lshift(1, dx)) + end + end + local hex = {} + for i = 0, 31 do hex[i + 1] = string.format("%08x", rows[i]) end + water_hex = table.concat(hex) + end + emit("c" .. cx .. ":" .. cy .. ":" .. water_hex) + + -- Ore tiles: tracked for amount reconciliation + local ores = surface.find_entities_filtered{ area = area, type = "resource" } + if #ores > 0 then + local cache = {} + for i = 1, #ores do + local o = ores[i] + local key = pos_key(o.position) + local bucket = math.floor(o.amount / RES_BUCKET) + cache[key] = bucket + emit("o" .. o.name .. ":" .. key .. ":" .. bucket) + end + s.res_chunks[#s.res_chunks + 1] = { area = area, cache = cache } + end + + for _, tree in ipairs(surface.find_entities_filtered{ area = area, type = "tree" }) do + emit("t" .. pos_key(tree.position)) + end + for _, rock in ipairs(surface.find_entities_filtered{ area = area, type = { "simple-entity", "cliff" } }) do + emit("k" .. pos_key(rock.position)) + end + for _, nest in ipairs(surface.find_entities_filtered{ area = area, force = "enemy", type = { "unit-spawner", "turret" } }) do + if nest.unit_number then + emit("n" .. nest.unit_number .. "," .. nest.name .. "," + .. nest.position.x .. "," .. nest.position.y) + end + end +end + +-- Event wiring --------------------------------------------------------------- + +local ev = defines.events + +script.on_event(ev.on_chunk_generated, function(event) + local s = state() + encode_chunk(event.surface, event.position.x, event.position.y, event.area, + function(rec) push_t(s, rec) end) +end) + +for _, id in pairs({ ev.on_built_entity, ev.on_robot_built_entity, + ev.script_raised_built, ev.script_raised_revive }) do + script.on_event(id, function(event) upsert(event.entity) end) +end +script.on_event(ev.on_entity_cloned, function(event) upsert(event.destination) end) +script.on_event(ev.on_player_rotated_entity, function(event) upsert(event.entity) end) + +local function handle_removal(event) + local e = event.entity + if not (e and e.valid) then return end + local s = state() + local t = e.type + if t == "tree" or t == "simple-entity" or t == "cliff" then + push_t(s, "x" .. pos_key(e.position)) + elseif e.unit_number then + if e.force.name == "enemy" then + push_t(s, "m" .. e.unit_number) + else + remove_key(s, e.unit_number) + end + end +end +for _, id in pairs({ ev.on_player_mined_entity, ev.on_robot_mined_entity, + ev.on_entity_died, ev.script_raised_destroy }) do + script.on_event(id, handle_removal) +end + +script.on_event(ev.on_resource_depleted, function(event) + local e = event.entity + if e and e.valid then + push_t(state(), "d" .. pos_key(e.position)) + end +end) + +script.on_event(ev.on_research_finished, function(event) + push_e(state(), "q" .. event.research.name) +end) + +-- Reconciliation: entity slice + one resource chunk per pass ---------------- + +script.on_nth_tick(RECONCILE_INTERVAL, function() + local s = state() + + -- Discovery sweep: entities created without raise_built never fire an + -- event, so periodically scan for player entities we aren't tracking. + s.discovery = (s.discovery or 0) + 1 + if s.discovery >= DISCOVERY_EVERY then + s.discovery = 0 + for _, e in ipairs(game.surfaces[1].find_entities_filtered{ force = "player" }) do + if e.unit_number and s.cache[e.unit_number] == nil then + upsert(e) + end + end + end + + for _ = 1, ENTITY_SLICE do + s.cursor = s.cursor + 1 + if s.cursor > #s.keys then + s.keys = {} + for key in pairs(s.ents) do + s.keys[#s.keys + 1] = key + end + s.cursor = 0 + break + end + local key = s.keys[s.cursor] + local e = s.ents[key] + if e then + if not e.valid then + remove_key(s, key) + else + local row, sig = rich_row(e) + if s.cache[key] ~= sig then + s.cache[key] = sig + push_e(s, "u" .. key .. "," .. row) + end + end + end + end + + local n_chunks = #s.res_chunks + for _ = 1, math.min(RES_CHUNKS_PER_PASS, n_chunks) do + s.res_cursor = (s.res_cursor % n_chunks) + 1 + local rc = s.res_chunks[s.res_cursor] + local surface = game.surfaces[1] + local found = {} + for _, o in ipairs(surface.find_entities_filtered{ area = rc.area, type = "resource" }) do + local key = pos_key(o.position) + local bucket = math.floor(o.amount / RES_BUCKET) + found[key] = true + if rc.cache[key] ~= bucket then + rc.cache[key] = bucket + push_t(s, "o" .. o.name .. ":" .. key .. ":" .. bucket) + end + end + for key in pairs(rc.cache) do + if not found[key] then + rc.cache[key] = nil + push_t(s, "d" .. key) + end + end + end +end) + +-- Drain / full sync ---------------------------------------------------------- + +local function header() + local force = game.forces.player + local research = force.current_research + -- Track the agent character so clients can build egocentric views. The + -- reference is cached in storage and revalidated when it dies/despawns. + local char = storage.obs_char + if not (char and char.valid) then + char = nil + local players = game.connected_players + if #players > 0 and players[1].character then + char = players[1].character + else + local found = game.surfaces[1].find_entities_filtered{ type = "character", limit = 1 } + char = found[1] + end + storage.obs_char = char + end + local px, py = 0, 0 + if char and char.valid then + px, py = char.position.x, char.position.y + end + return "h" .. game.tick .. ":" .. (research and research.name or "-") + .. ":" .. math.floor((force.research_progress or 0) * 100) + .. ":" .. px .. ":" .. py +end + +-- Direct visibility hook for FLE tool Lua: after an eventless mutation +-- (inventory insert, direction assignment), tools may call +-- obs_diff_touch(entity) to re-row it immediately instead of waiting for +-- the reconciler sweep. Guarded at call sites, so tools stay compatible +-- with scenarios that do not load this mod. +function obs_diff_touch(e) + upsert(e) +end + +function obs_diff_drain() + local s = state() + if s.ebuf_n == 0 then + rcon.print(header()) + return + end + rcon.print(header() .. ";" .. table.concat(s.ebuf, ";", 1, s.ebuf_n)) + s.ebuf = {} + s.ebuf_n = 0 +end + +function obs_terrain_drain() + local s = state() + if s.tbuf_n == 0 then + rcon.print("") + return + end + rcon.print(table.concat(s.tbuf, ";", 1, s.tbuf_n)) + s.tbuf = {} + s.tbuf_n = 0 +end + +function obs_all_drain() + local s = state() + local e = header() + if s.ebuf_n > 0 then + e = e .. ";" .. table.concat(s.ebuf, ";", 1, s.ebuf_n) + s.ebuf = {} + s.ebuf_n = 0 + end + local t = "" + if s.tbuf_n > 0 then + t = table.concat(s.tbuf, ";", 1, s.tbuf_n) + s.tbuf = {} + s.tbuf_n = 0 + end + rcon.print(e .. "~" .. t) +end + +function obs_diff_full_sync() + local s = state() + s.cache = {} + s.ents = {} + s.keys = {} + s.cursor = 0 + s.ebuf = {} + s.ebuf_n = 0 + local out = { header() } + local n = 1 + local ents = game.surfaces[1].find_entities_filtered{ force = "player" } + for i = 1, #ents do + local e = ents[i] + local key = e.unit_number + if key then + local row, sig = rich_row(e) + s.cache[key] = sig + s.ents[key] = e + n = n + 1 + out[n] = "u" .. key .. "," .. row + end + end + rcon.print(table.concat(out, ";", 1, n)) +end + +function obs_terrain_full_sync() + local s = state() + s.res_chunks = {} + s.res_cursor = 0 + s.tbuf = {} + s.tbuf_n = 0 + local surface = game.surfaces[1] + local out = {} + local n = 0 + local function emit(rec) + n = n + 1 + out[n] = rec + end + for chunk in surface.get_chunks() do + if surface.is_chunk_generated(chunk) then + encode_chunk(surface, chunk.x, chunk.y, chunk.area, emit) + end + end + rcon.print(table.concat(out, ";", 1, n)) +end \ No newline at end of file diff --git a/fle/env/tools/agent/insert_item/server.lua b/fle/env/tools/agent/insert_item/server.lua index 85afc87b9..0d35c992c 100644 --- a/fle/env/tools/agent/insert_item/server.lua +++ b/fle/env/tools/agent/insert_item/server.lua @@ -264,7 +264,7 @@ storage.actions.insert_item = function(player_index, insert_item, count, x, y, t if inserted > 0 then -- Only remove successfully inserted items from player player.remove_item{name=insert_item, count=inserted} - -- game.print("Successfully inserted " .. inserted .. " items.") + if obs_diff_touch then obs_diff_touch(closest_entity) end return storage.utils.serialize_entity(closest_entity) else local inventory_info = get_inventory_info(closest_entity) diff --git a/fle/env/tools/agent/place_entity/server.lua b/fle/env/tools/agent/place_entity/server.lua index 2b93d5a88..41e0477ef 100644 --- a/fle/env/tools/agent/place_entity/server.lua +++ b/fle/env/tools/agent/place_entity/server.lua @@ -174,6 +174,7 @@ storage.actions.place_entity = function(player_index, entity, direction, x, y, e force = "player", position = position, direction = entity_direction, + raise_built = true, } if placed_entity then @@ -271,6 +272,7 @@ storage.actions.place_entity = function(player_index, entity, direction, x, y, e force = player.force, position = new_position, direction = entity_direction, + raise_built = true, } if have_built then player.remove_item{name = entity, count = 1} @@ -372,6 +374,7 @@ storage.actions.place_entity = function(player_index, entity, direction, x, y, e force = player.force, position = position, direction = entity_direction, + raise_built = true, } if have_built then diff --git a/fle/env/tools/agent/rotate_entity/server.lua b/fle/env/tools/agent/rotate_entity/server.lua index a73ed8c76..c17674e14 100644 --- a/fle/env/tools/agent/rotate_entity/server.lua +++ b/fle/env/tools/agent/rotate_entity/server.lua @@ -95,7 +95,7 @@ storage.actions.rotate_entity = function(player_index, x, y, direction, entity) end -- Destroy the entity - closest_entity.destroy() + closest_entity.destroy{raise_destroy = true} -- Create new entity with target direction local new_entity = surface.create_entity{ @@ -103,7 +103,8 @@ storage.actions.rotate_entity = function(player_index, x, y, direction, entity) position = saved_position, direction = target, force = saved_force, - create_build_effect_smoke = false + create_build_effect_smoke = false, + raise_built = true } if new_entity then @@ -174,6 +175,8 @@ storage.actions.rotate_entity = function(player_index, x, y, direction, entity) closest_entity.teleport(aligned_position) end + if obs_diff_touch then obs_diff_touch(closest_entity) end + local serialized = storage.utils.serialize_entity(closest_entity) return serialized end \ No newline at end of file diff --git a/tests/benchmarks/benchmark_action_obs.py b/tests/benchmarks/benchmark_action_obs.py new file mode 100644 index 000000000..566bfd766 --- /dev/null +++ b/tests/benchmarks/benchmark_action_obs.py @@ -0,0 +1,194 @@ +"""Benchmark FLE action <-> observation round trips on the tiered protocol. + +Uses a real FactorioInstance so actions run through FLE's existing tool Lua +(verified semantics, unmodified). For each action we measure: + - action latency: the FLE tool call itself (RCON round trip + Lua) + - poll latency: one obs_all_drain -> reconcile -> observation() afterwards + - visibility latency: time until the action's effect appears in the client + state. FLE tools mutate the world via script (no raise_built / events), + so visibility rides the reconciler / discovery tiers - this measures the + real act->perceive loop an RL agent would experience. + +Usage: python tests/benchmarks/benchmark_action_obs.py [--port 27099] +""" + +import argparse +import statistics +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) +from benchmark_tensor_obs import TensorClient # noqa: E402 + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--port", type=int, default=27099) + ap.add_argument("--reps", type=int, default=8) + args = ap.parse_args() + + from fle.env import FactorioInstance + from fle.env.entities import Direction, Position + from fle.env.game_types import Prototype + + print("connecting FactorioInstance (injects tool Lua)...", flush=True) + instance = FactorioInstance( + address="localhost", + tcp_port=args.port, + fast=True, + cache_scripts=True, + all_technologies_researched=True, + inventory={ + "stone-furnace": 40, + "burner-inserter": 40, + "transport-belt": 40, + "coal": 400, + "iron-plate": 400, + }, + ) + instance.set_speed(10.0) + ns = instance.namespace + rc = instance.rcon_client + + # warm the server: first minutes after a container start run 10-50x + # slow under box64 dynarec; gate on a sub-5ms noop before measuring + t0 = time.perf_counter() + while time.perf_counter() - t0 < 90: + s = time.perf_counter() + rc.send_command("/sc rcon.print(1)") + if (time.perf_counter() - s) * 1000 < 5: + break + time.sleep(1) + print(f"warmup: {time.perf_counter() - t0:.0f} s", flush=True) + + client = TensorClient(table_rows=8192) + client.apply_entity(rc.send_command("/sc obs_diff_full_sync()") or "") + resp = rc.send_command("/sc obs_terrain_full_sync()") or "" + client.apply_terrain(resp) + print(f"synced: {len(client.entities)} entities, player at " + f"({client.player_x:.1f}, {client.player_y:.1f})\n", flush=True) + + def poll(): + resp = rc.send_command("/sc obs_all_drain()") or "" + epart, _, tpart = resp.partition("~") + client.apply_entity(epart) + client.apply_terrain(tpart) + _ = client.observation() + + def visible_within(predicate, timeout=90.0): + t0 = time.perf_counter() + while time.perf_counter() - t0 < timeout: + poll() + if predicate(): + return (time.perf_counter() - t0) * 1000 + time.sleep(0.02) + return float("inf") + + results = {} + + def record(name, act_ms, poll_ms, vis_ms): + results.setdefault(name, {"act": [], "poll": [], "vis": []}) + results[name]["act"].append(act_ms) + results[name]["poll"].append(poll_ms) + results[name]["vis"].append(vis_ms) + + def timed(fn): + t0 = time.perf_counter() + out = fn() + return out, (time.perf_counter() - t0) * 1000 + + start = ns.player_location + for i in range(args.reps): + # setup (untimed): walk to a fresh build site, everything within reach + ns.move_to(Position(x=start.x + i * 8, y=start.y)) + base = ns.player_location + fx, fy = base.x + 3, base.y + 3 + + # --- place_entity: script placement, eventless -> discovery sweep --- + furnace, act = timed(lambda: ns.place_entity( + Prototype.StoneFurnace, position=Position(x=fx, y=fy))) + n_before = len(client.entities) + _, p = timed(poll) + vis = visible_within(lambda: any( + r.startswith("stone-furnace") and abs(float(r.split(",")[1]) - furnace.position.x) < 1 + and abs(float(r.split(",")[2]) - furnace.position.y) < 1 + for r in client.entities.values())) + record("place_entity", act, p, vis) + + # --- insert_item into that furnace: bucket change -> reconciler --- + _, act = timed(lambda: ns.insert_item(Prototype.Coal, furnace, quantity=10)) + _, p = timed(poll) + + def furnace_has_coal(): + for r in client.entities.values(): + if (r.startswith("stone-furnace") + and abs(float(r.split(",")[1]) - furnace.position.x) < 1 + and abs(float(r.split(",")[2]) - furnace.position.y) < 1): + return ".coal:" in r + return False + + vis = visible_within(furnace_has_coal) + record("insert_item", act, p, vis) + + # --- place + rotate an inserter: script rotation, eventless --- + ins, _ = timed(lambda: ns.place_entity( + Prototype.BurnerInserter, position=Position(x=fx + 2.5, y=fy - 1))) + visible_within(lambda: any( + r.startswith("burner-inserter") and abs(float(r.split(",")[1]) - ins.position.x) < 1 + and abs(float(r.split(",")[2]) - ins.position.y) < 1 + for r in client.entities.values())) + def row_dirs(): + # FLE's Direction enum does not map 1:1 onto engine e.direction for + # inserters (drop-side semantics), and rotate_entity may recreate + # the entity for some types - so detect visibility as "the row + # direction at this position changed from its pre-rotation value". + return {int(r.split(",")[3]) for r in client.entities.values() + if r.startswith("burner-inserter") + and abs(float(r.split(",")[1]) - ins.position.x) < 1 + and abs(float(r.split(",")[2]) - ins.position.y) < 1} + + dirs_before = row_dirs() + rotated, act = timed(lambda: ns.rotate_entity(ins, Direction.LEFT)) + _, p = timed(poll) + vis = visible_within(lambda: bool(row_dirs() - dirs_before)) + record("rotate_entity", act, p, vis) + + # --- craft_item: player inventory change (not entity state) --- + _, act = timed(lambda: ns.craft_item(Prototype.IronGearWheel, quantity=1)) + _, p = timed(poll) + record("craft_item", act, p, 0.0) # player inventory is not entity state + + # --- move_to: header position, read live every drain --- + target = Position(x=base.x + 10 + (i % 3) * 2, y=base.y) + _, act = timed(lambda: ns.move_to(target)) + _, p = timed(poll) + vis = visible_within( + lambda: abs(client.player_x - ns.player_location.x) < 0.5 + and abs(client.player_y - ns.player_location.y) < 0.5) + record("move_to", act, p, vis) + + print(f"{'action':16s} {'action ms (p50)':>16s} {'obs poll ms':>12s} {'visible-in ms':>14s}") + for name, r in results.items(): + vis = [v for v in r["vis"] if v != float("inf")] + vis_str = f"{statistics.median(vis):>10.0f}" if vis else "TIMEOUT" + misses = len(r["vis"]) - len(vis) + note = f" ({misses} timeouts)" if misses else "" + print(f"{name:16s} {statistics.median(r['act']):>16.1f} " + f"{statistics.median(r['poll']):>12.1f} {vis_str:>14s}{note}", flush=True) + + # canonical RL step: action + one observation, no visibility wait + ts = [] + pos0 = ns.player_location + for i in range(20): + t0 = time.perf_counter() + ns.move_to(Position(x=pos0.x + (i % 5), y=pos0.y)) + poll() + ts.append((time.perf_counter() - t0) * 1000) + print(f"\nact+observe cycle (move_to + poll): p50 {statistics.median(ts):.1f} ms " + f"-> {1000 / statistics.median(ts):.0f} steps/s", flush=True) + instance.cleanup() + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/benchmarks/benchmark_event_diff.py b/tests/benchmarks/benchmark_event_diff.py new file mode 100644 index 000000000..d7d28b4f5 --- /dev/null +++ b/tests/benchmarks/benchmark_event_diff.py @@ -0,0 +1,159 @@ +"""Benchmark the control.lua event-driven diff mod on the open_world scenario.""" + +import statistics +import time + +from factorio_rcon import RCONClient + +rc = RCONClient("localhost", 27099, "factorio") +rc.connect() + +state = {} + + +def apply_diff(resp): + if not resp: + return 0 + changes = resp.split(";") + for c in changes: + if c[0] == "u": + key, row = c[1:].split(",", 1) + state[key] = row + elif c[0] == "r": + state.pop(c[1:], None) + return len(changes) + + +def drain(): + return rc.send_command("/sc obs_diff_drain()") or "" + + +def bench(fn, iters=15, warmup=2): + for _ in range(warmup): + fn() + ts = [] + for _ in range(iters): + t0 = time.perf_counter() + fn() + ts.append((time.perf_counter() - t0) * 1000) + return statistics.mean(ts), statistics.median(ts), min(ts), max(ts) + + +def fmt(label, r, extra=""): + m, p50, mn, mx = r + print(f"{label:34s} mean {m:7.1f} ms p50 {p50:7.1f} min {mn:7.1f} max {mx:7.1f} -> {1000 / m:6.0f} obs/s {extra}", flush=True) + + +# --- 1. spawn 1000 entities WITH raise_built: events populate the buffer --- +SPAWN = """ +local surface = game.surfaces[1] +local names = {"stone-furnace", "transport-belt", "burner-inserter", "assembling-machine-1", "iron-chest"} +local placed = 0 +for i = 0, 999 do + local e = surface.create_entity{name = names[(i % 5) + 1], + position = {-140 + (i % 70) * 4, -140 + math.floor(i / 70) * 4}, + force = "player", raise_built = true} + if e then placed = placed + 1 end +end +rcon.print(placed) +""" +t0 = time.perf_counter() +print("spawned:", rc.send_command("/sc " + SPAWN.strip()), f"({(time.perf_counter() - t0) * 1000:.0f} ms)") + +t0 = time.perf_counter() +resp = drain() +n = apply_diff(resp) +print(f"drain after spawn: {n} records, {len(resp) / 1024:.1f} KB, {(time.perf_counter() - t0) * 1000:.1f} ms, client entities = {len(state)}") + +# --- 2. full sync (cold client) --- +def full_sync(): + resp = rc.send_command("/sc obs_diff_full_sync()") or "" + state.clear() + apply_diff(resp) + return resp + +t0 = time.perf_counter() +resp = full_sync() +print(f"full_sync: {len(state)} entities, {len(resp) / 1024:.1f} KB, {(time.perf_counter() - t0) * 1000:.1f} ms") + +# --- 3. steady state: nothing changed --- +sizes = [] +def poll(): + resp = drain() + sizes.append(len(resp)) + apply_diff(resp) + +fmt("steady-state drain", bench(poll), f"(avg payload {statistics.mean(sizes):.0f} B)") + +# --- 4. event churn: 25 builds + 25 destroys (raised) between polls --- +CHURN = """ +local surface = game.surfaces[1] +local built = 0 +for i = 0, 24 do + local e = surface.create_entity{name = "wooden-chest", + position = {150 + (i % 5) * 2, 150 + math.floor(i / 5) * 2}, + force = "player", raise_built = true} + if e then built = built + 1 end +end +local chests = surface.find_entities_filtered{name = "wooden-chest", force = "player"} +local killed = 0 +for i = 1, math.min(25, #chests) do + chests[i].destroy{raise_destroy = true} + killed = killed + 1 +end +rcon.print(built .. "/" .. killed) +""" +sizes.clear() +ts = [] +for _ in range(12): + rc.send_command("/sc " + CHURN.strip()) + t0 = time.perf_counter() + poll() + ts.append((time.perf_counter() - t0) * 1000) +fmt("50-event churn drain", (statistics.mean(ts), statistics.median(ts), min(ts), max(ts)), f"(avg payload {statistics.mean(sizes):.0f} B)") + +# --- 5. eventless churn: script-set direction, reconciler must catch it --- +ROTATE = """ +local belts = game.surfaces[1].find_entities_filtered{name = "transport-belt", force = "player", limit = 50} +for _, b in ipairs(belts) do b.direction = (b.direction + 4) % 16 end +rcon.print(#belts) +""" +print("\nrotating 50 belts via script (no events fired)...") +rc.send_command("/sc " + ROTATE.strip()) +t0 = time.perf_counter() +caught = 0 +while caught < 50 and time.perf_counter() - t0 < 30: + resp = drain() + caught += sum(1 for c in resp.split(";") if c and c[0] == "u" and "transport-belt" in c) + apply_diff(resp) + time.sleep(0.25) +print(f"reconciler caught {caught}/50 rotations in {time.perf_counter() - t0:.1f} s") + +# --- 6. consistency check: client state vs authoritative full scan --- +SCAN = """ +local out = {} +local n = 0 +local ents = game.surfaces[1].find_entities_filtered{force = "player"} +for i = 1, #ents do + local e = ents[i] + if e.unit_number then + n = n + 1 + out[n] = e.unit_number .. "," .. e.name .. "," .. e.position.x .. "," .. e.position.y .. "," .. (e.direction or 0) + end +end +rcon.print(table.concat(out, ";")) +""" +resp = rc.send_command("/sc " + SCAN.strip()) or "" +truth = {} +for rec in resp.split(";"): + if rec: + key, rest = rec.split(",", 1) + truth[key] = rest +mine = {k: v.rsplit(",", 1)[0] for k, v in state.items()} # drop status (flickers) +missing = set(truth) - set(mine) +extra = set(mine) - set(truth) +mismatch = {k for k in set(truth) & set(mine) if truth[k] != mine[k]} +print(f"\nconsistency: server={len(truth)} client={len(mine)} missing={len(missing)} extra={len(extra)} field-mismatch={len(mismatch)}") +if mismatch: + k = next(iter(mismatch)) + print(" example mismatch:", k, "server:", truth[k], "client:", mine[k]) diff --git a/tests/benchmarks/benchmark_stress.py b/tests/benchmarks/benchmark_stress.py new file mode 100644 index 000000000..2d50dea6c --- /dev/null +++ b/tests/benchmarks/benchmark_stress.py @@ -0,0 +1,218 @@ +"""Stress benchmark: tiered observation protocol against a large active factory. + +Builds smelting cells (furnace + burner-inserter + chest + belt) in stages up +to ~20k entities spread over ~500x260 tiles, fuels everything, and measures at +each scale: steady-state drains, UPS, drift-detection latency, cold-attach +cost, and recenter cost. Table capacity is raised to 32k rows. + +Usage: python tests/benchmarks/benchmark_stress.py [--port 27099] +""" + +import argparse +import statistics +import sys +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent)) +from benchmark_tensor_obs import TensorClient # noqa: E402 + +STRESS_TABLE_ROWS = 32768 +SCALES = (5000, 10000, 20000) +CELLS_PER_ROW = 100 # cell = 5x5 tiles; 100 cells/row -> x in [-250, 250) + + +def drain_apply(rc, client): + resp = rc.send_command("/sc obs_all_drain()") or "" + epart, _, tpart = resp.partition("~") + client.apply_entity(epart) + client.apply_terrain(tpart) + return len(resp) + + +def spawn_cells(rc, first_cell, n_cells): + """Each cell: stone-furnace, burner-inserter, iron-chest, transport-belt.""" + lua = f""" + local surface = game.surfaces[1] + local placed = 0 + for i = {first_cell}, {first_cell + n_cells - 1} do + local cx = -250 + (i % {CELLS_PER_ROW}) * 5 + local cy = -250 + math.floor(i / {CELLS_PER_ROW}) * 5 + local specs = {{ + {{"stone-furnace", cx + 1, cy + 1}}, + {{"burner-inserter", cx + 2.5, cy + 0.5}}, + {{"iron-chest", cx + 3.5, cy + 0.5}}, + {{"transport-belt", cx + 2.5, cy + 2.5}}, + }} + for _, s in ipairs(specs) do + local e = surface.create_entity{{name = s[1], position = {{s[2], s[3]}}, + force = "player", raise_built = true}} + if e then placed = placed + 1 end + end + end + rcon.print(placed) + """ + return int(rc.send_command("/sc " + lua.strip())) + + +FUEL = """ +local n = 0 +for _, f in ipairs(game.surfaces[1].find_entities_filtered{name = "stone-furnace", force = "player"}) do + f.get_inventory(defines.inventory.fuel).insert{name = "coal", count = 50} + f.get_inventory(defines.inventory.furnace_source).insert{name = "iron-ore", count = 100} + n = n + 1 +end +for _, i in ipairs(game.surfaces[1].find_entities_filtered{name = "burner-inserter", force = "player"}) do + i.get_inventory(defines.inventory.fuel).insert{name = "coal", count = 5} +end +rcon.print(n) +""" + + +def measure_ups(rc, seconds=5): + t0 = time.perf_counter() + a = int(rc.send_command("/sc rcon.print(game.tick)")) + time.sleep(seconds) + b = int(rc.send_command("/sc rcon.print(game.tick)")) + return (b - a) / (time.perf_counter() - t0) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--port", type=int, default=27099) + args = ap.parse_args() + + from factorio_rcon import RCONClient + + rc = RCONClient("localhost", args.port, "factorio") + rc.connect() + rc.send_command("/sc game.speed = 10") + client = TensorClient(table_rows=STRESS_TABLE_ROWS) + + # terrain for the whole build area (+ some margin) before spawning + print("generating chunks (radius 12)...", flush=True) + t0 = time.perf_counter() + rc.send_command("/sc game.surfaces[1].request_to_generate_chunks({0, 0}, 12) " + "game.surfaces[1].force_generate_chunk_requests() rcon.print(1)") + print(f" chunk generation: {time.perf_counter() - t0:.1f} s", flush=True) + n = drain_apply(rc, client) + print(f" terrain drain: {n / 1024:.0f} KB -> {len(client.water)} chunks, " + f"{len(client.ores)} ores, {len(client.trees)} trees " + f"(overflow={client.overflow})", flush=True) + + rc.send_command("""/sc if #game.surfaces[1].find_entities_filtered{type = "character"} == 0 then + game.surfaces[1].create_entity{name = "character", position = {0, 0}, force = "player", raise_built = true} +end rcon.print(1)""".strip()) + + total_cells = 0 + for target in SCALES: + cells_needed = target // 4 + # spawn in batches, draining between to stay under buffer caps + t0 = time.perf_counter() + while total_cells < cells_needed: + batch = min(500, cells_needed - total_cells) + spawn_cells(rc, total_cells, batch) + total_cells += batch + drain_apply(rc, client) + build_t = time.perf_counter() - t0 + n_fueled = int(rc.send_command("/sc " + FUEL.strip())) + time.sleep(2.0) + drain_apply(rc, client) + + n_entities = len(client.entities) + print(f"\n=== scale ~{target}: {n_entities} tracked entities, " + f"{n_fueled} furnaces smelting (built in {build_t:.0f} s) ===", flush=True) + + # steady-state polls + totals, rcon_ts, payloads = [], [], [] + for _ in range(50): + t0 = time.perf_counter() + resp = rc.send_command("/sc obs_all_drain()") or "" + t1 = time.perf_counter() + epart, _, tpart = resp.partition("~") + client.apply_entity(epart) + client.apply_terrain(tpart) + _ = client.observation() + totals.append((time.perf_counter() - t0) * 1000) + rcon_ts.append((t1 - t0) * 1000) + payloads.append(len(resp)) + totals.sort() + print(f" steady poll: p50 {statistics.median(totals):6.1f} ms " + f"p95 {totals[47]:6.1f} max {totals[-1]:6.1f} " + f"(rcon p50 {statistics.median(rcon_ts):.1f} ms, " + f"payload p50 {statistics.median(payloads):.0f} B, " + f"p95 {sorted(payloads)[47]} B)", flush=True) + + ups = measure_ups(rc) + print(f" UPS at speed 10: {ups:.0f}", flush=True) + + # drift-detection latency: rotate 200 belts via script (no events) + n_rot = int(rc.send_command("""/sc local n = 0 +for _, b in ipairs(game.surfaces[1].find_entities_filtered{name = "transport-belt", force = "player"}) do + if n >= 200 then break end + b.direction = (b.direction == 4) and 8 or 4 + n = n + 1 +end +rcon.print(n)""".strip())) + t0 = time.perf_counter() + caught = 0 + while caught < n_rot and time.perf_counter() - t0 < 300: + resp = rc.send_command("/sc obs_diff_drain()") or "" + caught += sum(1 for r in resp.split(";") + if r[:1] == "u" and "transport-belt" in r) + client.apply_entity(resp) + time.sleep(0.25) + print(f" drift latency: {caught}/{n_rot} script-rotated belts " + f"surfaced in {time.perf_counter() - t0:.1f} s", flush=True) + + # cold attach: fresh client, full syncs + fresh = TensorClient(table_rows=STRESS_TABLE_ROWS) + t0 = time.perf_counter() + resp_e = rc.send_command("/sc obs_diff_full_sync()") or "" + resp_t = rc.send_command("/sc obs_terrain_full_sync()") or "" + t_rcon = (time.perf_counter() - t0) * 1000 + t0 = time.perf_counter() + fresh.apply_entity(resp_e) + fresh.apply_terrain(resp_t) + t_apply = (time.perf_counter() - t0) * 1000 + t0 = time.perf_counter() + fresh.rebuild() + t_rebuild = (time.perf_counter() - t0) * 1000 + print(f" cold attach: {(len(resp_e) + len(resp_t)) / 1024:.0f} KB, " + f"rcon {t_rcon:.0f} ms + apply {t_apply:.0f} ms; " + f"full rebuild {t_rebuild:.0f} ms; " + f"client match: {len(fresh.entities) == len(client.entities)}", flush=True) + client = fresh # continue with the freshly synced client + + # recenter cost: teleport player across the factory + ts = [] + for corner in ("{-240, -240}", "{240, -240}", "{240, 240}", "{0, 0}"): + rc.send_command( + "/sc local ch = game.surfaces[1].find_entities_filtered{type = \"character\"}[1] " + f"ch.teleport({corner}) rcon.print(1)") + t0 = time.perf_counter() + drain_apply(rc, client) + _ = client.observation() + ts.append((time.perf_counter() - t0) * 1000) + print(f" corner-teleport polls (forced recenter): " + f"{', '.join(f'{t:.0f}' for t in ts)} ms", flush=True) + assert not client.overflow, "buffer/table overflow during stress" + + # authoritative consistency at final scale. Pause the world first: on this + # map biters actively destroy the factory, so a live comparison lags by + # whatever died between the last drain and the scan. + rc.send_command("/sc game.tick_paused = true") + drain_apply(rc, client) + n_true = int(rc.send_command("""/sc local n = 0 +for _, e in ipairs(game.surfaces[1].find_entities_filtered{force = "player"}) do + if e.unit_number then n = n + 1 end +end +rcon.print(n)""".strip())) + rc.send_command("/sc game.tick_paused = false game.speed = 10") + print(f"\nfinal consistency (paused world): server={n_true} " + f"client={len(client.entities)} exact={n_true == len(client.entities)} " + f"table_slots={int(client.table_mask.sum())} overflow={client.overflow}", flush=True) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/benchmarks/benchmark_tensor_obs.py b/tests/benchmarks/benchmark_tensor_obs.py new file mode 100644 index 000000000..1da441046 --- /dev/null +++ b/tests/benchmarks/benchmark_tensor_obs.py @@ -0,0 +1,676 @@ +"""End-to-end observation tensor benchmark: RCON drain -> reconcile -> numpy. + +Builds an MLP-ready float32 tensor from the tiered diff protocol and measures +the full pipeline. The tensor is updated incrementally from diff records, so +steady-state cost is O(changes), not O(world). + +Observation views: [C, H, W] spatial grid of CELL-tile cells tracking the +player character (recentering when they leave a dead zone), a small global +feature vector including the exact player position, and an object-level +entity view of the VIEW_K entities nearest the player, nearest-first with +player-relative coordinates (the full absolute table stays available +internally at table_rows capacity for actions and consistency checks). + + channels 0-5 entity counts per type (furnace/belt/inserter/assembler/chest/other) + channel 6 status sum (working=1.0 scale) + channels 7-8 direction sin/cos sums + channel 9 energy (MJ) + channel 10 crafting progress (0-1) + channel 11 inventory fullness (log2 of total count) + channel 12 water tile count + channel 13 tree count + channel 14 rock/cliff count + channel 15 ore amount (bucket sum) + channel 16 enemy structure count + +All channels are additive so record application is subtract-old/add-new. + +Usage: python tests/benchmarks/benchmark_tensor_obs.py [--port 27099] +""" + +import argparse +import math +import statistics +import sys +import time +from pathlib import Path + +import numpy as np +from factorio_rcon import RCONClient + +sys.path.insert(0, str(Path(__file__).parent)) +from benchmark_tiered_obs import TieredClient # noqa: E402 + +GRID = 96 # cells per side +CELL = 3 # tiles per cell +HALF = GRID * CELL // 2 # window covers [center - HALF, center + HALF) tiles +RECENTER_DEADZONE = 24.0 # tiles the player may drift before the grid recenters +N_CHANNELS = 17 +ENTITY_CHANNEL = { + "stone-furnace": 0, + "transport-belt": 1, + "burner-inserter": 2, + "assembling-machine-1": 3, + "iron-chest": 4, +} +OTHER_CHANNEL = 5 +C_STATUS, C_SIN, C_COS, C_ENERGY, C_PROGRESS, C_INV = 6, 7, 8, 9, 10, 11 +C_WATER, C_TREE, C_ROCK, C_ORE, C_NEST = 12, 13, 14, 15, 16 +N_GLOBAL = 10 + +# Entity table: object-level view with exact positions and properties, +# mirroring the fields of the fle.env.entities object model. +F_X, F_Y = 0, 1 # exact tile coordinates +F_SIN, F_COS = 2, 3 # direction +F_TYPE = 4 # type id (session vocab) +F_STATUS = 5 +F_RECIPE = 6 # recipe id (0 = none) +F_ENERGY = 7 # exact joules +F_PROGRESS = 8 # exact crafting percent +F_HEALTH = 9 # exact hit points +F_TILE_W, F_TILE_H = 10, 11 # footprint (tile_dimensions) +F_DROP_DX, F_DROP_DY = 12, 13 # drop_position relative to entity (0,0 = none) +F_PICK_DX, F_PICK_DY = 14, 15 # pickup_position relative to entity +F_ELEC_ID = 16 # electric network id (0 = not connected) +F_TEMP = 17 # temperature +F_FLUID_ID, F_FLUID_AMT = 18, 19 # first fluidbox: fluid id, exact amount +F_INV_START = 20 # 8 slots x (item id, exact count), largest counts first +N_INV_SLOTS = 8 +F_INV_TOTAL = F_INV_START + 2 * N_INV_SLOTS # 36: total item count +F_INV_DISTINCT = F_INV_TOTAL + 1 # 37: number of distinct item stacks +TABLE_ROWS = 2048 +TABLE_FEATS = F_INV_DISTINCT + 1 +VIEW_K = 2048 # observation() returns the K entities nearest the player + + +class TensorClient(TieredClient): + """TieredClient that incrementally maintains MLP observation tensors. + + Two views of the same lossless client state: + grid (N_CHANNELS, GRID, GRID) - additive spatial context, cell-coarse + table (TABLE_ROWS, TABLE_FEATS) + mask - object-level view with EXACT + positions and per-entity properties; slot indices are stable for + an entity's lifetime (free-list allocation, no compaction), so a + policy can point at specific entities across steps. + """ + + def __init__(self, table_rows=TABLE_ROWS, view_k=VIEW_K): + super().__init__() + self.view_k = view_k + self.grid = np.zeros((N_CHANNELS, GRID, GRID), dtype=np.float32) + self.global_vec = np.zeros(N_GLOBAL, dtype=np.float32) + self.table = np.zeros((table_rows, TABLE_FEATS), dtype=np.float32) + self.table_mask = np.zeros(table_rows, dtype=np.float32) + self.view_units = [] # unit_number per view row, set by observation() + self._contrib = {} # unit_number -> (gy, gx, channel_values dict) + self._ore_contrib = {} # "x:y" -> (gy, gx, bucket) + self._water_contrib = {} # (cx, cy) -> list of (gy, gx, count) + self._slot_of = {} # unit_number -> table row + self._unit_at = {} # table row -> unit_number + self._free_slots = list(range(table_rows - 1, -1, -1)) + self._vocab = {} # "kind:name" -> stable int id (session-scoped, >0) + self.center_x = 0 # grid window center, CELL-quantized world coords + self.center_y = 0 + + def _cell_of(self, x, y): + gx = int((x - self.center_x + HALF) // CELL) + gy = int((y - self.center_y + HALF) // CELL) + if 0 <= gx < GRID and 0 <= gy < GRID: + return gx, gy + return None + + def _maybe_recenter(self): + """Recenter the grid on the player once they leave the dead zone. + Only grid contributions rebuild - table slots stay stable.""" + if (abs(self.player_x - self.center_x) <= RECENTER_DEADZONE + and abs(self.player_y - self.center_y) <= RECENTER_DEADZONE): + return + self.center_x = CELL * round(self.player_x / CELL) + self.center_y = CELL * round(self.player_y / CELL) + self._rebuild_grid() + + # -- entity rows --------------------------------------------------------- + + @staticmethod + def _parse_row(row): + """Parse an exact-valued row into a field dict. Items arrive as + I.:; positions in D/K tags are absolute.""" + fields = row.split(",") + p = { + "name": fields[0], + "x": float(fields[1]), + "y": float(fields[2]), + "direction": int(fields[3]), + "status": int(fields[4]), + "energy": 0.0, + "progress": 0.0, + "recipe": None, + "health": 0.0, + "tile_w": 0.0, + "tile_h": 0.0, + "drop": None, + "pickup": None, + "elec_id": 0.0, + "temp": 0.0, + "items": [], # (invidx, name, exact count) + "fluids": [], # (name, exact amount) + } + for f in fields[5:]: + tag = f[:1] + if tag == "E": + p["energy"] = float(f[1:]) + elif tag == "P": + p["progress"] = float(f[1:]) + elif tag == "R": + p["recipe"] = f[1:] + elif tag == "H": + p["health"] = float(f[1:]) + elif tag == "W": + w, h = f[1:].split(":") + p["tile_w"], p["tile_h"] = float(w), float(h) + elif tag == "D": + dx, dy = f[1:].split(":") + p["drop"] = (float(dx), float(dy)) + elif tag == "K": + kx, ky = f[1:].split(":") + p["pickup"] = (float(kx), float(ky)) + elif tag == "N": + p["elec_id"] = float(f[1:]) + elif tag == "T": + p["temp"] = float(f[1:]) + elif tag == "I": + slot, rest = f[1:].split(".", 1) + item, count = rest.rsplit(":", 1) + p["items"].append((int(slot), item, float(count))) + elif tag == "F": + fluid, amount = f[1:].rsplit(":", 1) + p["fluids"].append((fluid, float(amount))) + return p + + def _vid(self, kind, name): + key = kind + ":" + name + vid = self._vocab.get(key) + if vid is None: + vid = len(self._vocab) + 1 + self._vocab[key] = vid + return vid + + def _entity_contrib(self, row): + p = self._parse_row(row) + cell = self._cell_of(p["x"], p["y"]) + if cell is None: + return None + gx, gy = cell + angle = p["direction"] / 16.0 * 2.0 * math.pi + total_items = sum(c for _, _, c in p["items"]) + return gy, gx, { + ENTITY_CHANNEL.get(p["name"], OTHER_CHANNEL): 1.0, + C_STATUS: p["status"] / 10.0, + C_SIN: math.sin(angle), + C_COS: math.cos(angle), + C_ENERGY: p["energy"] / 1e6, + C_PROGRESS: p["progress"] / 100.0, + C_INV: math.log2(1.0 + total_items), + } + + # -- entity table -------------------------------------------------------- + + def _table_upsert(self, key, row): + p = self._parse_row(row) + slot = self._slot_of.get(key) + if slot is None: + if not self._free_slots: + self.overflow = True + return + slot = self._free_slots.pop() + self._slot_of[key] = slot + self._unit_at[slot] = key + angle = p["direction"] / 16.0 * 2.0 * math.pi + r = self.table[slot] + r[:] = 0.0 + r[F_X], r[F_Y] = p["x"], p["y"] # exact tile positions + r[F_SIN], r[F_COS] = math.sin(angle), math.cos(angle) + r[F_TYPE] = self._vid("type", p["name"]) + r[F_STATUS] = p["status"] + if p["recipe"]: + r[F_RECIPE] = self._vid("recipe", p["recipe"]) + r[F_ENERGY] = p["energy"] + r[F_PROGRESS] = p["progress"] + r[F_HEALTH] = p["health"] + r[F_TILE_W], r[F_TILE_H] = p["tile_w"], p["tile_h"] + if p["drop"]: + r[F_DROP_DX] = p["drop"][0] - p["x"] + r[F_DROP_DY] = p["drop"][1] - p["y"] + if p["pickup"]: + r[F_PICK_DX] = p["pickup"][0] - p["x"] + r[F_PICK_DY] = p["pickup"][1] - p["y"] + r[F_ELEC_ID] = p["elec_id"] + r[F_TEMP] = p["temp"] + if p["fluids"]: + r[F_FLUID_ID] = self._vid("fluid", p["fluids"][0][0]) + r[F_FLUID_AMT] = p["fluids"][0][1] + # merge duplicate items across inventories, then top-N by count + merged = {} + for _, item, count in p["items"]: + merged[item] = merged.get(item, 0.0) + count + ranked = sorted(merged.items(), key=lambda kv: -kv[1]) + for i, (item, count) in enumerate(ranked[:N_INV_SLOTS]): + r[F_INV_START + 2 * i] = self._vid("item", item) + r[F_INV_START + 2 * i + 1] = count + r[F_INV_TOTAL] = sum(merged.values()) + r[F_INV_DISTINCT] = len(merged) + self.table_mask[slot] = 1.0 + + def _table_remove(self, key): + slot = self._slot_of.pop(key, None) + if slot is not None: + self._unit_at.pop(slot, None) + self.table[slot] = 0.0 + self.table_mask[slot] = 0.0 + self._free_slots.append(slot) + + def _apply_contrib(self, contrib, sign): + gy, gx, values = contrib + for channel, v in values.items(): + self.grid[channel, gy, gx] += sign * v + + def apply_entity(self, resp): + if not resp: + return 0 + records = resp.split(";") + for rec in records: + tag = rec[:1] + if tag == "u": + key, row = rec[1:].split(",", 1) + old = self._contrib.pop(key, None) + if old is not None: + self._apply_contrib(old, -1.0) + contrib = self._entity_contrib(row) + if contrib is not None: + self._contrib[key] = contrib + self._apply_contrib(contrib, +1.0) + self._table_upsert(key, row) + self.entities[key] = row + elif tag == "r": + key = rec[1:] + old = self._contrib.pop(key, None) + if old is not None: + self._apply_contrib(old, -1.0) + self._table_remove(key) + self.entities.pop(key, None) + elif tag == "h": + parts = rec[1:].split(":") + self.tick = int(parts[0]) + self.research = None if parts[1] == "-" else parts[1] + self.research_pct = int(parts[2]) + if len(parts) >= 5: + self.player_x = float(parts[3]) + self.player_y = float(parts[4]) + self._maybe_recenter() + elif tag == "q": + self.techs_finished.append(rec[1:]) + elif tag == "!": + self.overflow = True + self._update_globals() + return len(records) + + # -- terrain records ----------------------------------------------------- + + def _apply_water_chunk(self, cx, cy, hexmask): + old = self._water_contrib.pop((cx, cy), None) + if old is not None: + for gy, gx, count in old: + self.grid[C_WATER, gy, gx] -= count + if not hexmask: + return + rows = np.array([int(hexmask[i * 8:(i + 1) * 8], 16) for i in range(32)], + dtype=np.uint32) + tile_mask = (rows[:, None] >> np.arange(32, dtype=np.uint32)[None, :]) & 1 + dy, dx = np.nonzero(tile_mask) + wx = cx * 32 + dx - self.center_x + HALF + wy = cy * 32 + dy - self.center_y + HALF + keep = (wx >= 0) & (wx < GRID * CELL) & (wy >= 0) & (wy < GRID * CELL) + if not keep.any(): + return + gx = wx[keep] // CELL + gy = wy[keep] // CELL + cells = {} + for a, b in zip(gy, gx): + cells[(int(a), int(b))] = cells.get((int(a), int(b)), 0) + 1 + contrib = [(a, b, c) for (a, b), c in cells.items()] + self._water_contrib[(cx, cy)] = contrib + for gy_, gx_, count in contrib: + self.grid[C_WATER, gy_, gx_] += count + + def _point_delta(self, key, channel, new_value): + """Set a point contribution on `channel`, replacing any previous one.""" + old = self._ore_contrib.pop(key, None) + if old is not None: + gy, gx, v = old + self.grid[channel, gy, gx] -= v + if new_value is None: + return + x, y = key.split(":") + cell = self._cell_of(float(x), float(y)) + if cell is None: + return + gx, gy = cell + self._ore_contrib[key] = (gy, gx, new_value) + self.grid[channel, gy, gx] += new_value + + def apply_terrain(self, resp): + if not resp: + return 0 + records = resp.split(";") + for rec in records: + tag = rec[:1] + if tag == "c": + cx, cy, hexmask = rec[1:].split(":", 2) + cx, cy = int(cx), int(cy) + self.water[(cx, cy)] = int(hexmask, 16) if hexmask else 0 + self._apply_water_chunk(cx, cy, hexmask) + elif tag == "o": + name, x, y, bucket = rec[1:].split(":") + key = f"{x}:{y}" + self.ores[key] = (name, int(bucket)) + self._point_delta(key, C_ORE, float(bucket)) + elif tag == "d": + key = rec[1:] + self.ores.pop(key, None) + self._point_delta(key, C_ORE, None) + elif tag == "t": + key = rec[1:] + if key not in self.trees: + self.trees.add(key) + cell = self._cell_of(*map(float, key.split(":"))) + if cell: + self.grid[C_TREE, cell[1], cell[0]] += 1 + elif tag == "x": + key = rec[1:] + cell = self._cell_of(*map(float, key.split(":"))) + if key in self.trees: + self.trees.discard(key) + if cell: + self.grid[C_TREE, cell[1], cell[0]] -= 1 + elif key in self.obstacles: + self.obstacles.discard(key) + if cell: + self.grid[C_ROCK, cell[1], cell[0]] -= 1 + elif tag == "k": + key = rec[1:] + if key not in self.obstacles: + self.obstacles.add(key) + cell = self._cell_of(*map(float, key.split(":"))) + if cell: + self.grid[C_ROCK, cell[1], cell[0]] += 1 + elif tag == "n": + key, name, x, y = rec[1:].split(",") + self.nests[key] = (name, float(x), float(y)) + cell = self._cell_of(float(x), float(y)) + if cell: + self.grid[C_NEST, cell[1], cell[0]] += 1 + elif tag == "m": + key = rec[1:] + nest = self.nests.pop(key, None) + if nest: + cell = self._cell_of(nest[1], nest[2]) + if cell: + self.grid[C_NEST, cell[1], cell[0]] -= 1 + elif tag == "!": + self.overflow = True + self._update_globals() + return len(records) + + def _update_globals(self): + g = self.global_vec + g[0] = self.tick / 1e6 + g[1] = self.research_pct / 100.0 + g[2] = len(self.entities) / 1e4 + g[3] = len(self.ores) / 1e4 + g[4] = len(self.trees) / 1e5 + g[5] = len(self.nests) / 100.0 + g[6] = len(self.techs_finished) / 100.0 + g[7] = 1.0 if self.overflow else 0.0 + g[8] = self.player_x + g[9] = self.player_y + + def observation(self): + """Observation views for an MLP/transformer policy: (flat grid, + global vec, K-nearest entity view, view mask). + + The view is a fixed (view_k, TABLE_FEATS) array holding the view_k + entities nearest the player, nearest-first, with player-RELATIVE + x/y. self.view_units gives the unit_number behind each view row + (distance ordering means view rows are not lifetime-stable), and + the full absolute table remains available via self.table.""" + k = self.view_k + view = np.zeros((k, TABLE_FEATS), dtype=np.float32) + view_mask = np.zeros(k, dtype=np.float32) + live_idx = np.nonzero(self.table_mask)[0] + if live_idx.size: + dx = self.table[live_idx, F_X] - self.player_x + dy = self.table[live_idx, F_Y] - self.player_y + d2 = dx * dx + dy * dy + if live_idx.size > k: + sel = np.argpartition(d2, k - 1)[:k] + sel = sel[np.argsort(d2[sel])] + else: + sel = np.argsort(d2) + chosen = live_idx[sel] + n = chosen.size + view[:n] = self.table[chosen] + view[:n, F_X] -= self.player_x + view[:n, F_Y] -= self.player_y + view_mask[:n] = 1.0 + self.view_units = [self._unit_at[int(s)] for s in chosen] + else: + self.view_units = [] + return self.grid.reshape(-1), self.global_vec, view, view_mask + + def _rebuild_grid(self): + """Rebuild grid contributions for the current window center. Table + slots are untouched, so slot indices stay stable across recenters.""" + self.grid[:] = 0.0 + self._contrib.clear() + self._ore_contrib.clear() + self._water_contrib.clear() + for key, row in self.entities.items(): + contrib = self._entity_contrib(row) + if contrib is not None: + self._contrib[key] = contrib + self._apply_contrib(contrib, +1.0) + for key, (_, bucket) in self.ores.items(): + self._point_delta(key, C_ORE, float(bucket)) + for key in self.trees: + cell = self._cell_of(*map(float, key.split(":"))) + if cell: + self.grid[C_TREE, cell[1], cell[0]] += 1 + for key in self.obstacles: + cell = self._cell_of(*map(float, key.split(":"))) + if cell: + self.grid[C_ROCK, cell[1], cell[0]] += 1 + for _, x, y in self.nests.values(): + cell = self._cell_of(x, y) + if cell: + self.grid[C_NEST, cell[1], cell[0]] += 1 + for (cx, cy), mask in self.water.items(): + hexmask = format(mask, "0256x") if mask else "" + self._apply_water_chunk(cx, cy, hexmask) + + def rebuild(self): + """Full tensor rebuild from client dicts (recovery path). + + Note: table slots are reallocated, so slot indices are NOT stable + across a full rebuild (they are stable under incremental updates + and grid recenters).""" + self._rebuild_grid() + self.table[:] = 0.0 + self.table_mask[:] = 0.0 + self._slot_of.clear() + self._unit_at.clear() + self._free_slots = list(range(self.table.shape[0] - 1, -1, -1)) + for key, row in sorted(self.entities.items()): + self._table_upsert(key, row) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--port", type=int, default=27099) + args = ap.parse_args() + + rc = RCONClient("localhost", args.port, "factorio") + rc.connect() + rc.send_command("/sc game.speed = 10") + client = TensorClient() + + flat, gvec, table, mask = client.observation() + total = flat.size + gvec.size + table.size + mask.size + total_kb = (flat.nbytes + gvec.nbytes + table.nbytes + mask.nbytes) / 1024 + print(f"tensor: grid {client.grid.shape} + global {gvec.shape} + " + f"table {table.shape} + mask {mask.shape} = {total} float32 ({total_kb:.0f} KB)\n") + + # --- 1. initial sync: terrain burst + entity snapshot -> tensor --- + t0 = time.perf_counter() + resp_t = rc.send_command("/sc obs_terrain_drain()") or "" + if resp_t.count(";") < 100: # burst already consumed: mid-episode attach + resp_t = rc.send_command("/sc obs_terrain_full_sync()") or "" + resp_e = rc.send_command("/sc obs_diff_full_sync()") or "" + t_rcon = (time.perf_counter() - t0) * 1000 + t0 = time.perf_counter() + n_t = client.apply_terrain(resp_t) + n_e = client.apply_entity(resp_e) + t_apply = (time.perf_counter() - t0) * 1000 + print(f"initial sync: {n_t + n_e} records ({(len(resp_t) + len(resp_e)) / 1024:.0f} KB); " + f"rcon {t_rcon:.0f} ms + apply-to-tensor {t_apply:.0f} ms", flush=True) + + # --- 2. spawn factory + make it hot --- + SPAWN = """ + local surface = game.surfaces[1] + local names = {"stone-furnace", "transport-belt", "burner-inserter", "assembling-machine-1", "iron-chest"} + local placed = 0 + for i = 0, 999 do + local e = surface.create_entity{name = names[(i % 5) + 1], + position = {-140 + (i % 70) * 4, -140 + math.floor(i / 70) * 4}, + force = "player", raise_built = true} + if e then placed = placed + 1 end + end + for _, f in ipairs(surface.find_entities_filtered{name = "stone-furnace", force = "player"}) do + f.get_inventory(defines.inventory.fuel).insert{name = "coal", count = 25} + f.get_inventory(defines.inventory.furnace_source).insert{name = "iron-ore", count = 50} + end + rcon.print(placed) + """ + spawned = rc.send_command("/sc " + SPAWN.strip()) + time.sleep(1.0) + resp = rc.send_command("/sc obs_all_drain()") or "" + epart, _, tpart = resp.partition("~") + t0 = time.perf_counter() + client.apply_entity(epart) + client.apply_terrain(tpart) + t = (time.perf_counter() - t0) * 1000 + print(f"spawned {spawned} + fueled furnaces -> burst applied to tensor in {t:.1f} ms") + + # --- 3. end-to-end steady-state: drain -> reconcile -> tensor --- + t_rcons, t_applies, sizes = [], [], [] + + def e2e_poll(): + t0 = time.perf_counter() + resp = rc.send_command("/sc obs_all_drain()") or "" + t1 = time.perf_counter() + epart, _, tpart = resp.partition("~") + client.apply_entity(epart) + client.apply_terrain(tpart) + _ = client.observation() + t2 = time.perf_counter() + t_rcons.append((t1 - t0) * 1000) + t_applies.append((t2 - t1) * 1000) + sizes.append(len(resp)) + return (t2 - t0) * 1000 + + for _ in range(3): + e2e_poll() + t_rcons.clear(); t_applies.clear(); sizes.clear() + totals = [e2e_poll() for _ in range(30)] + print(f"\nE2E poll (hot factory): mean {statistics.mean(totals):.1f} ms " + f"p50 {statistics.median(totals):.1f} min {min(totals):.1f} max {max(totals):.1f}" + f" -> {1000 / statistics.mean(totals):.0f} obs/s") + print(f" breakdown p50: rcon {statistics.median(t_rcons):.2f} ms, " + f"reconcile+tensor {statistics.median(t_applies):.3f} ms " + f"(payload p50 {statistics.median(sizes):.0f} B)") + + # --- 3b. egocentric tracking: stationary, walking, sprinting --- + rc.send_command("""/sc if #game.surfaces[1].find_entities_filtered{type = "character"} == 0 then + game.surfaces[1].create_entity{name = "character", position = {-100, -100}, force = "player", raise_built = true} +end rcon.print(1)""".strip()) + + def measure_moving(step_tiles, label, iters=30): + totals, recenters = [], 0 + last_center = (client.center_x, client.center_y) + for _ in range(iters): + if step_tiles: + rc.send_command( + "/sc local ch = game.surfaces[1].find_entities_filtered{type = \"character\"}[1] " + f"ch.teleport({{ch.position.x + {step_tiles}, ch.position.y}}) rcon.print(1)") + t0 = time.perf_counter() + resp = rc.send_command("/sc obs_all_drain()") or "" + epart, _, tpart = resp.partition("~") + client.apply_entity(epart) + client.apply_terrain(tpart) + _ = client.observation() + totals.append((time.perf_counter() - t0) * 1000) + center = (client.center_x, client.center_y) + if center != last_center: + recenters += 1 + last_center = center + print(f" {label:26s} mean {statistics.mean(totals):6.1f} ms " + f"p50 {statistics.median(totals):6.1f} max {max(totals):6.1f} " + f"({recenters} recenters/{iters} polls)", flush=True) + + print("\negocentric tracking (poll incl. drain + apply + observation()):") + measure_moving(0, "stationary player") + measure_moving(6, "walking (6 tiles/poll)") + measure_moving(60, "sprinting (60 tiles/poll)") + + ts = [] + for _ in range(50): + t0 = time.perf_counter() + _ = client.observation() + ts.append((time.perf_counter() - t0) * 1000) + print(f" observation() view alone mean {statistics.mean(ts):6.2f} ms " + f"p50 {statistics.median(ts):6.2f} (table copy + relativize)") + + # --- 4. full rebuild (recenter / recovery path) --- + ts = [] + for _ in range(10): + t0 = time.perf_counter() + client.rebuild() + ts.append((time.perf_counter() - t0) * 1000) + print(f"\nfull tensor rebuild: mean {statistics.mean(ts):.1f} ms p50 {statistics.median(ts):.1f} ms " + f"(from {len(client.entities)} entities, {len(client.ores)} ores, {len(client.trees)} trees)") + + # --- 5. sanity checks (bring the player back to the factory first) --- + rc.send_command("/sc local ch = game.surfaces[1].find_entities_filtered{type = \"character\"}[1] " + "if ch then ch.teleport({0, 0}) end rcon.print(1)") + time.sleep(0.2) + resp = rc.send_command("/sc obs_all_drain()") or "" + epart, _, tpart = resp.partition("~") + client.apply_entity(epart) + client.apply_terrain(tpart) + grid = client.grid + n_in_window = sum(1 for c in client._contrib.values()) + type_sum = float(grid[0:6].sum()) + print(f"\nsanity: entity-channel sum {type_sum:.0f} vs tracked-in-window {n_in_window}" + f" | ore cells {int((grid[C_ORE] > 0).sum())}, tree cells {int((grid[C_TREE] > 0).sum())}," + f" water cells {int((grid[C_WATER] > 0).sum())}") + print(f"nonzero grid values: {int((grid != 0).sum())}/{grid.size} " + f"({100 * (grid != 0).sum() / grid.size:.1f}%)") + n_slots = int(client.table_mask.sum()) + print(f"entity table: {n_slots}/{TABLE_ROWS} slots occupied, " + f"vocab size {len(client._vocab)}") + assert type_sum == n_in_window, "entity channel counts drifted from contrib map" + assert n_slots == len(client.entities), "table occupancy drifted from entity dict" + print("PASS: incremental tensor consistent") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/benchmarks/benchmark_tiered_obs.py b/tests/benchmarks/benchmark_tiered_obs.py new file mode 100644 index 000000000..0a414317b --- /dev/null +++ b/tests/benchmarks/benchmark_tiered_obs.py @@ -0,0 +1,286 @@ +"""Benchmark the tiered event-driven observation mod (open_world scenario). + +Requires a server running the scenario with observation_diff.lua loaded: + python tests/benchmarks/benchmark_tiered_obs.py [--port 27099] +""" + +import argparse +import statistics +import time + +from factorio_rcon import RCONClient + + +class TieredClient: + """Reconciles drained records into a full client-side game state.""" + + def __init__(self): + self.entities = {} # unit_number -> row str + self.water = {} # (cx, cy) -> 1024-bit int mask + self.ores = {} # "x:y" -> (name, bucket) + self.trees = set() # "x:y" + self.obstacles = set() # "x:y" rocks + cliffs + self.nests = {} # unit_number -> (name, x, y) + self.tick = 0 + self.research = None + self.research_pct = 0 + self.player_x = 0.0 + self.player_y = 0.0 + self.techs_finished = [] + self.overflow = False + + def apply_entity(self, resp): + if not resp: + return 0 + records = resp.split(";") + for rec in records: + tag = rec[:1] + if tag == "u": + key, row = rec[1:].split(",", 1) + self.entities[key] = row + elif tag == "r": + self.entities.pop(rec[1:], None) + elif tag == "h": + parts = rec[1:].split(":") + self.tick = int(parts[0]) + self.research = None if parts[1] == "-" else parts[1] + self.research_pct = int(parts[2]) + if len(parts) >= 5: + self.player_x = float(parts[3]) + self.player_y = float(parts[4]) + elif tag == "q": + self.techs_finished.append(rec[1:]) + elif tag == "!": + self.overflow = True + return len(records) + + def apply_terrain(self, resp): + if not resp: + return 0 + records = resp.split(";") + for rec in records: + tag = rec[:1] + if tag == "c": + cx, cy, hexmask = rec[1:].split(":", 2) + self.water[(int(cx), int(cy))] = int(hexmask, 16) if hexmask else 0 + elif tag == "o": + name, x, y, bucket = rec[1:].split(":") + self.ores[f"{x}:{y}"] = (name, int(bucket)) + elif tag == "d": + self.ores.pop(rec[1:], None) + elif tag == "t": + self.trees.add(rec[1:]) + elif tag == "x": + self.trees.discard(rec[1:]) + self.obstacles.discard(rec[1:]) + elif tag == "k": + self.obstacles.add(rec[1:]) + elif tag == "n": + key, name, x, y = rec[1:].split(",") + self.nests[key] = (name, float(x), float(y)) + elif tag == "m": + self.nests.pop(rec[1:], None) + elif tag == "!": + self.overflow = True + return len(records) + + +def bench(fn, iters=15, warmup=2): + for _ in range(warmup): + fn() + ts = [] + for _ in range(iters): + t0 = time.perf_counter() + fn() + ts.append((time.perf_counter() - t0) * 1000) + return statistics.mean(ts), statistics.median(ts), min(ts), max(ts) + + +def fmt(label, r, extra=""): + m, p50, mn, mx = r + print(f"{label:36s} mean {m:7.1f} ms p50 {p50:7.1f} min {mn:7.1f} max {mx:7.1f} {extra}", flush=True) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--port", type=int, default=27099) + args = ap.parse_args() + + rc = RCONClient("localhost", args.port, "factorio") + rc.connect() + rc.send_command("/sc game.speed = 10") + client = TieredClient() + + def drain_entities(): + return rc.send_command("/sc obs_diff_drain()") or "" + + def drain_terrain(): + return rc.send_command("/sc obs_terrain_drain()") or "" + + # --- 1. initial terrain burst from map generation --- + t0 = time.perf_counter() + resp = drain_terrain() + n = client.apply_terrain(resp) + t = (time.perf_counter() - t0) * 1000 + water_chunks = sum(1 for m in client.water.values() if m) + print(f"initial terrain drain: {n} records, {len(resp) / 1024:.0f} KB, {t:.0f} ms") + print(f" client terrain: {len(client.water)} chunks ({water_chunks} with water), " + f"{len(client.ores)} ore tiles, {len(client.trees)} trees, " + f"{len(client.obstacles)} rocks/cliffs, {len(client.nests)} enemy structures", flush=True) + + # --- 2. spawn factory (raise_built -> events) --- + SPAWN = """ + local surface = game.surfaces[1] + local names = {"stone-furnace", "transport-belt", "burner-inserter", "assembling-machine-1", "iron-chest"} + local placed = 0 + for i = 0, 999 do + local e = surface.create_entity{name = names[(i % 5) + 1], + position = {-140 + (i % 70) * 4, -140 + math.floor(i / 70) * 4}, + force = "player", raise_built = true} + if e then placed = placed + 1 end + end + rcon.print(placed) + """ + spawned = rc.send_command("/sc " + SPAWN.strip()) + resp = drain_entities() + client.apply_entity(resp) + print(f"\nspawned {spawned} entities -> drain {len(resp) / 1024:.1f} KB, client tracks {len(client.entities)}") + + # --- 3. hot factory: fuel + feed every furnace, machines run --- + FUEL = """ + local n = 0 + for _, f in ipairs(game.surfaces[1].find_entities_filtered{name = "stone-furnace", force = "player"}) do + f.get_inventory(defines.inventory.fuel).insert{name = "coal", count = 25} + f.get_inventory(defines.inventory.furnace_source).insert{name = "iron-ore", count = 50} + n = n + 1 + end + rcon.print(n) + """ + n_fueled = rc.send_command("/sc " + FUEL.strip()) + time.sleep(1.0) + client.apply_entity(drain_entities()) # absorb the initial burst of status changes + print(f"\nfueled {n_fueled} furnaces (smelting at speed 10). steady-state polls:") + sizes, recs = [], [] + + def hot_poll(): + resp = drain_entities() + sizes.append(len(resp)) + recs.append(client.apply_entity(resp)) + + fmt("hot-factory entity drain", bench(hot_poll, iters=30), + f"(payload p50 {statistics.median(sizes):.0f} B, records p50 {statistics.median(recs):.0f})") + + # --- 4. chunk streaming: generate 121 new chunks mid-episode --- + t0 = time.perf_counter() + rc.send_command("/sc game.surfaces[1].request_to_generate_chunks({2000, 2000}, 5) " + "game.surfaces[1].force_generate_chunk_requests() rcon.print(1)") + gen_t = (time.perf_counter() - t0) * 1000 + t0 = time.perf_counter() + resp = drain_terrain() + n = client.apply_terrain(resp) + t = (time.perf_counter() - t0) * 1000 + print(f"\nnew-chunk streaming: generation {gen_t:.0f} ms; drain {n} records, " + f"{len(resp) / 1024:.1f} KB, {t:.1f} ms -> client now {len(client.water)} chunks", flush=True) + + # --- 5. resource depletion: eventless amount drift --- + DEPLETE = """ + local n = 0 + for _, o in ipairs(game.surfaces[1].find_entities_filtered{type = "resource", limit = 400}) do + if o.amount > 600 then + o.amount = o.amount - 500 + n = n + 1 + end + end + rcon.print(n) + """ + n_dep = int(rc.send_command("/sc " + DEPLETE.strip())) + t0 = time.perf_counter() + caught = 0 + while caught < n_dep and time.perf_counter() - t0 < 60: + resp = drain_terrain() + caught += sum(1 for r in resp.split(";") if r[:1] == "o") + client.apply_terrain(resp) + time.sleep(0.1) + print(f"\nresource reconciler: {caught}/{n_dep} depleted tiles surfaced in {time.perf_counter() - t0:.1f} s") + + # --- 6. tree removal events --- + CHOP = """ + local n = 0 + for _, t in ipairs(game.surfaces[1].find_entities_filtered{type = "tree", limit = 20}) do + t.destroy{raise_destroy = true} + n = n + 1 + end + rcon.print(n) + """ + n_chop = rc.send_command("/sc " + CHOP.strip()) + resp = drain_terrain() + n_x = sum(1 for r in resp.split(";") if r[:1] == "x") + client.apply_terrain(resp) + print(f"tree chop: {n_chop} destroyed -> {n_x} removal records in next drain") + + # --- 7. combined full-tier poll: one RCON call for both buffers --- + sizes.clear() + + def full_poll(): + resp = rc.send_command("/sc obs_all_drain()") or "" + sizes.append(len(resp)) + epart, _, tpart = resp.partition("~") + client.apply_entity(epart) + client.apply_terrain(tpart) + + r = bench(full_poll, iters=30) + fmt("combined poll (single call)", r, + f"(payload p50 {statistics.median(sizes):.0f} B) -> {1000 / r[0]:.0f} obs/s") + + # --- 8. effective UPS under mod load --- + t0 = time.perf_counter() + tick0 = int(rc.send_command("/sc rcon.print(game.tick)")) + time.sleep(5) + tick1 = int(rc.send_command("/sc rcon.print(game.tick)")) + ups = (tick1 - tick0) / (time.perf_counter() - t0) + print(f"\neffective UPS at speed 10: {ups:.0f} (target 600)") + + # --- 9. consistency checks --- + SCAN = """ + local out = {} + local n = 0 + for _, e in ipairs(game.surfaces[1].find_entities_filtered{force = "player"}) do + if e.unit_number then + n = n + 1 + out[n] = e.unit_number .. "|" .. e.name .. "|" .. e.position.x .. "," .. e.position.y + end + end + rcon.print(table.concat(out, ";")) + """ + time.sleep(1) + client.apply_entity(drain_entities()) + resp = rc.send_command("/sc " + SCAN.strip()) or "" + truth = {} + for rec in resp.split(";"): + if rec: + key, name, pos = rec.split("|") + truth[key] = (name, pos) + mine = {} + for key, row in client.entities.items(): + f = row.split(",") + mine[key] = (f[0], f[1] + "," + f[2]) + missing = set(truth) - set(mine) + extra = set(mine) - set(truth) + mismatch = {k for k in set(truth) & set(mine) if truth[k] != mine[k]} + print(f"entity consistency: server={len(truth)} client={len(mine)} " + f"missing={len(missing)} extra={len(extra)} mismatch={len(mismatch)}") + + ORE_CHECK = """ + local total = 0 + for _, o in ipairs(game.surfaces[1].find_entities_filtered{type = "resource"}) do + total = total + 1 + end + rcon.print(total) + """ + n_true = int(rc.send_command("/sc " + ORE_CHECK.strip())) + print(f"ore consistency: server={n_true} tiles, client={len(client.ores)}") + print(f"overflow flagged: {client.overflow}") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/tests/observation_diff/conftest.py b/tests/observation_diff/conftest.py new file mode 100644 index 000000000..64de54cd6 --- /dev/null +++ b/tests/observation_diff/conftest.py @@ -0,0 +1,118 @@ +"""Fixtures for tiered-observation-protocol tests. + +Launches a dedicated Factorio container on a random free port with the +open_world scenario from this checkout, and removes it afterward. It never +connects to (or discovers) any already-running FLE cluster server. +""" + +import platform +import shutil +import socket +import subprocess +import time +from pathlib import Path + +import pytest +from factorio_rcon import RCONClient + +REPO_ROOT = Path(__file__).resolve().parents[2] +IMAGE = "factoriotools/factorio:2.0.73" +RCON_PASSWORD = "factorio" +STARTUP_TIMEOUT = 300 + + +@pytest.fixture(autouse=True) +def _reset_between_tests(): + """Shadow the repo-level autouse fixture that connects to the shared FLE + cluster; these tests run exclusively against their own container.""" + yield + + +def _free_port(): + with socket.socket() as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _docker_ok(): + return ( + shutil.which("docker") is not None + and subprocess.run( + ["docker", "info"], capture_output=True, timeout=30 + ).returncode + == 0 + ) + + +@pytest.fixture(scope="session") +def tiered_rcon(): + """RCON client for a dedicated, isolated tiered-observation server.""" + if not _docker_ok(): + pytest.skip("docker is not available") + + port = _free_port() + name = f"fle-test-tiered-{port}" + cluster = REPO_ROOT / "fle" / "cluster" + emulator = "/bin/box64 " if platform.machine() in ("arm64", "aarch64") else "" + factorio_cmd = ( + "rm -rf /opt/factorio/data/elevated-rails /opt/factorio/data/quality " + "/opt/factorio/data/space-age && " + f"exec {emulator}/opt/factorio/bin/x64/factorio " + "--start-server-load-scenario open_world --port 34197 " + "--server-settings /opt/factorio/config/server-settings.json " + "--map-gen-settings /opt/factorio/config/map-gen-settings.json " + "--map-settings /opt/factorio/config/map-settings.json " + "--rcon-port 27015 " + f'--rcon-password "{RCON_PASSWORD}" ' + "--server-adminlist /opt/factorio/config/server-adminlist.json " + "--mod-directory /opt/factorio/mods --map-gen-seed 44340" + ) + run = subprocess.run( + [ + "docker", "run", "-d", "--name", name, + "-p", f"127.0.0.1:{port}:27015", + "-v", f"{cluster / 'scenarios'}:/opt/factorio/scenarios", + "-v", f"{cluster / 'config'}:/opt/factorio/config", + "-v", f"{cluster / 'mods'}:/opt/factorio/mods", + "--entrypoint", "/bin/sh", + IMAGE, "-c", factorio_cmd, + ], + capture_output=True, + text=True, + ) + if run.returncode != 0: + pytest.skip(f"could not start factorio container: {run.stderr.strip()}") + + rc = None + try: + deadline = time.time() + STARTUP_TIMEOUT + last_error = None + while time.time() < deadline: + try: + rc = RCONClient("127.0.0.1", port, RCON_PASSWORD) + rc.connect() + rc.send_command("/sc rcon.print(1)") + break + except Exception as e: # noqa: BLE001 - retry until server is up + last_error = e + rc = None + time.sleep(3) + if rc is None: + logs = subprocess.run( + ["docker", "logs", "--tail", "15", name], + capture_output=True, text=True, + ).stdout + raise RuntimeError( + f"factorio server did not come up in {STARTUP_TIMEOUT}s: " + f"{last_error}\n{logs}" + ) + + loaded = rc.send_command("/sc rcon.print(type(obs_diff_drain))") + assert loaded == "function", ( + "observation_diff.lua not loaded by the open_world scenario " + f"(type(obs_diff_drain) == {loaded!r})" + ) + rc.send_command("/sc game.speed = 10") + yield rc + finally: + subprocess.run(["docker", "rm", "-f", name], capture_output=True) \ No newline at end of file diff --git a/tests/observation_diff/test_tiered_protocol.py b/tests/observation_diff/test_tiered_protocol.py new file mode 100644 index 000000000..c5052a4e4 --- /dev/null +++ b/tests/observation_diff/test_tiered_protocol.py @@ -0,0 +1,492 @@ +"""Tests for the tiered event-driven observation protocol (observation_diff.lua). + +Runs against a dedicated container provided by the tiered_rcon fixture; see +conftest.py. Each test uses its own map region so state does not interfere. +""" + +import sys +import time +from pathlib import Path + +import numpy as np +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "benchmarks")) +from benchmark_tensor_obs import ( # noqa: E402 + F_DROP_DX, + F_DROP_DY, + F_HEALTH, + F_INV_DISTINCT, + F_INV_START, + F_INV_TOTAL, + F_PICK_DX, + F_PICK_DY, + F_TILE_H, + F_TILE_W, + F_TYPE, + F_X, + F_Y, + N_INV_SLOTS, + TensorClient, +) +from benchmark_tiered_obs import TieredClient # noqa: E402 + + +def drain_entities(rc): + return rc.send_command("/sc obs_diff_drain()") or "" + + +def drain_terrain(rc): + return rc.send_command("/sc obs_terrain_drain()") or "" + + +def drain_all(rc): + resp = rc.send_command("/sc obs_all_drain()") or "" + epart, _, tpart = resp.partition("~") + return epart, tpart + + +def absorb(rc, client): + """Drain and apply everything pending.""" + epart, tpart = drain_all(rc) + client.apply_entity(epart) + client.apply_terrain(tpart) + + +def spawn(rc, x0, y0, n, name="iron-chest", raise_built=True, spacing=2): + flag = "true" if raise_built else "false" + lua = f""" + local placed = 0 + for i = 0, {n - 1} do + local e = game.surfaces[1].create_entity{{name = "{name}", + position = {{{x0} + (i % 10) * {spacing}, {y0} + math.floor(i / 10) * {spacing}}}, + force = "player", raise_built = {flag}}} + if e then placed = placed + 1 end + end + rcon.print(placed) + """ + return int(rc.send_command("/sc " + lua.strip())) + + +def poll_until(rc, client, predicate, timeout=60): + deadline = time.time() + timeout + while time.time() < deadline: + absorb(rc, client) + if predicate(): + return True + time.sleep(0.2) + return False + + +@pytest.fixture() +def client(tiered_rcon): + """A fresh client, synced to current server state.""" + c = TieredClient() + resp = tiered_rcon.send_command("/sc obs_terrain_full_sync()") or "" + c.apply_terrain(resp) + resp = tiered_rcon.send_command("/sc obs_diff_full_sync()") or "" + c.apply_entity(resp) + return c + + +def test_rcon_functions_exist(tiered_rcon): + for fn in ("obs_diff_drain", "obs_terrain_drain", "obs_all_drain", + "obs_diff_full_sync", "obs_terrain_full_sync"): + assert tiered_rcon.send_command(f"/sc rcon.print(type({fn}))") == "function" + + +def test_initial_terrain_state(client): + """Map-gen chunk events (or full sync) populate the terrain tiers.""" + assert len(client.water) > 50, "expected many generated chunks" + assert len(client.ores) > 100, "expected ore tiles on the starting map" + assert len(client.trees) > 100, "expected trees on the starting map" + + +def test_header_reports_advancing_tick(tiered_rcon, client): + absorb(tiered_rcon, client) + t1 = client.tick + time.sleep(0.5) + absorb(tiered_rcon, client) + assert client.tick > t1 + assert 0 <= client.research_pct <= 100 + + +def test_build_events_upsert(tiered_rcon, client): + before = len(client.entities) + placed = spawn(tiered_rcon, 100, 100, 20) + assert placed == 20 + assert poll_until(tiered_rcon, client, lambda: len(client.entities) == before + 20) + row = next(r for r in client.entities.values() if r.startswith("iron-chest,1")) + name, x, y, direction, status = row.split(",")[:5] + assert name == "iron-chest" + assert 100 <= float(x) <= 120 + + +def test_destroy_events_remove(tiered_rcon, client): + spawn(tiered_rcon, 140, 100, 10) + absorb(tiered_rcon, client) + before = len(client.entities) + n = int(tiered_rcon.send_command("""/sc local n = 0 +for _, e in ipairs(game.surfaces[1].find_entities_filtered{name = "iron-chest", + force = "player", area = {{139, 99}, {160, 121}}}) do + e.destroy{raise_destroy = true} + n = n + 1 +end +rcon.print(n)""".strip())) + assert n == 10 + assert poll_until(tiered_rcon, client, lambda: len(client.entities) == before - 10) + + +def test_client_matches_authoritative_scan(tiered_rcon, client): + spawn(tiered_rcon, 180, 100, 15, name="stone-furnace") + time.sleep(0.5) + absorb(tiered_rcon, client) + resp = tiered_rcon.send_command("""/sc local out = {} +for _, e in ipairs(game.surfaces[1].find_entities_filtered{force = "player"}) do + if e.unit_number then + out[#out + 1] = e.unit_number .. "|" .. e.name .. "|" .. e.position.x .. "," .. e.position.y + end +end +rcon.print(table.concat(out, ";"))""".strip()) or "" + truth = {} + for rec in resp.split(";"): + if rec: + key, name, pos = rec.split("|") + truth[key] = (name, pos) + mine = {} + for key, row in client.entities.items(): + f = row.split(",") + mine[key] = (f[0], f[1] + "," + f[2]) + assert mine == truth + + +def test_eventless_mutation_reconciled(tiered_rcon, client): + """Script-set direction fires no event; the slice reconciler catches it.""" + spawn(tiered_rcon, 220, 100, 5, name="transport-belt") + absorb(tiered_rcon, client) + tiered_rcon.send_command("""/sc for _, b in ipairs(game.surfaces[1].find_entities_filtered{ + name = "transport-belt", force = "player", area = {{219, 99}, {240, 121}}}) do + b.direction = 4 +end rcon.print(1)""".strip()) + + def rotated(): + rows = [r for r in client.entities.values() + if r.startswith("transport-belt,2")] + return len(rows) == 5 and all(r.split(",")[3] == "4" for r in rows) + + assert poll_until(tiered_rcon, client, rotated), "reconciler missed direction change" + + +def test_eventless_creation_discovered(tiered_rcon, client): + """create_entity without raise_built fires no event; the discovery sweep + (every DISCOVERY_EVERY reconcile passes) must still find it.""" + before = len(client.entities) + placed = spawn(tiered_rcon, 260, 100, 8, raise_built=False) + assert placed == 8 + assert poll_until( + tiered_rcon, client, lambda: len(client.entities) >= before + 8, timeout=120 + ), "discovery sweep missed eventless entities" + + +def test_hot_field_quantization_stays_sparse(tiered_rcon, client): + """A running furnace changes state every tick, but quantized rows should + only emit records on bucket transitions.""" + spawn(tiered_rcon, 300, 100, 5, name="stone-furnace", spacing=3) + tiered_rcon.send_command("""/sc for _, f in ipairs(game.surfaces[1].find_entities_filtered{ + name = "stone-furnace", force = "player", area = {{299, 99}, {320, 121}}}) do + f.get_inventory(defines.inventory.fuel).insert{name = "coal", count = 20} + f.get_inventory(defines.inventory.furnace_source).insert{name = "iron-ore", count = 40} +end rcon.print(1)""".strip()) + time.sleep(1.0) + absorb(tiered_rcon, client) # absorb the ignition burst + + # a fueled furnace row carries quantized hot fields + hot = [r for r in client.entities.values() + if r.startswith("stone-furnace") and ".coal:" in r] + assert hot, "expected fueled furnace rows with inventory contents" + + total_records = 0 + for _ in range(10): + epart, tpart = drain_all(tiered_rcon) + total_records += sum(1 for r in epart.split(";") if r[:1] == "u") + client.apply_entity(epart) + client.apply_terrain(tpart) + time.sleep(0.1) + # 5 furnaces smelting at speed 10 for ~1s: without quantization this would + # be hundreds of updates; bucketed rows should emit only a handful. + assert total_records < 50, f"hot rows not sparse: {total_records} upserts" + + +def test_resource_depletion_reconciled(tiered_rcon, client): + changed = int(tiered_rcon.send_command("""/sc local n = 0 +for _, o in ipairs(game.surfaces[1].find_entities_filtered{type = "resource", limit = 50}) do + if o.amount > 600 then o.amount = o.amount - 500 n = n + 1 end +end +rcon.print(n)""".strip())) + assert changed > 0, "test map had no depletable ore" + seen = {"n": 0} + + def caught(): + return seen["n"] >= changed + + deadline = time.time() + 60 + while time.time() < deadline and not caught(): + _, tpart = drain_all(tiered_rcon) + seen["n"] += sum(1 for r in tpart.split(";") if r[:1] == "o") + client.apply_terrain(tpart) + time.sleep(0.2) + assert caught(), f"only {seen['n']}/{changed} depletion updates surfaced" + + +def test_tree_removal_events(tiered_rcon, client): + before = len(client.trees) + assert before > 0 + n = int(tiered_rcon.send_command("""/sc local n = 0 +for _, t in ipairs(game.surfaces[1].find_entities_filtered{type = "tree", limit = 10}) do + t.destroy{raise_destroy = true} + n = n + 1 +end +rcon.print(n)""".strip())) + assert n == 10 + assert poll_until(tiered_rcon, client, lambda: len(client.trees) == before - 10) + + +def test_new_chunks_stream_to_client(tiered_rcon, client): + before = len(client.water) + tiered_rcon.send_command( + "/sc game.surfaces[1].request_to_generate_chunks({3000, 3000}, 3) " + "game.surfaces[1].force_generate_chunk_requests() rcon.print(1)") + assert poll_until(tiered_rcon, client, lambda: len(client.water) > before), \ + "newly generated chunks never reached the client" + + +def test_full_sync_resets_consistently(tiered_rcon, client): + """After arbitrary churn, a fresh client full-syncs to identical state.""" + spawn(tiered_rcon, 340, 100, 12) + absorb(tiered_rcon, client) + fresh = TieredClient() + fresh.apply_entity(tiered_rcon.send_command("/sc obs_diff_full_sync()") or "") + assert set(fresh.entities) == set(client.entities) + # rows may differ only in flickering status; compare structural fields + for key in fresh.entities: + assert fresh.entities[key].split(",")[:4] == client.entities[key].split(",")[:4] + + +def test_tensor_incremental_matches_rebuild(tiered_rcon): + """Incrementally-maintained grid equals a from-scratch rebuild after churn.""" + c = TensorClient() + c.apply_terrain(tiered_rcon.send_command("/sc obs_terrain_full_sync()") or "") + c.apply_entity(tiered_rcon.send_command("/sc obs_diff_full_sync()") or "") + spawn(tiered_rcon, 380, 100, 10) + time.sleep(0.5) + absorb(tiered_rcon, c) + incremental = c.grid.copy() + c.rebuild() + np.testing.assert_allclose(incremental, c.grid, atol=1e-4) + n_tracked = len(c._contrib) + assert float(c.grid[0:6].sum()) == pytest.approx(n_tracked) + +def _tensor_client(rc): + c = TensorClient() + c.apply_terrain(rc.send_command("/sc obs_terrain_full_sync()") or "") + c.apply_entity(rc.send_command("/sc obs_diff_full_sync()") or "") + return c + + +def test_entity_table_exact_position_and_properties(tiered_rcon): + """The table row carries the exact server position and exact item counts + (wire rows transmit exact values; only the change trigger is bucketed).""" + c = _tensor_client(tiered_rcon) + unit = int(tiered_rcon.send_command("""/sc local e = game.surfaces[1].create_entity{ + name = "iron-chest", position = {420, 100}, force = "player", raise_built = true} +e.get_inventory(defines.inventory.chest).insert{name = "iron-plate", count = 37} +rcon.print(e.unit_number .. "," .. e.position.x .. "," .. e.position.y)""".strip()).split(",")[0]) + truth = tiered_rcon.send_command(f"""/sc for _, e in ipairs(game.surfaces[1].find_entities_filtered{{ + name = "iron-chest", force = "player", area = {{{{419, 99}}, {{422, 102}}}}}}) do + if e.unit_number == {unit} then rcon.print(e.position.x .. "," .. e.position.y) end +end""".strip()) + tx, ty = map(float, truth.split(",")) + + # The build event fires during create_entity, BEFORE the insert, so the + # first row snapshot has an empty chest; the reconciler must then pick up + # the eventless insert. Poll until the exact count lands. + def has_exact_count(): + slot = c._slot_of.get(str(unit)) + return slot is not None and c.table[slot][F_INV_TOTAL] == 37.0 + + assert poll_until(tiered_rcon, c, has_exact_count), \ + "reconciler never delivered the exact inventory count" + row = c.table[c._slot_of[str(unit)]] + assert (row[F_X], row[F_Y]) == (tx, ty), "table position is not exact" + assert row[F_INV_START + 1] == 37.0 # top inventory slot: exact count + assert row[F_HEALTH] > 0.0 + assert (row[F_TILE_W], row[F_TILE_H]) == (1.0, 1.0) + assert c.table_mask[c._slot_of[str(unit)]] == 1.0 + + +def test_entity_table_slot_stability(tiered_rcon): + """Slots are stable for an entity's lifetime under incremental updates: + removing one entity must not move any other entity's row.""" + c = _tensor_client(tiered_rcon) + spawn(tiered_rcon, 460, 100, 3, spacing=4) + absorb(tiered_rcon, c) + def in_region(row): + f = row.split(",") + return f[0] == "iron-chest" and 459 <= float(f[1]) <= 470 and 99 <= float(f[2]) <= 102 + + keys = sorted(k for k, r in c.entities.items() if in_region(r)) + assert len(keys) == 3 + slots_before = {k: c._slot_of[k] for k in keys} + victim = keys[1] + vx, vy = c.entities[victim].split(",")[1:3] + tiered_rcon.send_command(f"""/sc for _, e in ipairs(game.surfaces[1].find_entities_filtered{{ + name = "iron-chest", position = {{{vx}, {vy}}}, radius = 0.5}}) do + e.destroy{{raise_destroy = true}} +end rcon.print(1)""".strip()) + assert poll_until(tiered_rcon, c, lambda: victim not in c._slot_of) + assert c.table_mask[slots_before[victim]] == 0.0 + for k in (keys[0], keys[2]): + assert c._slot_of[k] == slots_before[k], "survivor slot moved" + assert c.table_mask[slots_before[k]] == 1.0 + + +def test_entity_table_tracks_rebuild_and_dict(tiered_rcon): + """Occupancy always equals the entity dict; a rebuild reproduces the same + set of (position, type) rows even though slot assignment may differ.""" + c = _tensor_client(tiered_rcon) + spawn(tiered_rcon, 500, 100, 6) + absorb(tiered_rcon, c) + assert int(c.table_mask.sum()) == len(c.entities) + live = c.table[c.table_mask > 0] + rows_before = {tuple(r[[F_X, F_Y, F_TYPE]]) for r in live} + c.rebuild() + assert int(c.table_mask.sum()) == len(c.entities) + live = c.table[c.table_mask > 0] + assert {tuple(r[[F_X, F_Y, F_TYPE]]) for r in live} == rows_before + + +def test_entity_table_interlink_fields(tiered_rcon): + """Inserters expose exact drop/pickup positions and footprints - the + fields needed to build interlinked factories.""" + c = _tensor_client(tiered_rcon) + truth = tiered_rcon.send_command("""/sc local e = game.surfaces[1].create_entity{ + name = "burner-inserter", position = {540, 100}, direction = defines.direction.east, + force = "player", raise_built = true} +rcon.print(e.unit_number .. "|" .. e.position.x .. "," .. e.position.y + .. "|" .. e.drop_position.x .. "," .. e.drop_position.y + .. "|" .. e.pickup_position.x .. "," .. e.pickup_position.y)""".strip()) + unit, epos, dpos, kpos = truth.split("|") + ex, ey = map(float, epos.split(",")) + dx, dy = map(float, dpos.split(",")) + kx, ky = map(float, kpos.split(",")) + assert poll_until(tiered_rcon, c, lambda: unit in c._slot_of) + row = c.table[c._slot_of[unit]] + assert (row[F_X], row[F_Y]) == (ex, ey) + assert row[F_X] + row[F_DROP_DX] == pytest.approx(dx) + assert row[F_Y] + row[F_DROP_DY] == pytest.approx(dy) + assert row[F_X] + row[F_PICK_DX] == pytest.approx(kx) + assert row[F_Y] + row[F_PICK_DY] == pytest.approx(ky) + assert (row[F_DROP_DX], row[F_DROP_DY]) != (0.0, 0.0) + assert (row[F_TILE_W], row[F_TILE_H]) == (1.0, 1.0) + + +def test_entity_table_top8_inventory(tiered_rcon): + """9 distinct item types in one chest: the 8 largest stacks fill the + slots exactly, the smallest is dropped, totals count everything.""" + c = _tensor_client(tiered_rcon) + unit = tiered_rcon.send_command("""/sc local e = game.surfaces[1].create_entity{ + name = "iron-chest", position = {580, 100}, force = "player", raise_built = true} +local inv = e.get_inventory(defines.inventory.chest) +local names = {"iron-plate", "copper-plate", "coal", "stone", "iron-ore", + "copper-ore", "iron-gear-wheel", "copper-cable", "stone-brick"} +for i, name in ipairs(names) do + inv.insert{name = name, count = 10 * (10 - i)} -- 90, 80, ..., 10 +end +rcon.print(e.unit_number)""".strip()) + + def full_inventory(): + slot = c._slot_of.get(unit) + return slot is not None and c.table[slot][F_INV_DISTINCT] == 9.0 + + assert poll_until(tiered_rcon, c, full_inventory) + row = c.table[c._slot_of[unit]] + counts = [row[F_INV_START + 2 * i + 1] for i in range(N_INV_SLOTS)] + ids = [row[F_INV_START + 2 * i] for i in range(N_INV_SLOTS)] + assert counts == [90.0, 80.0, 70.0, 60.0, 50.0, 40.0, 30.0, 20.0] + assert len(set(ids)) == N_INV_SLOTS and all(i > 0 for i in ids) + assert row[F_INV_TOTAL] == float(sum(range(10, 100, 10))) # all 9 stacks + assert row[F_INV_DISTINCT] == 9.0 + + +def test_egocentric_tracking(tiered_rcon): + """The grid window and table view track the player character: the header + carries its exact position, the grid recenters onto it, and observation() + reports player-relative coordinates.""" + c = _tensor_client(tiered_rcon) + truth = tiered_rcon.send_command("""/sc local ch = game.surfaces[1].create_entity{ + name = "character", position = {800, 100}, force = "player", raise_built = true} +local chest = game.surfaces[1].create_entity{ + name = "iron-chest", position = {810, 100}, force = "player", raise_built = true} +rcon.print(ch.position.x .. "," .. ch.position.y .. "|" .. chest.unit_number + .. "|" .. chest.position.x .. "," .. chest.position.y)""".strip()) + chpos, chest_unit, chestpos = truth.split("|") + chx, chy = map(float, chpos.split(",")) + cx, cy = map(float, chestpos.split(",")) + try: + assert poll_until(tiered_rcon, c, lambda: c.player_x == chx and c.player_y == chy), \ + "header never reported the character position" + # grid recentered onto the character + assert abs(c.center_x - chx) <= 2 and abs(c.center_y - chy) <= 2 + # chest at x=810 is far outside an origin-centered window but must be + # in the egocentric grid (channel 4 = iron-chest) + assert poll_until(tiered_rcon, c, lambda: chest_unit in c._slot_of) + assert float(c.grid[4].sum()) >= 1.0 + # table view is player-relative; internal table stays absolute + _, gvec, view, _ = c.observation() + slot = c._slot_of[chest_unit] + assert (c.table[slot][F_X], c.table[slot][F_Y]) == (cx, cy) + vidx = c.view_units.index(chest_unit) + assert view[vidx][F_X] == pytest.approx(cx - chx) + assert view[vidx][F_Y] == pytest.approx(cy - chy) + assert (gvec[8], gvec[9]) == (chx, chy) + finally: + # remove the character so earlier-region tests behave if re-run + tiered_rcon.send_command("""/sc for _, e in ipairs(game.surfaces[1].find_entities_filtered{ + type = "character"}) do e.destroy{raise_destroy = true} end +storage.obs_char = nil rcon.print(1)""".strip()) + + +def test_k_nearest_view(tiered_rcon): + """observation() returns the view_k entities nearest the player, + nearest-first, dropping distant ones - against a live server.""" + c = TensorClient(view_k=5) + c.apply_entity(tiered_rcon.send_command("/sc obs_diff_full_sync()") or "") + truth = tiered_rcon.send_command("""/sc local ch = game.surfaces[1].create_entity{ + name = "character", position = {1200, 100}, force = "player", raise_built = true} +for i = 0, 3 do + game.surfaces[1].create_entity{name = "iron-chest", + position = {1202 + i * 2, 100}, force = "player", raise_built = true} +end +game.surfaces[1].create_entity{name = "iron-chest", + position = {1300, 100}, force = "player", raise_built = true} +rcon.print(ch.position.x .. "," .. ch.position.y)""".strip()) + chx, chy = map(float, truth.split(",")) + try: + assert poll_until(tiered_rcon, c, lambda: c.player_x == chx) + _, _, view, mask = c.observation() + assert view.shape[0] == 5 and int(mask.sum()) == 5 + # nearest-first: the character itself (distance 0), then the 4 close + # chests; the chest 100 tiles away must be excluded + rel_x = [float(view[i][F_X]) for i in range(5)] + assert rel_x[0] == pytest.approx(0.0) # the character + assert all(abs(r) < 12 for r in rel_x), rel_x + assert sorted(rel_x) == sorted(rel_x, key=abs) or all( + abs(rel_x[i]) <= abs(rel_x[i + 1]) + 1e-6 for i in range(4)) + # the far chest is tracked in the table, just not in the view + far_units = [u for u, row in c.entities.items() + if row.startswith("iron-chest,1300")] + assert far_units and far_units[0] not in c.view_units + finally: + tiered_rcon.send_command("""/sc for _, e in ipairs(game.surfaces[1].find_entities_filtered{ + type = "character"}) do e.destroy{raise_destroy = true} end +storage.obs_char = nil rcon.print(1)""".strip())