"""Package validated clips; pixel checks are technical only, not scene interpretation.""" import collections import csv import json import math import os from pathlib import Path import re import subprocess from PIL import Image, ImageChops, ImageStat import make_clips as b def pixel_frame(path, ticks): b.resources() cmd = ['ffmpeg','-nostdin','-hide_banner','-loglevel','error','-copyts','-threads','2', '-ss',f'{max(0,ticks-1000000)/1e6:.6f}','-i',str(path),'-map','0:v:0','-an', '-filter_threads','1','-vf',f'trim=start_pts={ticks}:end_pts={ticks+1}', '-frames:v','1','-fps_mode','passthrough','-threads','1','-pix_fmt','rgb24', '-f','rawvideo','pipe:1'] result = subprocess.run(cmd,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=90,check=True) assert not result.stderr.strip(), result.stderr.decode(errors='replace')[-1200:] assert len(result.stdout)==1280*720*3 return Image.frombytes('RGB',(1280,720),result.stdout),cmd def pixel_checks(manifest,ticks): target = b.META/'validation/technical_pixel_samples.json' if target.exists(): return b.load(target) samples = [] for ident in ('C01','C12','L01'): spec = next(s for s in manifest['clips'] if s['id']==ident) for n in sorted({spec['start'],(spec['start']+spec['end'])//2,spec['end']}): source,source_cmd = pixel_frame(b.SOURCE,ticks[n-1]) output,output_cmd = pixel_frame(b.DOWNLOAD/spec['output'],ticks[n-1]-spec['first_pts_ticks']) diff = ImageChops.difference(source,output) mse = sum(v*v for v in ImageStat.Stat(diff).rms)/3 psnr = 10*math.log10(255*255/mse) if mse else None assert psnr is None or psnr>=28, (ident,n,psnr) samples.append(dict(clip_id=ident,review_frame=n,clip_frame=n-spec['start']+1, rgb_psnr_db=psnr,source_command=source_cmd,output_command=output_cmd)) report = dict(status='PASS',samples=samples,count=len(samples), purpose='source/output frame and geometry sanity for lossy encoding; NOT model or semantic accuracy', semantic_review=False,saved_images=False) b.save(target,report) return report def intervals(numbers): result = [] for n in sorted(numbers): if result and result[-1][1]+1==n: result[-1][1]=n else: result.append([n,n]) return result def main(): os.sched_setaffinity(0,b.CPUS) os.nice(10) manifest,rows,ticks = b.prepare() observed = re.findall(r'^\| O(\d{2}) \| (\d+)~(\d+) \|',b.OBS.read_text(),re.M) assert [(int(i),int(a),int(z)) for i,a,z in observed] == [(i,a,z) for i,(a,z) in enumerate(b.OBS_RANGES,1)] results = [] for spec in manifest['clips']: result = b.load(b.META/'validation'/f"{spec['id']}.json") assert result['status']=='PASS' and result['frames']==spec['frames'] assert b.digest(b.DOWNLOAD/spec['output'])==result['sha256'] probe = b.load(b.REPORT/result['decode_evidence']) assert probe['streams'][0].get('sample_aspect_ratio') == manifest['source_stream'].get('sample_aspect_ratio') path = b.META/'frame_maps'/f"{spec['id']}.csv" assert b.digest(path)==result['map_sha256'] with path.open(encoding='utf-8-sig',newline='') as f: table = list(csv.DictReader(f)) assert [int(r['clip_frame_1based']) for r in table]==list(range(1,spec['frames']+1)) assert [int(r['review_frame_1based']) for r in table]==list(range(spec['start'],spec['end']+1)) assert len(table)==len({int(r['review_frame_1based']) for r in table}) assert all(r['region']==('core' if spec['core_start']<=int(r['review_frame_1based'])<=spec['core_end'] else 'context') for r in table) results.append(result) pixel_report = pixel_checks(manifest,ticks) short_counts = collections.Counter(n for s in manifest['clips'] if s['id']!='L01' for n in range(s['start'],s['end']+1)) long_frames = set(range(11963,31938)) all_frames = set(short_counts)|long_frames observations = [] for i,(a,z) in enumerate(b.OBS_RANGES,1): ident = f'O{i:02d}' needed = set(range(a,z+1)) assert needed<=all_frames,ident linked = [dict(clip_id=s['id'],review_start=max(a,s['start']),review_end=min(z,s['end']), clip_frame_start=max(a,s['start'])-s['start']+1,clip_frame_end=min(z,s['end'])-s['start']+1, selected_core=ident in s['core_observation_ids']) for s in manifest['clips'] if a<=s['end'] and z>=s['start']] observations.append(dict(observation_id=ident,review_start=a,review_end=z, individual_coverage=len(needed&set(short_counts)),continuous_only_ranges=intervals(needed-set(short_counts)), linked_clips=linked,fall_expectation='non_event_reference_not_mandatory', fight_expectation='USER_JUDGMENT_PENDING' if i==20 else 'use_action_subintervals_in_observation_snapshot')) coverage = dict(status='PASS',observations=observations, individual_unique_review_frames=len(short_counts),individual_total_frames=sum(short_counts.values()), intentional_individual_overlap_frames=sum(v>1 for v in short_counts.values()), continuous_overlap_with_individual_frames=len(set(short_counts)&long_frames), warning='Overlapping source frames and L01 are not independent evaluation samples; core is not precise action GT.') b.save(b.META/'observation_clip_links.json',coverage) c07 = next(r for r in results if r['id']=='C07') tail = c07['output_duration_seconds']-c07['relative_last_pts_seconds'] assert abs(tail-1.016668)<0.000002 b.save(b.META/'validation/time_boundary_notes.json',dict( C07=dict(last_review_frame=14504,last_sample_display_seconds=tail,duplicated_frames=0, note='Original final sample duration retained; do not interpret as extra temporal observations.'), C08=dict(first_review_frame=14505,relative_first_pts=0), L01=dict(gap_previous_review_frame=14504,gap_next_review_frame=14505,gap_seconds=1.016668, loop_join='last-to-first is not continuous source action'))) with (b.DOWNLOAD/'영상목록.csv').open('x',encoding='utf-8-sig',newline='') as f: writer = csv.writer(f) writer.writerow(['clip_id','file','core_review_start','core_review_end','actual_review_start','actual_review_end', 'frames','duration_seconds','context_before_seconds','context_after_seconds','observation_ids', 'pts_gap','sha256']) for spec,result in zip(manifest['clips'],results): writer.writerow([spec['id'],spec['output'],spec['core_start'],spec['core_end'],spec['start'],spec['end'], result['frames'],f"{result['output_duration_seconds']:.6f}", f"{spec['context_before_seconds']:.6f}",f"{spec['context_after_seconds']:.6f}", ';'.join(spec['core_observation_ids']),spec['contains_pts_gap'],result['sha256']]) sample_values = [s['rgb_psnr_db'] for s in pixel_report['samples'] if s['rgb_psnr_db'] is not None] technical = dict(status='PASS',individual_clips=22,continuous_clips=1,total_videos=len(results), decoded_total_frames=sum(r['frames'] for r in results),all_source_selection_exact=True, max_relative_pts_error_ticks=max(r['relative_pts_error_ticks'] for r in results), max_abs_container_duration_delta_seconds=max(abs(r['duration_delta_seconds']) for r in results), all_decode_errors_empty=all(not r['decode_errors'] for r in results), source_after_sha256=b.digest(b.SOURCE),source_map_after_sha256=b.digest(b.SOURCE_MAP), observation_snapshot_sha256=b.digest(b.OBS),pixel_sample_count=pixel_report['count'], pixel_sample_min_psnr_db=min(sample_values,default=None),no_video_semantic_analysis=True,no_inference=True, no_runtime_changes=True,selected_source_gap_preserved_ticks=1016668, helpers={p.name:b.digest(p) for p in sorted((b.ROOT/'helpers').glob('*.py'))},results=results) assert technical['source_after_sha256']==manifest['source_sha256'] assert technical['source_map_after_sha256']==manifest['source_map_sha256'] assert technical['observation_snapshot_sha256']==manifest['observations_sha256'] b.save(b.META/'validation/completion.json',technical) readme = '''아이서티 · 컨텍스트라 · 싸움/쓰러짐 자체촬영 원본01 클립자료 준비 완료: 사례별 무표시 클립22개 + 연속본1개 = 총23개. clips/: C01~C22. continuous/: L01. 영상목록.csv는 Windows용 UTF-8 BOM이다. 모든 영상은 번호/bbox/skeleton 없는 입력용이다. 원본1280x720·가변Frame간격·정상속도를 유지했다. CPU libx264 veryfast/CRF18/B-frame0으로 재인코딩했으므로 원본 비트스트림/픽셀 완전 동일본은 아니다. 10FPS 변환·Frame 복제/보간·배속·의미 분석·모델 추론·RTSP 송출은 하지 않았다. 번호 읽는 법 - 클립의1-based Frame은 metadata/frame_maps/Cxx.csv에서 기존 검토Frame 및 원본PTS로 연결된다. - 기존 검토Frame은 오류구간 제외 검토본과 clean_joined의 번호다. 원본 전체 파일 번호가 아니다. - source_packet_1based_not_decoded_frame은 원본packet 참조번호다. 디코딩Frame 번호로 읽지 않는다. - core/context는 추출 범위 구분이며 정확한 행동 GT가 아니다. context에도 인접 사례의 행동이 포함될 수 있다. - 중첩되는 클립과L01은 같은 원본을 공유한다. 독립 사례 수나 독립 정확도 검증으로 중복 집계하지 않는다. 시각 경계와 클립 예외 - C07은14504에서 끝나고 C08은14505에서 시작한다. O09의 앞3Frame은 C07에 있으며 원래 관찰 범위는 보존했다. - C07 마지막 Frame의 약1.016668초 표시시간도 원본 그대로다. 복제 Frame은 없으며 이 시간을 추가 관측으로 세지 않는다. - L01은11963~31937 전체이며 14504/14505의 약1.016668초 간격을 그대로 유지했다. 빈 Frame을 만들어 메우지 않았다. - L01 반복 송출의 끝→처음은 실제 연속 행동이 아니다. 이번에는 파일만 준비했고 송출/Tracker 설정은 변경하지 않았다. - C01은 불필요한1~3093을 포함하지 않도록 앞 여유가 없다. C22는 판단 대기 범위를 독립 보존해 앞 여유를 추가하지 않았다. - O18/O19의 미설명 대기 부분 일부는 개별 클립을 늘리지 않고 L01에 보존했다. 정확한 범위는 observation_clip_links.json을 본다. 관찰과기대 - metadata/observations의 문서는 사용자 관찰 당시 snapshot이다. 문서의 클립 미생성 상태는 당시 기록이고, 현재 완료 상태는 validation/completion.json이 소유한다. - 관찰 snapshot 내부의 상대 문서 링크는 원래 Git 저장소 기준이다. 이 다운로드 폴더의 사용 안내와 대응표 위치는 본 README를 따른다. - a=ltb/b=mhj는 사용자 인물 기호다. 모델이 항상 같은 수로 탐지하거나 T ID를 유지한다는 정답이 아니다. - Frame 부근/전후는 대략적인 연속 행동 anchor다. 실제 관측·사용자 설명·추정을 구분하고 중요한 충돌만 질문한다. - 쓰러짐은 전체 비감지 참고다. 싸움은 지정 상호작용 감지 희망/그 외 비감지 참고이며, O20(C22구간)의 싸움만 사용자 판단 대기다. - 혼자 정권지르기는 싸움 비감지/높은 우선도, 허공 밀치기는 비감지/낮은 우선도, 상대 어깨 흔들기는 감지 희망이다. - 필수 수용 기준은 아직 없다. 같은 클립에 감지 희망/비감지 참고가 섞일 수 있으므로 파일 하나를 단일 정답 label로 쓰지 않는다. 다음작업 - Pose/Tracking cache는 이 무표시 영상과 대응표를 기준으로 별도 요청 후 준비한다. 현재 cache는 새로 만들지 않았다. - 기술 검증: 전체 디코딩·source 선택 PTS·출력 상대 PTS·색상/해상도·대응표·9개 pixel 표본과 SHA256. - pixel PSNR은 재인코딩된 동일 시점의 기술 대조이며 인물/행동/모델 정확도 평가가 아니다. 원본 내용을 눈으로 판정하지 않았다. - SHA256SUMS.json은 이 파일을 포함한 배포 파일을 검증한다(자기 자신 제외). 원본 전체 파일과 무거운 cache/가중치는 넣지 않았다. ''' readme += '\n클립 목록 (숫자는 기존 검토본 Frame, 양 끝 포함)\n' for spec,result in zip(manifest['clips'],results): readme += (f"{spec['id']} {spec['name']}\n" f" 파일: {spec['output']}\n" f" 실제 {spec['start']}~{spec['end']} / 핵심 범위 {spec['core_start']}~{spec['core_end']} / " f"{result['output_duration_seconds']:.3f}초 / 관찰 {','.join(spec['core_observation_ids'])}\n") (b.DOWNLOAD/'00_README_ko.txt').write_text(readme,encoding='utf-8') hashes = {str(p.relative_to(b.DOWNLOAD)):dict(bytes=p.stat().st_size,sha256=b.digest(p)) for p in sorted(b.DOWNLOAD.rglob('*')) if p.is_file()} b.save(b.DOWNLOAD/'SHA256SUMS.json',dict(files=hashes,excluded_self=True)) for relative,item in hashes.items(): path = b.DOWNLOAD/relative assert path.stat().st_size==item['bytes'] and b.digest(path)==item['sha256'] samples = [] for p in b.REPORT.glob('*.resources.json'): x=b.load(p) samples.extend([x['before'],*x['samples'],x['after']]) summary = dict(status='COMPLETE',download=str(b.DOWNLOAD),videos=len(results), verified_files=len(hashes),total_bytes=sum(x['bytes'] for x in hashes.values()), checksum_manifest_sha256=b.digest(b.DOWNLOAD/'SHA256SUMS.json'), root_free_min_bytes=min(s['root_free'] for s in samples),ram_available_min_bytes=min(s['ram_available'] for s in samples), observed_child_rss_max_bytes=max(s.get('child_rss',0) for s in samples), max_load_1m=max(s['load'][0] for s in samples),cpu_affinity=b.CPUS,nice=10, own_ffmpeg_ffprobe_children_reaped=True,runtime_performance_measured=False) b.save(b.ROOT/'checkpoint_completed.json',summary) print(json.dumps(summary,ensure_ascii=False,indent=2),flush=True) if __name__=='__main__': main()