processing_videojev.py

"""Canonical VideoJev prompts, intact query groups, and video preprocessing."""
import math
import re

import numpy as np
import torch
from transformers.video_utils import VideoMetadata

from .videojev_temporal import pair_timeline, decode
from .videojev_video import read_video

DECIDE = '<|decide|>'
OPTION_END = '<|option_end|>'


def suffix(row, stamps):
    task = row.get('task', 'grounding')
    if task not in ('grounding', 'choice'):
        raise ValueError('task must be grounding or choice')
    for text in [row['query'], *row.get('options', [])]:
        if not isinstance(text, str) or not text.strip() or '<|' in text:
            raise ValueError('Questions/options must be nonempty text without control tokens')
    if task == 'choice' and len(row.get('options', [])) < 2:
        raise ValueError('Choice questions need at least two options')
    instruction = 'temporal grounding' if task == 'grounding' else 'single-choice video question answering'
    text = f'\nTask: {instruction}\nQuestion: {row["query"]}\nOptions:\n'
    candidates = stamps if task == 'grounding' else row['options']
    return text + ''.join(value + OPTION_END + '\n' for value in candidates) + DECIDE


def native_inputs(processor, record, rows):
    grid = record['grid_thw']
    meta = VideoMetadata(total_num_frames=record['frames'],
        fps=record.get('effective_fps', record['fps']), frames_indices=list(range(record['frames'])))
    replacement = processor.replace_video_token(dict(video_grid_thw=np.asarray([[grid[0], 2, 2]]),
        video_metadata=[meta]), video_idx=0)
    stamps = re.findall(r'<[0-9]+\.[0-9] seconds>', replacement)
    raw_times = processor._calculate_timestamps(list(range(record['frames'])), meta.fps, 2)
    if len(stamps) != grid[0] or not np.allclose(raw_times, record['score_times'], rtol=1e-10, atol=1e-8):
        raise ValueError('Timestamp candidates do not match native video midpoints')
    texts = []
    for i, row in enumerate(rows):
        content = ([dict(type='video', video='provided_in_memory')] if i == 0 else [])
        content += [dict(type='text', text=suffix(row, stamps))]
        text = processor.apply_chat_template([dict(role='user', content=content)],
            tokenize=False, add_generation_prompt=False)
        if i == 0:
            replaced, _ = processor.get_text_with_replacements([text], [], [replacement], [])
            value = processor(text=replaced, return_tensors='pt')
            spatial = grid[1] * grid[2] // 4
            slots = value['input_ids'][0] == processor.video_token_id
            if int(slots.sum()) != grid[0] or record['tokens'] != grid[0]*spatial:
                raise ValueError('Native grid/token count mismatch')
            repeats = torch.ones(value['input_ids'].shape[1], dtype=torch.long)
            repeats[slots] = spatial
            first_ids = torch.repeat_interleave(value['input_ids'], repeats, dim=1)
            first_types = torch.repeat_interleave(value['mm_token_type_ids'], repeats, dim=1)
        else:
            texts.append(torch.tensor([processor.tokenizer.encode(text, add_special_tokens=False)]))
    return torch.cat([first_ids, *texts], dim=1), torch.cat([first_types, *[torch.zeros_like(t) for t in texts]], dim=1)


