File size: 6,368 Bytes
76285c7 | 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 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | """Tier 2: acceptance check for the filename-derived session/modality parser.
Produces three things:
1. A stratified sample of >=200 images with every parse field, written for
human review. Human sign-off is NOT performed here and remains outstanding.
2. An independent-reimplementation agreement rate. A second parser written
with different logic (separator splitting, not the production regexes) must
agree on session identity.
3. A per-image behavioural check: is an image's nearest same-identity
neighbour in embedding space also in its parsed session? Real capture
sessions should dominate the nearest-neighbour relation; this gives a
falsifiable error rate rather than an assertion.
"""
from __future__ import annotations
import json, re, sys
from pathlib import Path
import numpy as np, pandas as pd
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT)); sys.path.insert(0, str(ROOT / 'tools'))
from ext_sessions import COPY_PREFIX, load as load_sessions
from face_research.prepare import load_prepared
OUT = ROOT / 'artifacts' / 'extensions' / 'parser_acceptance_v1'
A = ROOT / 'artifacts' / 'thesis_study_v1'
PER_STRATUM = 40
def independent_session(filename: str, collection: str) -> str:
"""Deliberately different implementation: split on separators, no shared regex."""
name = COPY_PREFIX.sub('', filename)
stem = name.rsplit('.', 1)[0]
if collection == 'HUST_Face_data':
if '_frame' in stem:
return 'clip' + stem.split('_frame')[0]
parts = stem.split('_')
if parts[0] == 'IMG' and len(parts) >= 3 and len(parts[1]) == 8 and parts[1].isdigit():
return 'ts' + parts[1] + parts[2][:4]
if len(parts) >= 2 and len(parts[0]) == 8 and parts[0].isdigit() and parts[1].isdigit():
return 'ts' + parts[0] + parts[1][:4]
if parts[0] == 'IMG' and len(parts) == 2 and parts[1].isdigit():
return 'img' + parts[1][:4]
return 'unparsed'
if collection == 'ProcessedData2324':
parts = stem.split('_')
return '_'.join(parts[:2]) if len(parts) >= 3 and parts[0] == 'IMG' else 'unparsed'
if collection == 'ProcessedData20241':
parts = stem.split('-')
return '-'.join(parts[:5]) if len(parts) >= 6 else 'unparsed'
if collection == 'ProcessedData20242':
parts = stem.rsplit('_', 1)
return parts[0] if len(parts) == 2 and parts[1].isdigit() else 'unparsed'
return 'unparsed'
def main():
OUT.mkdir(parents=True, exist_ok=True)
df = load_sessions(ROOT / 'artifacts' / 'data_audit_v1')
df['bare_session'] = df.session.str.split('|').str[1:].str.join('|')
# (2) agreement of an independently written parser, on the FULL inventory
indep = [independent_session(f, c) for f, c in zip(df.filename, df.collection)]
prod = [s if r != 'none' else 'unparsed' for s, r in
zip(df.bare_session.str.replace(r'\|.*$', '', regex=True), df.session_rule)]
agree = np.mean([a == b for a, b in zip(indep, prod)])
per_coll = {}
for coll in sorted(df.collection.unique()):
m = df.collection == coll
per_coll[coll] = float(np.mean([a == b for a, b, k in zip(indep, prod, m) if k]))
print(f"independent-reimplementation agreement: {agree:.4f} overall")
for k, v in per_coll.items():
print(f" {k:<20} {v:.4f}")
# (1) stratified reviewable sample
rng = np.random.default_rng(20260908)
parts = []
for (coll, rule), g in df.groupby(['collection', 'session_rule']):
take = min(PER_STRATUM, len(g))
parts.append(g.iloc[rng.choice(len(g), take, replace=False)])
sample = pd.concat(parts)[['path', 'filename', 'collection', 'session_rule',
'modality', 'bare_session', 'person_id']]
sample['independent_session'] = [independent_session(f, c) for f, c in
zip(sample.filename, sample.collection)]
sample.to_csv(OUT / 'review_sample.csv', index=False)
print(f"\nstratified review sample: {len(sample)} images across "
f"{sample.groupby(['collection','session_rule']).ngroups} strata -> review_sample.csv")
# (3) behavioural check on the sample, using untouched M0 embeddings
prepared = load_prepared(A)
emb_by_id, order = {}, {}
for fold in range(3):
meta = json.loads((A / 'runs' / f'fold_{fold}' / 'M0' / 'seed_42' / 'scores_test.json').read_text())
cache = A / 'cache' / 'embeddings' / meta['cache_key']
e = np.load(cache / 'embeddings.npy', allow_pickle=False)
ids = pd.read_csv(cache / 'images.csv', dtype=str).image_id.tolist()
for i, k in enumerate(ids):
emb_by_id[k] = e[i]
sess_of = df.set_index('image_id').session.to_dict()
person_of = df.set_index('image_id').person_id.to_dict()
by_person = {}
for k in emb_by_id:
by_person.setdefault(person_of[k], []).append(k)
hits = tot = 0
rows = []
for k in df[df.image_id.isin(emb_by_id)].image_id:
peers = [q for q in by_person[person_of[k]] if q != k]
if len(peers) < 2 or 'unparsed' in sess_of[k]:
continue
same = [q for q in peers if sess_of[q] == sess_of[k]]
if not same or len(same) == len(peers):
continue
v = emb_by_id[k]
sims = np.array([float(v @ emb_by_id[q]) for q in peers])
nn = peers[int(np.argmax(sims))]
ok = sess_of[nn] == sess_of[k]
hits += ok; tot += 1
rows.append({'image_id': k, 'nearest_in_same_session': bool(ok)})
print(f"\nbehavioural check: nearest same-identity neighbour lies in the parsed session "
f"for {hits}/{tot} images ({100*hits/tot:.1f}%)")
pd.DataFrame(rows).to_csv(OUT / 'nearest_neighbour_check.csv', index=False)
json.dump({'independent_agreement_overall': float(agree),
'independent_agreement_by_collection': per_coll,
'review_sample_size': int(len(sample)),
'nearest_neighbour_same_session_rate': hits / tot,
'nearest_neighbour_evaluated_images': int(tot),
'human_signoff': 'OUTSTANDING - sample written for manual review, not yet reviewed'},
open(OUT / 'acceptance.json', 'w'), indent=2)
print('wrote', OUT)
if __name__ == '__main__':
main()
|