Per-sheep pen-time metrics, seed support, make webots → menu
* `controllers/shepherd_dog/shepherd_dog.py`
- Tracks the first step at which each sheep crosses the gate; on
auto-finish (all sheep penned) prints a `[results]` summary
block: mode/drive/world/lidar/dogs/seed line, total simulated
time, per-sheep penning order with absolute step + seconds since
sim start, and the gate spread between the first and last
penning.
- Reads `HERDING_SEED` (env / runtime cfg) and seeds the
controller's RNG when set. Empty = time-based default = old
non-deterministic behaviour.
* `controllers/sheep/sheep.py` reads `HERDING_SEED` the same way
(loading `herding_runtime.cfg` itself so it works even when
Webots strips env vars) and seeds Python's RNG XOR'd with the
sheep's name hash, so a fixed seed gives a reproducible flock
trajectory without all sheep starting from identical wander state.
* `tools/run_webots.sh` writes `HERDING_SEED` into the runtime cfg
(empty when unset so existing scripts keep their stochastic
behaviour).
* `tools/webots_menu.sh` gains a Seed prompt (random / fixed
integer); the launch summary box shows the choice next to the
perception row.
* `Makefile`
- `make webots` now fires the interactive picker (replacing the
old positional invocation).
- `make webots_quick MODE=… DRIVE=… WORLD=… N=…` is the old
positional path, kept for batch / scripted use.
Smoke-tested: menu renders Mode → Drive → World → LiDAR → Dogs
→ Sheep → Perception → Seed → Headless prompts and shows the
selected Seed value in the launch summary. 126 pytest cases still
pass.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -37,6 +37,37 @@ timestep = int(robot.getBasicTimeStep())
|
||||
name = robot.getName()
|
||||
self_node = robot.getSelf()
|
||||
|
||||
# Seed Python's RNG (shared with the dog controller) so a fixed
|
||||
# HERDING_SEED produces reproducible runs. Each sheep mixes its name
|
||||
# into the seed so the flock isn't all identical.
|
||||
def _read_runtime_cfg():
|
||||
cfg_path = os.path.join(_PROJECT_ROOT, "herding_runtime.cfg")
|
||||
out = {}
|
||||
if os.path.exists(cfg_path):
|
||||
try:
|
||||
with open(cfg_path) as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
k, _, v = line.partition("=")
|
||||
out[k.strip().upper()] = v.strip()
|
||||
except OSError:
|
||||
pass
|
||||
return out
|
||||
|
||||
_rt = _read_runtime_cfg()
|
||||
_seed_raw = (os.environ.get("HERDING_SEED")
|
||||
or _rt.get("HERDING_SEED")
|
||||
or "").strip()
|
||||
if _seed_raw:
|
||||
try:
|
||||
# XOR with hash(name) so different sheep have different seeds
|
||||
# but the flock as a whole is deterministic for a given seed.
|
||||
random.seed(int(_seed_raw) ^ (hash(name) & 0x7FFFFFFF))
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
left_motor = robot.getDevice("left wheel motor")
|
||||
right_motor = robot.getDevice("right wheel motor")
|
||||
left_motor.setPosition(float("inf"))
|
||||
|
||||
@@ -351,10 +351,32 @@ if MODE in ("strombom", "sequential"):
|
||||
elif MODE == "universal":
|
||||
analytic_teacher = ActiveScanTeacher(universal_action)
|
||||
|
||||
# Optional deterministic seed for the controller's RNG. The sheep
|
||||
# controller seeds itself the same way, so identical HERDING_SEED
|
||||
# values give reproducible trials. If unset (empty), Python uses its
|
||||
# time-based default and runs are non-deterministic.
|
||||
import random as _random
|
||||
_seed_raw = (os.environ.get("HERDING_SEED")
|
||||
or _runtime_cfg.get("HERDING_SEED")
|
||||
or "").strip()
|
||||
if _seed_raw:
|
||||
try:
|
||||
HERDING_SEED = int(_seed_raw)
|
||||
except ValueError:
|
||||
HERDING_SEED = None
|
||||
print(f"[dog] could not parse HERDING_SEED={_seed_raw!r}; using random")
|
||||
else:
|
||||
_random.seed(HERDING_SEED)
|
||||
else:
|
||||
HERDING_SEED = None
|
||||
|
||||
# GT positions from sheep emitters — used **only** for the auto-finish
|
||||
# sentinel and the GT_penned diagnostic line. Never fed into control.
|
||||
# sentinel, the GT_penned diagnostic line, and the per-sheep pen-time
|
||||
# metrics printed at end of run. Never fed into control.
|
||||
_gt_sheep: dict = {}
|
||||
_pen_step: dict = {} # sheep name -> step at which it first became penned
|
||||
_run_done = False
|
||||
_t_start = None # step at which we first see GT positions (sim start)
|
||||
|
||||
prev_action = (0.0, 0.0, 0.0) if DRIVE_MODE == "mecanum" else (0.0, 0.0)
|
||||
step_count = 0
|
||||
@@ -536,10 +558,20 @@ while robot.step(timestep) != -1:
|
||||
left_ear.setPosition(ear_pos)
|
||||
right_ear.setPosition(-ear_pos)
|
||||
|
||||
# Auto-finish: when all GT sheep are penned, write the sentinel.
|
||||
# The launcher polls for it and closes Webots so batch evals don't
|
||||
# hang after the task is done. Bounded by `_gt_sheep` so we don't
|
||||
# fire during the first few steps while the receiver fills.
|
||||
# First step we have GT visibility — record the simulation start
|
||||
# so per-sheep pen times can be reported relative to it.
|
||||
if _gt_sheep and _t_start is None:
|
||||
_t_start = step_count
|
||||
|
||||
# Record the first step at which each sheep is observed penned.
|
||||
for _sname, (_sx, _sy) in _gt_sheep.items():
|
||||
if _sname not in _pen_step and is_penned(_sx, _sy):
|
||||
_pen_step[_sname] = step_count
|
||||
|
||||
# Auto-finish: when all GT sheep are penned, write the sentinel
|
||||
# and print the per-sheep penning summary so the operator sees
|
||||
# the metrics in the terminal. The launcher polls for the
|
||||
# sentinel and closes Webots cleanly.
|
||||
if _gt_sheep and not _run_done:
|
||||
gt_active = sum(1 for x, y in _gt_sheep.values()
|
||||
if not is_penned(x, y))
|
||||
@@ -549,6 +581,26 @@ while robot.step(timestep) != -1:
|
||||
_run_done = True
|
||||
print(f"[dog] all {len(_gt_sheep)} sheep penned at step "
|
||||
f"{step_count} — wrote sentinel, launcher will close Webots")
|
||||
# Only the first dog to detect the finish prints the
|
||||
# summary (in dual-dog mode both run in lock-step but the
|
||||
# sentinel acts as a one-shot lock).
|
||||
_dt = 0.016 # Webots basicTimeStep, seconds
|
||||
print("")
|
||||
print(f"[results] mode={MODE} drive={DRIVE_MODE} world={WORLD} "
|
||||
f"lidar={LIDAR_FOV_VARIANT} dogs={DOG_NAME}"
|
||||
+ (f" seed={HERDING_SEED}" if HERDING_SEED is not None else ""))
|
||||
print(f"[results] total steps: {step_count} "
|
||||
f"({step_count * _dt:.1f} s simulated)")
|
||||
ordered = sorted(_pen_step.items(), key=lambda kv: kv[1])
|
||||
for i, (sn, st) in enumerate(ordered, 1):
|
||||
rel = st - (_t_start or 0)
|
||||
print(f"[results] #{i} {sn:8s} penned at step {st:>6d} "
|
||||
f"({rel * _dt:6.2f} s)")
|
||||
if len(ordered) >= 2:
|
||||
first = ordered[0][1]
|
||||
last = ordered[-1][1]
|
||||
print(f"[results] gate spread: {last - first} steps "
|
||||
f"({(last - first) * _dt:.2f} s) between first and last pen")
|
||||
|
||||
if step_count % 200 == 0:
|
||||
gt_penned = sum(1 for x, y in _gt_sheep.values()
|
||||
|
||||
Reference in New Issue
Block a user