dasha_f
0d32f32db0
Замкнутый контур "поток -> CV -> механика": товары идут по конвейеру с шагом 700 мм, класс определяется стереопайплайном во время движения, пушер и плуг реагируют физически. Состав: * control_test/ - ячейка и CV. run_sorting_cv.py + cv_worker.py (два процесса, потому что torch внутри Isaac роняет сцену), cell.py (физика лент, плуга, пушера), measure_plane.py (замер габаритов), README.md и .memory.md с замерами, проблемами и ловушками * robozon_sorter/ - модули симуляции, scripts/ - утилиты, scene/ - сцены * assets/ - меши товаров, плуг, объекты Objaverse Бейзлайн CV: DEFOM-Stereo vitl, вход 480, iters 24, кроп зоны осмотра, без сегментации. На потоке 700 мм - классы 8/9, габариты MAE 32.8 мм, 469 мс на товар при такте 700 мс. Веса моделей (4.5 ГБ) и пропсы конвейера NVIDIA (274 МБ) не включены - источники и команды скачивания в MODELS.md. Выход прогонов (captures/, runtime/) не включён: воспроизводится. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
439 lines
21 KiB
Python
439 lines
21 KiB
Python
"""Controlled sorting test over the whole item library, with per-item kinematics.
|
||
|
||
isaacsim_send.py --context test --file scripts/test_sorting_run.py \
|
||
--args-json '{"vision": true, "preset": "bright", "repeats": 2}'
|
||
|
||
What it records, per dispatched item:
|
||
|
||
* **dispatch** when it was released and with what ground-truth class
|
||
* **detection** predicted class, dimensions, roundness, views, CRE time
|
||
* **kinematics** at the moment the item is level with the plow: the commanded angle, the
|
||
angle the arm had actually reached, and the arm's **angular rate** in deg/s. Commanded
|
||
and reached are different numbers - the drive is compliant - and a blade that is still
|
||
travelling when the item arrives deflects it differently from one that has settled.
|
||
* **outcome** where it came to rest: tray B, tray C, the D bin, a lane, the line, or the
|
||
floor, plus the resting pose and whether it matches the tray its class maps to.
|
||
|
||
Two metric blocks are reported separately, because they fail independently: classification
|
||
(what the vision stack decided) and delivery (where the mechanics actually put it). An item
|
||
can be classified perfectly and still be left on the line, and the run is only useful if
|
||
those two are not conflated.
|
||
|
||
Lighting preset and the floor come from `sim/staging`, so a run can be repeated under
|
||
`bright` / `dim` / `harsh` to see how much of the classification error is illumination.
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from collections import Counter, defaultdict
|
||
|
||
REPO = "/home/dasha/robozon-sorter"
|
||
if REPO not in sys.path:
|
||
sys.path.insert(0, REPO)
|
||
import importlib
|
||
|
||
# The live Isaac process keeps every module it has ever imported, so an edited
|
||
# robozon_sorter/ on disk is invisible to a second run. Dropping the package is not enough
|
||
# on its own: a module file that did not exist when the directory was first scanned stays
|
||
# invisible until the import finder's cached listing is thrown away too.
|
||
for _m in [k for k in list(sys.modules) if k.startswith("robozon_sorter")]:
|
||
del sys.modules[_m]
|
||
importlib.invalidate_caches()
|
||
|
||
import omni.timeline
|
||
import isaacsim.core.experimental.utils.app as app_utils
|
||
|
||
from robozon_sorter import config as C
|
||
from robozon_sorter.sim import lane_beams, plow_sort, plow_vision, staging
|
||
from robozon_sorter.sim.mechanics import Cell
|
||
from robozon_sorter.sim.spawner import AutoFeeder
|
||
|
||
USE_VISION = bool(globals().get("vision", True))
|
||
PRESET = globals().get("preset", "bright")
|
||
REPEATS = int(globals().get("repeats", 1))
|
||
PITCH = float(globals().get("pitch", 2.5))
|
||
SPEED = float(globals().get("speed", 1.0))
|
||
LIMIT = int(globals().get("limit", 0)) # 0 = whole library
|
||
CLASSES_ONLY = set(str(globals().get("classes", "")).upper()) or None
|
||
# Explicit dispatch list, in order, repeats allowed. `limit`/`classes` cannot express
|
||
# "these exact items, plus an even 10/10/10 of the rest" when a class has fewer than 10
|
||
# unique members - the only way to balance is to send some of them twice.
|
||
ONLY = [n for n in str(globals().get("only", "")).split(",") if n.strip()]
|
||
# Items to score SEPARATELY as well as in the overall figures.
|
||
FOCUS = [n for n in str(globals().get("focus", "")).split(",") if n.strip()]
|
||
BUDGET = float(globals().get("max_seconds", 240.0))
|
||
TRACK_END_X = float(globals().get("track_end_x", -11.5)) # past both trays
|
||
ITEMS_DIR = globals().get("items_dir", f"{REPO}/assets/items")
|
||
OUT = globals().get("out", f"{REPO}/runs/test_sorting_{PRESET}.json")
|
||
|
||
EXPECT = {"D": "bin", "B": "container_B", "C": "container_C"}
|
||
CLASSES = ("B", "C", "D")
|
||
|
||
# ---------------------------------------------------------------- scene
|
||
C.BELT_SPEED = SPEED
|
||
stage, info = plow_vision.load(belt_speed=SPEED, script_control=True,
|
||
meshes_dir=ITEMS_DIR)
|
||
staged = staging.stage_cell(stage, preset=PRESET, floor=True)
|
||
plow_sort.keep_lanes_active(stage)
|
||
lanes = plow_sort.configure_lanes(stage, SPEED)
|
||
opened = plow_sort.open_junction(stage)
|
||
|
||
items = {k: v["zone"] for k, v in info["items"].items()}
|
||
gt_dims = {k: v.get("gt_dims_mm") for k, v in info["items"].items()}
|
||
print(f"library {len(items)} items {dict(Counter(items.values()))} | light={PRESET} "
|
||
f"| floor={'yes' if staged.get('floor') else 'no'} | lanes={len(lanes)} "
|
||
f"| junction opened={len(opened)}")
|
||
|
||
vision = None
|
||
if USE_VISION:
|
||
# The streaming launcher starts Kit WITHOUT the user site-packages, so ultralytics and
|
||
# torch installed under ~/.local are invisible to the running app even though
|
||
# `python.sh` imports them fine - it is the same interpreter (3.12.13), just a
|
||
# different sys.path. Appending (not prepending) leaves Kit's own bundled copies first.
|
||
for _sp in ("/home/dasha/.local/lib/python3.12/site-packages",):
|
||
if os.path.isdir(_sp) and _sp not in sys.path:
|
||
sys.path.append(_sp)
|
||
from robozon_sorter.cv.pipeline import CreRoiV2b
|
||
vision = CreRoiV2b()
|
||
vision.attach_cameras()
|
||
_w = await vision.warmup()
|
||
print(f"CRE-ROI v2b attached | прогрев камер: {'ок' if _w['ok'] else 'НЕ УДАЛСЯ'} "
|
||
f"за {_w['attempts']} подход(а), самый тёмный глаз max={_w['darkest_eye_max']}")
|
||
if not _w["ok"]:
|
||
print(" ВНИМАНИЕ: камеры всё ещё отдают чёрное - классификация будет пустой")
|
||
|
||
await app_utils.update_app_async(steps=40)
|
||
# Aim the viewport at the plow before anything else. The default Persp framing tries to
|
||
# fit the WHOLE stage, and the stage contains the parked queue off at x +37 - so the cell
|
||
# ends up a few pixels wide and the stream looks black with only the emissive laser stripe
|
||
# in it. That is what "renders wrong" was: aim, not lighting.
|
||
try:
|
||
from isaacsim.core.rendering_manager import ViewportManager
|
||
ViewportManager.set_camera_view("/OmniverseKit_Persp", eye=[-4.5, -5.0, 5.0],
|
||
target=[-6.5, 0.0, 1.8])
|
||
except Exception as _e:
|
||
print(" (камеру навести не удалось:", _e, ")")
|
||
|
||
cell = Cell(stage, items.keys())
|
||
cell.park_all()
|
||
await app_utils.update_app_async(steps=15)
|
||
|
||
base_order = sorted(items)
|
||
# Optional class filter, e.g. classes="BC" runs only the B and C items. Without it a small
|
||
# `limit` just takes the first N alphabetically, which can miss a whole class: limit=8 gave
|
||
# B=4 D=4 and not one C, so the C route went untested.
|
||
if ONLY:
|
||
missing = [n for n in ONLY if n not in items]
|
||
if missing:
|
||
print(f" ВНИМАНИЕ: нет в библиотеке: {missing}")
|
||
base_order = [n.strip() for n in ONLY if n.strip() in items]
|
||
elif CLASSES_ONLY:
|
||
base_order = [n for n in base_order if items[n] in CLASSES_ONLY]
|
||
if LIMIT:
|
||
base_order = base_order[:LIMIT]
|
||
order = base_order * max(1, REPEATS) # repeat the library to reach a dispatch count
|
||
route, classes = ({}, {}) if USE_VISION else (dict(items), dict(items))
|
||
sorter = plow_sort.PlowSorter(stage, cell, classes, plow_sort.calibrate_mapping())
|
||
BEAMS = lane_beams.LaneBeams(stage, cell, plow=sorter.plow)
|
||
print(f"dispatching {len(order)} ({len(base_order)} unique x{max(1, REPEATS)}) | "
|
||
f"mapping {sorter.mapping} | pitch {PITCH} m @ {SPEED} m/s")
|
||
|
||
# ---------------------------------------------------------------- logging
|
||
def blank(name, pas):
|
||
return dict(item=name, pass_no=pas, gt=items[name], gt_dims=gt_dims.get(name),
|
||
released_t=None, pred=None, dims=None, k=None, views=None, cre_ms=None,
|
||
sensed=False, commanded=None,
|
||
arm_at_plow=None, rate_at_plow=None, arm_max_rate=0.0,
|
||
max_speed=0.0, blowup=None, trace=[],
|
||
contact=[], contact_first=None, contact_last=None,
|
||
outcome=None, expected=EXPECT.get(items[name]), delivered=None,
|
||
final=None)
|
||
|
||
pas = defaultdict(int)
|
||
rec = {} # name -> record for the pass currently on the line
|
||
done_records = []
|
||
events = []
|
||
t_sim = 0.0
|
||
|
||
DECIMATE = 8
|
||
BLOWUP_MS = 5.0
|
||
_tick = 0
|
||
_prev_angle = 0.0
|
||
|
||
|
||
def on_event(kind, name, payload):
|
||
events.append(dict(t=round(t_sim, 2), kind=kind, item=name, payload=str(payload)))
|
||
if kind in ("release", "divert", "error"):
|
||
print(f" {kind:8s} {name:20s} {payload if payload else ''}")
|
||
|
||
|
||
feeder = AutoFeeder(cell, order=order, pitch=PITCH, route=route, on_event=on_event)
|
||
|
||
|
||
def _speed(name):
|
||
try:
|
||
v = cell._rp[name].get_velocities()[0].numpy()[0]
|
||
return float((v[0] ** 2 + v[1] ** 2 + v[2] ** 2) ** 0.5)
|
||
except Exception:
|
||
return 0.0
|
||
|
||
|
||
def _step(dt):
|
||
"""service the plow and sample kinematics as goods cross it"""
|
||
global t_sim, _tick, _prev_angle
|
||
t_sim += dt
|
||
_tick += 1
|
||
try:
|
||
sorter.update(dt)
|
||
BEAMS.tick(dt)
|
||
BEAMS.poll(rate=C.PLOW_SWEEP_RATE)
|
||
arm = sorter.plow.angle
|
||
rate = (arm - _prev_angle) / dt if dt > 0 else 0.0 # deg/s, measured not commanded
|
||
_prev_angle = arm
|
||
for n in list(feeder.active):
|
||
r = rec.get(n)
|
||
if r is None:
|
||
continue
|
||
p = cell.pose(n)
|
||
x, y, z = float(p[0]), float(p[1]), float(p[2])
|
||
if x >= -5.6:
|
||
continue
|
||
spd = _speed(n)
|
||
r["max_speed"] = max(r["max_speed"], round(spd, 2))
|
||
r["arm_max_rate"] = max(r["arm_max_rate"], round(abs(rate), 1))
|
||
if spd > BLOWUP_MS and r["blowup"] is None:
|
||
r["blowup"] = dict(t=round(t_sim, 2), x=round(x, 3), y=round(y, 3),
|
||
z=round(z, 3), speed=round(spd, 1),
|
||
arm=round(arm, 1), rate=round(rate, 1))
|
||
if _tick % DECIMATE == 0 and len(r["trace"]) < 50:
|
||
r["trace"].append(dict(t=round(t_sim, 2), x=round(x, 3), y=round(y, 3),
|
||
z=round(z, 3), v=round(spd, 2),
|
||
cmd=round(sorter.plow.commanded, 1),
|
||
arm=round(arm, 1), rate=round(rate, 1)))
|
||
if n in sorter.decided and not r["sensed"]:
|
||
r["sensed"] = True
|
||
r["commanded"] = round(sorter.decided[n], 1)
|
||
# the instant the item is level with the plow: this is the state that decides
|
||
if abs(x - C.PLOW_POS[0]) < 0.25 and r["arm_at_plow"] is None:
|
||
r["arm_at_plow"] = round(arm, 1)
|
||
r["rate_at_plow"] = round(rate, 1)
|
||
# CONTACT WINDOW: while the item is inside the arm's sweep radius, record how
|
||
# the blade is actually turning. This is what says whether it leaned the item
|
||
# over at tip speed or arrived as a hit - a single sample at the plow centre
|
||
# cannot tell those apart.
|
||
reach = (x - C.PLOW_POS[0]) ** 2 + (y - C.PLOW_POS[1]) ** 2
|
||
if reach < (C.PLOW_ARM_LEN + 0.10) ** 2:
|
||
if r["contact_first"] is None:
|
||
r["contact_first"] = dict(t=round(t_sim, 2), x=round(x, 3),
|
||
y=round(y, 3), arm=round(arm, 1),
|
||
rate=round(rate, 1), v=round(spd, 2))
|
||
if len(r["contact"]) < 40:
|
||
r["contact"].append(dict(t=round(t_sim, 2), y=round(y, 3),
|
||
arm=round(arm, 1), rate=round(rate, 1),
|
||
v=round(spd, 2)))
|
||
r["contact_last"] = dict(t=round(t_sim, 2), y=round(y, 3),
|
||
arm=round(arm, 1), v=round(spd, 2))
|
||
except Exception as exc:
|
||
events.append(dict(t=round(t_sim, 2), kind="step-error", item="", payload=repr(exc)))
|
||
|
||
|
||
from omni.physx import get_physx_interface
|
||
sub = get_physx_interface().subscribe_physics_step_events(_step)
|
||
feeder.install()
|
||
|
||
timeline = omni.timeline.get_timeline_interface()
|
||
app_utils.play(commit=True)
|
||
await app_utils.update_app_async(steps=20)
|
||
|
||
# ---------------------------------------------------------------- run
|
||
seen, settled = set(), {}
|
||
t0 = time.time()
|
||
while time.time() - t0 < BUDGET:
|
||
await app_utils.update_app_async(steps=15)
|
||
|
||
for n in feeder.active: # open a record when an item is released
|
||
if n not in rec:
|
||
pas[n] += 1
|
||
rec[n] = blank(n, pas[n])
|
||
rec[n]["released_t"] = round(t_sim, 2)
|
||
|
||
if vision is not None:
|
||
for name in list(feeder.active):
|
||
if name in seen or name not in rec:
|
||
continue
|
||
if abs(float(cell.pose(name)[0]) - C.CAM_X) < 0.10:
|
||
was = timeline.is_playing()
|
||
res = vision.measure()
|
||
if was and not timeline.is_playing():
|
||
timeline.play()
|
||
await app_utils.update_app_async(steps=2)
|
||
seen.add(name)
|
||
r = rec[name]
|
||
r.update(pred=res.get("cls"), dims=res.get("dims"),
|
||
k=round(res.get("k", 0.0), 3), views=res.get("views"),
|
||
cre_ms=res.get("cre_ms"))
|
||
route[name] = res.get("cls")
|
||
sorter.classes[name] = res.get("cls")
|
||
|
||
for n in list(rec): # freeze an outcome once the item stops
|
||
if n in settled:
|
||
continue
|
||
where = sorter.lane_of(n)
|
||
if where == "line" and cell.where(n) == "bin":
|
||
where = "bin"
|
||
p = cell.pose(n)
|
||
resting = where.startswith("container") or where in ("bin", "floor")
|
||
# Freeze only once the item is genuinely done. The old cutoff was MAIN_X0 + 0.35 =
|
||
# -7.65, which is the fork apex - every item was declared "line-end" at full 0.80 m/s
|
||
# the instant it entered its branch, so no B or C delivery could ever be observed.
|
||
if resting or (where == "line" and float(p[0]) < TRACK_END_X):
|
||
settled[n] = where if resting else "line-end"
|
||
r = rec.pop(n)
|
||
r["outcome"] = settled[n]
|
||
r["final"] = [round(float(v), 3) for v in p[:3]]
|
||
r["delivered"] = (r["outcome"] == r["expected"])
|
||
done_records.append(r)
|
||
seen.discard(n)
|
||
settled.pop(n, None)
|
||
if len(done_records) >= len(order):
|
||
break
|
||
|
||
for n, r in list(rec.items()): # whatever is still on the line at the end
|
||
p = cell.pose(n)
|
||
r["outcome"] = sorter.lane_of(n)
|
||
r["final"] = [round(float(v), 3) for v in p[:3]]
|
||
r["delivered"] = (r["outcome"] == r["expected"])
|
||
done_records.append(r)
|
||
|
||
app_utils.stop()
|
||
await app_utils.update_app_async(steps=10)
|
||
sub = None
|
||
feeder.remove()
|
||
|
||
# ---------------------------------------------------------------- metrics
|
||
print(f"\n===== DISPATCHED {len(done_records)} =====")
|
||
print(f"{'item':22s} {'gt':2s} {'pred':4s} {'outcome':13s} {'want':13s} "
|
||
f"{'arm':>6s} {'rate':>8s} {'vmax':>6s}")
|
||
for r in done_records:
|
||
print(f"{r['item']:22s} {r['gt']:2s} {str(r['pred'] or '-'):4s} "
|
||
f"{str(r['outcome']):13s} {str(r['expected']):13s} "
|
||
f"{str(r['arm_at_plow']):>6s} {str(r['rate_at_plow']):>8s} "
|
||
f"{r['max_speed']:>6.1f} {'OK' if r['delivered'] else ''}")
|
||
|
||
# --- classification -------------------------------------------------------
|
||
graded = [r for r in done_records if r["pred"] in CLASSES]
|
||
print("\n===== CLASSIFICATION (CV) =====")
|
||
if graded:
|
||
conf = {a: Counter() for a in CLASSES}
|
||
for r in graded:
|
||
conf[r["gt"]][r["pred"]] += 1
|
||
hits = sum(conf[a][a] for a in CLASSES)
|
||
print(f" accuracy {hits}/{len(graded)} = {hits / len(graded):.2f}")
|
||
print(" confusion (rows GT, cols pred): " + " ".join(CLASSES))
|
||
for a in CLASSES:
|
||
print(f" {a}: " + " ".join(f"{conf[a][b]:3d}" for b in CLASSES))
|
||
for a in CLASSES:
|
||
tp = conf[a][a]
|
||
fp = sum(conf[g][a] for g in CLASSES) - tp
|
||
fn = sum(conf[a].values()) - tp
|
||
pr = tp / (tp + fp) if tp + fp else 0.0
|
||
rc = tp / (tp + fn) if tp + fn else 0.0
|
||
f1 = 2 * pr * rc / (pr + rc) if pr + rc else 0.0
|
||
print(f" {a}: precision {pr:.2f} recall {rc:.2f} F1 {f1:.2f} (n={tp + fn})")
|
||
cre = [r["cre_ms"] for r in graded if r.get("cre_ms")]
|
||
if cre:
|
||
print(f" CRE {sum(cre) / len(cre):.0f} ms/item over {len(cre)}")
|
||
|
||
if FOCUS:
|
||
fset = {n.strip() for n in FOCUS}
|
||
fg = [r for r in graded if r["item"] in fset]
|
||
print(f"\n ----- ОТДЕЛЬНО ПО НАЗВАННЫМ ТОВАРАМ ({len(fg)} из {len(fset)}) -----")
|
||
print(f" {'товар':<20} {'GT':<3} {'пред':<5} {'дim пред, мм':<20} {'GT дим, мм':<20} {'k':<6} верно")
|
||
okn = 0
|
||
for r in sorted(fg, key=lambda r: r["item"]):
|
||
good = r["pred"] == r["gt"]
|
||
okn += bool(good)
|
||
dp = "x".join(str(int(x)) for x in (r.get("dims") or [])) or "-"
|
||
dg = "x".join(str(int(x)) for x in (r.get("gt_dims") or [])) or "-"
|
||
print(f" {r['item']:<20} {r['gt']:<3} {str(r['pred']):<5} {dp:<20} {dg:<20} "
|
||
f"{(r.get('k') or 0):<6.3f} {'да' if good else 'НЕТ'}")
|
||
if fg:
|
||
print(f" точность по названным: {okn}/{len(fg)} = {okn / len(fg):.2f}")
|
||
miss = sorted(fset - {r["item"] for r in fg})
|
||
if miss:
|
||
print(f" не получили предсказания: {miss}")
|
||
else:
|
||
print(" no vision this run")
|
||
|
||
# --- delivery -------------------------------------------------------------
|
||
if sorter.contact is not None:
|
||
rep = sorter.contact.report()
|
||
print("\n===== PLOW CONTACT SENSOR =====")
|
||
print(f" {len(rep['touches'])} items touched the blade")
|
||
print(f" {'item':22s} {'cls':4s} {'angle@touch':>12s} {'range':>14s} {'dur s':>7s}")
|
||
for t in rep["touches"]:
|
||
print(f" {t['item']:22s} {str(t['cls']):4s} {str(t['angle_at_touch']):>12s} "
|
||
f"{str(t['angle_min']) + '..' + str(t['angle_max']):>14s} "
|
||
f"{str(t['duration']):>7s}"
|
||
+ ("" if t["classified"] else " UNCLASSIFIED - not steered"))
|
||
|
||
rep = BEAMS.report()
|
||
cf = sorted(getattr(sorter, "conflicts", set()))
|
||
print("\n===== КОНФЛИКТЫ ОЧЕРЕДИ ПЛУГА =====")
|
||
if not cf:
|
||
print(" нет: в зоне лезвия ни разу не оказалось двух классов одновременно")
|
||
else:
|
||
print(f" {len(cf)} товар(ов) делили зону лезвия с товаром ДРУГОГО класса.")
|
||
print(" Один нож не может держать два угла сразу - это предел подачи, не сбой:")
|
||
print(" " + ", ".join(cf))
|
||
|
||
print("\n===== ЛАЗЕР ПЕРЕД ПЛУГОМ (предустановка угла) =====")
|
||
gl = getattr(sorter, "gate_log", [])
|
||
if not gl:
|
||
print(" створ не сработал ни разу")
|
||
else:
|
||
print(f" сработал {len(gl)} раз | створ x={plow_sort.SENSE_X}, лезвие с x=-7.32")
|
||
print(f" {'товар':<20} {'класс':<6} {'угол':>7} {'x на срабатывании':>18}")
|
||
for g in gl:
|
||
print(f" {g['item']:<20} {str(g['cls']):<6} {g['angle']:>+7.1f} {g['x']:>18.2f}")
|
||
|
||
print("\n===== ЛАЗЕРНЫЕ ДАТЧИКИ НА ЛЕНТАХ B/C =====")
|
||
print(f" доехали до ленты: {len(rep)} из {len(done_records)} отправленных")
|
||
for c in rep:
|
||
print(f" {c['item']:20s} -> {c['lane']:7s} t={c['t']:6.2f}s угол ножа={c['angle']} v={c['speed']}")
|
||
if not rep:
|
||
print(" ни один товар не доехал ни до одной ленты")
|
||
|
||
print("\n===== DELIVERY (mechanics) =====")
|
||
ok = [r for r in done_records if r["delivered"]]
|
||
print(f" delivered {len(ok)}/{len(done_records)} = {len(ok) / max(len(done_records), 1):.2f}")
|
||
per_class = defaultdict(lambda: [0, 0])
|
||
for r in done_records:
|
||
per_class[r["gt"]][1] += 1
|
||
per_class[r["gt"]][0] += bool(r["delivered"])
|
||
for a in CLASSES:
|
||
got, tot = per_class[a]
|
||
if tot:
|
||
print(f" {a}: {got}/{tot} into {EXPECT[a]}")
|
||
print(" where everything ended up: " +
|
||
str(dict(Counter(r["outcome"] for r in done_records))))
|
||
thrown = [r for r in done_records if r["blowup"]]
|
||
stalled = [r for r in done_records if r["outcome"] in ("line", "lane_B", "lane_C")]
|
||
print(f" thrown by the mechanics: {len(thrown)} | stalled short of a tray: {len(stalled)}")
|
||
if thrown:
|
||
r = thrown[0]
|
||
print(f" e.g. {r['item']}: {r['blowup']}")
|
||
if stalled:
|
||
r = stalled[0]
|
||
print(f" e.g. {r['item']}: stopped at {r['final']} arm={r['arm_at_plow']}")
|
||
|
||
os.makedirs(os.path.dirname(OUT), exist_ok=True)
|
||
contact_report = sorter.contact.report() if sorter.contact else None
|
||
json.dump(dict(contact=contact_report, config=dict(preset=PRESET, vision=USE_VISION, pitch=PITCH, speed=SPEED,
|
||
repeats=REPEATS, mapping=sorter.mapping, expect=EXPECT,
|
||
staged=staged, items_dir=ITEMS_DIR),
|
||
records=done_records, events=events[-400:]),
|
||
open(OUT, "w"), indent=2, ensure_ascii=False)
|
||
print(f"\nlog -> {OUT}")
|