File size: 3,762 Bytes
00c7586 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 | import bisect, hashlib
from fractions import Fraction
import u2_contract as U2
ROLES = ('main', 'left_wrist', 'right_wrist')
HEADING = 'Each frame is a synchronized camera mosaic: main camera above, left wrist camera below left, and right wrist camera below right. All panels show the same physical time. Describe each physical operation once across the views.\n\n'
INSTRUCTION = 'Write one concise, self-contained English instruction for the local operation being attempted in TARGET.\nUse a natural imperative, concrete objects and the working hand; include meaningful support by the other hand and necessary source or destination. Say left hand, right hand or both hands. For a handover, name the object and both the giving and receiving hands.\nInfer the local goal from the video sequence. Use BEFORE and AFTER only to disambiguate TARGET, not to add operations occurring outside it. A failed or repeated attempt retains its established instruction; do not replace the goal with an accident, a success claim or a description of mechanical adjustments. Do not invent an object, hand role or hidden goal.\nOutput only the instruction, on one line, without timestamps, execution-quality commentary or the block names.'
def plans_for_episode(ep, clocks):
clock = clocks['main']
pts = [float(t * Fraction(clock['time_base'])) for t in clock['ticks']]
fps = Fraction(clock['fps'])
plans = []
for unit in ep['units']:
start, end = unit['native_scope_s']
assert 0 <= start < end <= clock['duration_s'] + 1e-05
blocks = []
spans = [('BEFORE', max(0.0, start - 2.0), start), ('TARGET', start, end), ('AFTER', end, min(clock['duration_s'], end + 2.0))]
for role, a, b in spans:
lo, hi = (bisect.bisect_left(pts, a - 1e-09), bisect.bisect_left(pts, b - 1e-09))
hi = min(hi, len(pts))
factor = U2.TARGET_FRAME_FACTOR if role == 'TARGET' else U2.CONTEXT_FRAME_FACTOR
if hi - lo < factor:
if role != 'TARGET':
continue
assert hi - lo == 1, (unit['target_id'], 'empty target')
indices = [lo, lo]
receipt = U2.repeated_single_target_receipt(lo, source_fps_num=fps.numerator, source_fps_den=fps.denominator)
padding = [False, True]
else:
target = role == 'TARGET'
indices, receipt = U2.duration_adaptive_sample(lo, hi, source_fps_num=fps.numerator, source_fps_den=fps.denominator, sample_fps=U2.TARGET_SAMPLE_FPS if target else U2.CONTEXT_SAMPLE_FPS, min_frames=U2.TARGET_MIN_FRAMES if target else U2.CONTEXT_MIN_FRAMES, max_frames=U2.TARGET_MAX_FRAMES if target else U2.CONTEXT_MAX_FRAMES, frame_factor=factor)
padding = [False] * len(indices)
blocks.append(dict(role=role, scope_s=[a, b], source_start_frame=lo, source_end_frame_exclusive=hi, indices=indices, pts_ticks=[clock['ticks'][i] for i in indices], padding_mask=padding, sampling_receipt=receipt))
assert any((b['role'] == 'TARGET' for b in blocks))
assert sum((len(b['indices']) for b in blocks)) <= U2.MAX_TOTAL_FRAMES
prompt = HEADING + '\n\n'.join((b['role'] + ':\n<video>' for b in blocks)) + '\n\n' + INSTRUCTION
plans.append(dict(target_id=unit['target_id'], unit_id=unit['unit_id'], episode_id=ep['episode_id'], source_id=ep['source_id'], group=ep['group'], native_scope_s=unit['native_scope_s'], camera_roles=list(ROLES), clock_fps=str(fps), time_base=clock['time_base'], blocks=blocks, prompt=prompt, prompt_sha256=hashlib.sha256(prompt.encode()).hexdigest(), context_policy='common_real_up_to_2s_each_side', source_clock_origin=clock['first_pts'], do_sample_frames=False))
return plans
|