def prepare_query_groups(processor, record, questions, context):
    if not questions or context < 64:
        raise ValueError('Provide questions and a context of at least 64 tokens')
    if len({q.get('id', i) for i, q in enumerate(questions)}) != len(questions):
        raise ValueError('Question IDs must be unique')
    # Query lengths are computed with the exact native tokenizer. No labels enter
    # this function, and no answer history is carried from one group to the next.
    start = 0
    decide_id = processor.tokenizer.convert_tokens_to_ids(DECIDE)
    option_id = processor.tokenizer.convert_tokens_to_ids(OPTION_END)
    while start < len(questions):
        end = start + 1
        ids, types = native_inputs(processor, record, questions[start:end])
        if (ids.shape[1]+63)//64*64 > context:
            raise ValueError(f'A complete query exceeds context; no question was skipped: {start}')
        while end < len(questions):
            candidate_ids, candidate_types = native_inputs(processor, record, questions[start:end+1])
            if (candidate_ids.shape[1]+63)//64*64 > context:
                break
            ids, types = candidate_ids, candidate_types
            end += 1
        rows = questions[start:end]
        query_indices = (ids[0] == decide_id).nonzero().flatten()
        option_indices = (ids[0] == option_id).nonzero().flatten()
        counts = [len(row['options']) if row.get('task') == 'choice' else record['grid_thw'][0] for row in rows]
        if len(query_indices) != len(rows) or len(option_indices) != sum(counts):
            raise ValueError('Incomplete decision/candidate markers')
        pairs = torch.zeros((len(rows), max(counts)), dtype=torch.long)
        mask = torch.zeros_like(pairs, dtype=torch.bool)
        offset, previous = 0, -1
        for i, count in enumerate(counts):
            selected = option_indices[offset:offset+count]
            if not ((selected > previous) & (selected < query_indices[i])).all():
                raise ValueError('Candidate readouts cross query boundaries')
            pairs[i, :count], mask[i, :count] = selected, True
            previous = query_indices[i]
            offset += count
        yield dict(input_ids=ids, mm_token_type_ids=types,
            video_grid_thw=torch.tensor([record['grid_thw']]), pair_indices=pairs,
            query_indices=query_indices, score_mask=mask,
            decision_tasks=torch.tensor([int(row.get('task') == 'choice') for row in rows])), rows
        start = end


def prepare_video(processor, path, *, fps=2., max_frames=2048, context=262144):
    import av
    from transformers.models.qwen3_vl.video_processing_qwen3_vl import smart_resize
    with av.open(str(path)) as container:
        stream = container.streams.video[0]
        duration = float(stream.duration * stream.time_base) if stream.duration is not None else container.duration/1e6
    sample = read_video(dict(video_path=str(path), duration=duration), fps, max_frames=max_frames)
    n, height, width, _ = sample['video'].shape
    vp = processor.video_processor
    budget = context * 4 // 5
    def geometry(cap):
        h, w = smart_resize(n, height, width, temporal_factor=vp.temporal_patch_size,
            factor=vp.patch_size*vp.merge_size, min_pixels=vp.size['shortest_edge'], max_pixels=n*cap)
        return h, w, (n+1)//2*(h//32)*(w//32)
    low, high = max(1024, math.ceil(vp.size['shortest_edge']/n)), 640*640
    if geometry(low)[2] > budget:
        raise ValueError('Complete video cannot fit the context at minimum native resolution')
    while low < high:
        mid = (low+high+1)//2
        if geometry(mid)[2] <= budget:
            low = mid
        else:
            high = mid-1
    meta = VideoMetadata(total_num_frames=n, fps=sample['fps'], frames_indices=list(range(n)))
    native = vp(videos=[sample['video']], video_metadata=[meta], do_sample_frames=False,
        size=dict(shortest_edge=vp.size['shortest_edge'], longest_edge=n*low), return_tensors='pt')
    grid = native['video_grid_thw'][0].tolist()
    times, bounds = pair_timeline(sample['times'], sample['sampled_duration'])
    record = dict(frames=n, fps=sample['fps'], grid_thw=grid, tokens=int(np.prod(grid))//4,
        score_times=times.tolist(), score_bounds=bounds.tolist(), sampled_duration=sample['sampled_duration'])
    return native['pixel_values_videos'], record


def format_predictions(output, rows, record, decoder):
    results = []
    for i, row in enumerate(rows):
        count = int(output.score_mask[i].sum())
        logits = output.logits[i, :count].float().cpu().tolist()
        probabilities = output.probabilities[i, :count].float().cpu().tolist()
        value = dict(id=row.get('id', i), task=row.get('task', 'grounding'),
                     probabilities=probabilities)
        if row.get('task') == 'choice':
            answer = int(np.argmax(probabilities))
            value.update(answer_index=answer, answer=row['options'][answer])
        else:
            value['interval'] = decode(logits, record['score_times'], record['sampled_duration'],
                decoder['threshold'], decoder['smooth'], bounds=record['score_bounds'],
                postprocess=decoder['postprocess'])
        results.append(value)
    return results