chickenbstrips/TheTree_mock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P23-03: 生活の観察

Browse Source
This commit is contained in:
Agent Control Runner
2026-10-16 17:15:06 +09:00
parent 93248f4d79
commit 26f60095bf
7 changed files with 422 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/check_p23_life_observation_evidence.py
+108 -48
View File
@@ -1455,6 +1455,40 @@ def actor_track_transit(report, key, cell):
return not report['paused'] and origin != cell and now-started < .42
def landing_actor_region(state, report, transform, cell, region, transit):
"""Derive corridor landings from raw topology, never from actor flags."""
if not any(intersects(region, cabin['rect']) for cabin in report['cabins']):
return dict(bounds=region, transit=transit)
runtime = state['systems']['runtime']
footprint = {(stop[0]+dx, stop[1], stop[2])
for shaft in elevator_report(state)['shafts'] for stop in shaft['stops']
for dx in range(max(int(shaft.get('width', 2)), 1))}
walkable = {tuple(c) for c in runtime['world']['walkable_cells']}
facilities = codec.indexed(runtime['facilities']['instances'], 'instance_id')
owners = {tuple(c): r['entity_id'] for r in runtime['world']['occupancies'] for c in r['cells']}
distance = lambda c: (abs(c[0]-cell[0]), c[0])
origins = sorted((c for c in footprint if c[1:] == tuple(cell[1:])), key=distance)
lane = set()
for origin in origins:
for direction in (-1, 1):
for step in range(1, 9):
target = (origin[0]+direction*step, origin[1], origin[2])
facility = facilities.get(owners.get(target))
if (target in footprint or target not in walkable or
(facility and facility['definition_id'] not in ('guest_corridor', 'service_corridor'))):
break
lane.add(target)
lane = sorted(lane, key=distance)
if not lane:
rect = transform.rect(cell)
badges = [b for b in report['landing_badges'] if b['cell'][1:] == list(cell[1:])]
anchor = min(badges, key=lambda b: distance(b['cell']))['at'] if badges else [rect[0]-80, rect[1]-16]
return dict(bounds=region, transit=False, no_lane=True, badge_at=anchor)
reserved = list(report['queue_positions'].values())
selected = next((c for c in lane if transform.center(c) not in reserved), lane[0])
return dict(bounds=transform.rect(selected), transit=False, center=transform.center(selected))
def compare_actor_geometry(state, report):
runtime = state['systems']['runtime']
compare_queues(state, report)
@@ -1467,6 +1501,9 @@ def compare_actor_geometry(state, report):
facilities = codec.indexed(runtime['facilities']['instances'], 'instance_id')
staff = codec.indexed(runtime['staff']['staff'], 'staff_id')
occupancy = {r['entity_id']: r['cells'] for r in runtime['world']['occupancies']}
# Cabin rectangles and badges are independently recomputed before use below.
if report['cabins']:
compare_cabin_geometry(state, report)
counts, expected = {}, []
def track_position(key, cell):
track = report['actor_tracks'][key]
@@ -1489,11 +1526,62 @@ def compare_actor_geometry(state, report):
anchors = room_slots(transform.room(cells), p, definition)
index = users.index(identifier)
return [anchors[index][0], anchors[index][1]+dy] if index < len(anchors) else None
people = {'guest/'+r['guest_id']: r for r in runtime['guests']['guests']}
people.update({'staff/'+sid: r for sid, r in staff.items()})
regions = {}
def actor_region(key):
if key in regions:
return regions[key]
person = people[key]
cell = person['current_cell']
rect = transform.rect(cell)
transit = actor_track_transit(report, key, cell)
if transit:
other = transform.rect(report['actor_tracks'][key]['from'])
x, y = min(rect[0], other[0]), min(rect[1], other[1])
rect = [x, y, max(rect[0]+rect[2], other[0]+other[2])-x,
max(rect[1]+rect[3], other[1]+other[3])-y]
result = dict(bounds=rect, transit=transit)
fid = ''
if key in report['queue_positions']:
cx, cy = report['queue_positions'][key]
result = dict(bounds=[cx-transform.dim[0]/2, cy-height/2, transform.dim[0], height], transit=False)
elif key.startswith('guest/'):
target = person['target_facility_id']
if person['current_service_id'] and person['guest_id'] in report['rooms'].get(target, {}).get('people', []):
fid = target
else:
sid = person['staff_id']
if sid in report['carried_patients']:
result = copy.deepcopy(actor_region('staff/'+report['carried_patients'][sid]))
result['bounds'][1] -= 12*p
if 'center' in result:
result['center'][1] -= 12*p
elif sid in report['resting_staff']:
fid = report['resting_staff'][sid]
elif sid in report['treated_patients']:
fid = report['treated_patients'][sid]
else:
target = person['assigned_facility_id']
cells = occupancy.get(target, [])
workers = report['rooms'].get(target, {}).get('staff', [])
if person['state'] in ('WORKING', 'AVAILABLE') and cell in cells and sid in workers:
rect = transform.room(cells)
columns = max(int(rect[2]/(16*p)), 1)
slot = workers.index(sid)
if rect[3] >= 40*p and slot < columns:
result = dict(bounds=[rect[0]+slot*rect[2]/columns, rect[1], rect[2]/columns, rect[3]], transit=False)
if fid and occupancy.get(fid):
result = dict(bounds=transform.room(occupancy[fid]), transit=False)
if key not in report['queue_positions'] and 'center' not in result and not result.get('no_lane'):
result = landing_actor_region(state, report, transform, cell, result['bounds'], result['transit'])
regions[key] = result
return result
for guest in runtime['guests']['guests']:
key, cell = 'guest/'+guest['guest_id'], guest['current_cell']
if key in report['boarded'] or key in report['queue_hidden'] or cell[2] != report['layer']:
continue
center = track_position(key, cell)
center = actor_region(key).get('center', track_position(key, cell))[:]
if key in report['queue_positions']:
center = report['queue_positions'][key]
elif guest['current_service_id']:
@@ -1511,7 +1599,7 @@ def compare_actor_geometry(state, report):
key, cell = 'staff/'+sid, member['current_cell']
if key in report['boarded'] or key in report['queue_hidden'] or cell[2] != report['layer']:
continue
center = track_position(key, cell)
center = actor_region(key).get('center', track_position(key, cell))[:]
special = True
if key in report['queue_positions']:
special = False
@@ -1521,6 +1609,7 @@ def compare_actor_geometry(state, report):
if carrier:
center = track_position('staff/'+carrier['staff_id'], carrier['current_cell'])
center[1] -= 12*p
center = actor_region(key).get('center', center)[:]
elif sid in report['resting_staff']:
center = slot_position(report['resting_staff'][sid], sid, 'staff_waiting_room', center, -height*.5)
elif sid in report['treated_patients']:
@@ -1555,55 +1644,16 @@ def compare_actor_geometry(state, report):
carried = bool(staff.get(key.removeprefix('staff/'), {}).get('carried_inventory_id')) if is_staff else False
expected.append((key, center, is_staff, False, carried, True))
cabin_bounds[key] = cabin['rect']
people = {'guest/'+r['guest_id']: r for r in runtime['guests']['guests']}
people.update({'staff/'+sid: r for sid, r in staff.items()})
def tracked_bounds(key, cell):
track = report['actor_tracks'][key]
transit = actor_track_transit(report, key, cell)
rect = transform.rect(cell)
if transit:
other = transform.rect(track['from'])
x, y = min(rect[0], other[0]), min(rect[1], other[1])
rect = [x, y, max(rect[0]+rect[2], other[0]+other[2])-x,
max(rect[1]+rect[3], other[1]+other[3])-y]
return rect, transit
members = []
members, stranded = [], []
for key, center, is_staff, receiving, carrying, riding in expected:
person = people[key]
cell = person['current_cell']
track = report['actor_tracks'].get(key, {})
region, transit = (cabin_bounds[key], False) if riding else tracked_bounds(key, cell)
fid = ''
if not riding:
if key in report['queue_positions']:
cx, cy = report['queue_positions'][key]
region, transit = [cx-transform.dim[0]/2, cy-height/2, transform.dim[0], height], False
elif not is_staff:
target = person['target_facility_id']
if receiving and target in facilities and key.split('/', 1)[1] in report['rooms'].get(target, {}).get('people', []):
fid = target
else:
sid = person['staff_id']
if sid in report['carried_patients']:
carrier = staff[report['carried_patients'][sid]]
region, transit = tracked_bounds('staff/'+carrier['staff_id'], carrier['current_cell'])
region[1] -= 12*p
elif sid in report['resting_staff']:
fid = report['resting_staff'][sid]
elif sid in report['treated_patients']:
fid = report['treated_patients'][sid]
else:
target = person['assigned_facility_id']
cells = occupancy.get(target, [])
workers = report['rooms'].get(target, {}).get('staff', [])
if person['state'] in ('WORKING', 'AVAILABLE') and target in facilities and cell in cells and sid in workers:
rect = transform.room(cells)
columns = max(int(rect[2]/(16*p)), 1)
slot = workers.index(sid)
if rect[3] >= 40*p and slot < columns:
region, transit = [rect[0]+slot*rect[2]/columns, rect[1], rect[2]/columns, rect[3]], False
if fid and occupancy.get(fid):
region, transit = transform.room(occupancy[fid]), False
relocation = {} if riding else actor_region(key)
if riding:
region, transit = cabin_bounds[key], False
else:
region, transit = relocation['bounds'], relocation['transit']
scale = .5 if riding else .75
scale = max(math.floor(p*scale), 1) if p >= 2 else p*scale
at = [center[0]-8*scale, center[1]+height*.5-3*p-17*scale]
@@ -1650,13 +1700,23 @@ def compare_actor_geometry(state, report):
frame=frame, variant=variant, flip=track.get('direction', 1) < 0,
staff=is_staff, state=states.index(person['state']), receiving=receiving,
carrying=carrying, riding=riding)
members.append(dict(row=row, bounds=[region[0]+p, region[1]+p, region[2]-2*p, region[3]-2*p], transit=transit))
if relocation.get('no_lane'):
aggregate = next((a for a in stranded if a['at'] == relocation['badge_at']), None)
if aggregate is None:
aggregate = dict(members=[], drawn_keys=[], at=relocation['badge_at'], text='')
stranded.append(aggregate)
aggregate['members'].append(dict(key=key, id=key.split('/', 1)[1], state=row['state'], cell=cell, mode=mode))
aggregate['text'] = '+'+str(len(aggregate['members']))
else:
members.append(dict(row=row, bounds=[region[0]+p, region[1]+p, region[2]-2*p, region[3]-2*p], transit=transit))
fitted, aggregations = fitted_actor_layout(members, p, height)
aggregations = stranded + aggregations
close(report.get('actor_aggregations'), aggregations, 'actor aggregation/no-fit')
require([r['key'] for r in report['actor_drawings']] == [k for k in floor_keys if k in fitted], 'actor drawing coverage/order')
actual = list(report['actor_drawings']) + [r for c in report['cabins'] for r in c['passengers']]
require(len(actual) == len(fitted) and {r['key'] for r in actual} == set(fitted), 'missing or duplicate actor')
for row in actual:
require('region' not in row and 'boarded' not in row, 'unsupported actor projection flags')
for field, value in fitted[row['key']].items():
close(row[field], value, 'actor fitted '+field)
exemptions = {m['row']['key'] for m in members if m['transit']}
tools/run_p23_life_observation.py
+2
View File
@@ -311,6 +311,8 @@ def collect(args):
os.environ.update(P23_LAYOUT=layout, P23_OUTPUT=str(directory))
phase('layout-'+layout.lower(), 'tests/ui/test_p23_life_observation.gd')
cases.append({'directory':directory.name, 'phase':phases[-1], 'files':r.inventory(directory)})
for layout in ('A', 'B'):
checker.inspect_stream(out/('layout-'+layout), root)
if args.input_mode == 'native_os':
os_phases = collect_os(r, root, out, stage, engine, commit, refs, protected, outer)
checker = module('p23_os_input_map', root/'tools/check_p23_life_observation_evidence.py')
tools/test_p23_life_observation_evidence.py
+121
View File
@@ -1848,6 +1848,127 @@ class AnchoredGeometryTests(unittest.TestCase):
check.compare_actor_geometry(self.snapshot, report)
class CabinLandingTests(unittest.TestCase):
"""Constructed raw topology; explicit expected draws, never OS evidence."""
setUp = AnchoredGeometryTests.setUp
person = AnchoredGeometryTests.person
drawing = AnchoredGeometryTests.drawing
def lift(self, floor=0):
self.runtime['elevators']['shafts'] = [dict(
config=dict(shaft_id='lift', stops=[[1,0,0], [1,1,0]], width=2, capacity=4),
current_floor=0, phase='PICKUP', boarding_floor=1, phase_origin_floor=0,
remaining_minutes=10*(1-floor), phase_total_minutes=10, operational=True, boarded=[])]
self.report['elevator_tracks'] = {'lift': dict(floor=floor, **{'from': floor}, at=0)}
# The transform is 64x96; the full two-cell cabin is 96x76.
y = check.math.copysign(check.math.floor(abs(174-96*floor)+.5), 174-96*floor)
self.report['cabins'] = [dict(shaft_id='lift', rect=[156,y,96,76], pixel=4,
operational=True, doors_open=False, badge_at=[156,y-14], passengers=[])]
synthetic_repair28_fields(self.report, self.snapshot)
def test_moving_guest_same_floor_and_old_position_negative(self):
self.lift()
self.person('guest/g', [1,0,0])
self.drawing('guest/g', [108,214])
check.compare_actor_geometry(self.snapshot, self.report)
bad = copy.deepcopy(self.report)
bad['actor_drawings'][0].update(center=[172,214], at=[148,199])
with self.assertRaisesRegex(ValueError, 'non-passenger inside cabin'):
check.compare_actor_geometry(self.snapshot, bad)
def test_transit_from_lift_to_corridor_is_relocated_before_jitter(self):
self.lift()
self.person('guest/g', [0,0,0])
self.report.update(paused=False, animation_time=.21)
self.report['actor_tracks']['guest/g']['from'] = [1,0,0]
row = self.drawing('guest/g', [108,214], 'walk')
row['frame'] += 2
check.compare_actor_geometry(self.snapshot, self.report)
bad = copy.deepcopy(self.report)
bad['actor_drawings'][0]['center'][0] += 1
with self.assertRaisesRegex(ValueError, 'actor fitted center'):
check.compare_actor_geometry(self.snapshot, bad)
def test_carrier_and_raised_patient_follow_relocated_region(self):
self.lift()
self.person('staff/carrier', [1,0,0], 'AVAILABLE')
self.person('staff/patient', [1,0,0], 'INCAPACITATED')
self.runtime['staff']['tasks'] = [dict(task_type='rescue_transport', state='MOVING',
assigned_staff_id='carrier', target_entity_id='patient', rescue_facility_id='')]
self.report['carried_patients'] = {'patient': 'carrier'}
self.drawing('staff/carrier', [108,214])
self.drawing('staff/patient', [108,166], 'collapsed')
check.compare_actor_geometry(self.snapshot, self.report)
self.assertTrue(all(not check.intersects(check.actor_identity_core(r), self.report['cabins'][0]['rect'])
for r in self.report['actor_drawings']))
def test_fractional_moving_cabin_overlaps_floor_above(self):
self.lift(.8)
self.person('guest/g', [1,1,0])
self.drawing('guest/g', [108,118])
check.compare_actor_geometry(self.snapshot, self.report)
def test_no_lane_aggregates_identity_and_never_drops_silently(self):
self.lift()
self.runtime['world']['walkable_cells'] = [[1,0,0], [2,0,0]]
self.person('guest/g', [1,0,0])
self.report['actor_aggregations'] = [dict(members=[dict(key='guest/g', id='g', state=3,
cell=[1,0,0], mode='idle')], drawn_keys=[], at=[60,150], text='+1')]
check.compare_actor_geometry(self.snapshot, self.report)
for mutation in ('badge', 'identity', 'hidden'):
bad = copy.deepcopy(self.report)
if mutation == 'badge': bad['actor_aggregations'][0]['at'][0] += 1
if mutation == 'identity': bad['actor_aggregations'][0]['members'] = []
if mutation == 'hidden':
bad['actor_aggregations'] = []
bad['queue_hidden'] = {'guest/g': True}
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
check.compare_actor_geometry(self.snapshot, bad)
self.person('guest/h', [1,0,0])
aggregate = self.report['actor_aggregations'][0]
aggregate['members'].append(dict(key='guest/h', id='h', state=3, cell=[1,0,0], mode='idle'))
aggregate['text'] = '+2'
check.compare_actor_geometry(self.snapshot, self.report)
def test_reserved_lane_prefers_free_then_shares_with_queue(self):
self.lift()
self.person('guest/q', [1,0,0], 'QUEUING')
self.person('guest/g', [1,0,0])
self.runtime['elevators']['trips'] = [dict(request_id='queue', actor_id='guest:q', state='QUEUED',
shaft_id='lift', connector_id='lift.link.0.1', enqueued_at=0,
**{'from': [1,0,0], 'to': [1,1,0]})]
self.report.update(queue_positions={'guest/q':[108,214]},
landing_badges=[dict(at=[60,150], text='待ち1人', cell=[1,0,0],
members=['guest/q'], visible_count=1, overflow=0)])
self.drawing('guest/q', [108,214], 'wait')
self.drawing('guest/g', [300,214])
check.compare_actor_geometry(self.snapshot, self.report)
self.runtime['world']['walkable_cells'].remove([3,0,0])
self.report['actor_drawings'][0].update(at=[100,199])
self.report['actor_drawings'][1].update(center=[108,214], at=[82,199])
check.compare_actor_geometry(self.snapshot, self.report)
def test_flags_drop_and_unnecessary_relocation_fail(self):
self.lift()
self.person('guest/g', [1,0,0])
self.drawing('guest/g', [108,214])
check.compare_actor_geometry(self.snapshot, self.report)
for mutation in ('riding', 'region', 'boarded', 'drop'):
bad = copy.deepcopy(self.report)
if mutation == 'riding': bad['actor_drawings'][0]['riding'] = True
if mutation == 'region':
bad['actor_drawings'][0]['region'] = 'landing'
if mutation == 'boarded': bad['boarded']['guest/g'] = True
if mutation == 'drop': bad['actor_drawings'] = []
with self.subTest(mutation=mutation), self.assertRaises(ValueError):
check.compare_actor_geometry(self.snapshot, bad)
# A free corridor actor must remain at its actual cell, even beside a lift.
self.runtime['guests']['guests'][0]['current_cell'] = [3,0,0]
self.report['actor_tracks']['guest/g'].update(cell=[3,0,0], **{'from': [3,0,0]})
with self.assertRaisesRegex(ValueError, 'actor fitted'):
check.compare_actor_geometry(self.snapshot, self.report)
class ComparisonTests(unittest.TestCase):
def pair(self):
declaration = dict(metric='sum_destination_arrival_minutes_entire_cohort',