"""P06 real-image segmentation baseline, Python 3.11+, standard library only. Run --help. Supplied implementation and public data are practice evidence. """ import argparse import hashlib import json import math import platform from pathlib import Path SOURCE_HASH = '329506078d0305e3261d91610fa9244b72b3cafe45365c26e08b036b943b18b7' SOURCE_BYTES = 181192 VERSION = 'p06-pixel-baseline-v1' def load_data(path): with Path(path).open('rb') as stream: raw = stream.read(SOURCE_BYTES + 1) if len(raw) != SOURCE_BYTES or hashlib.sha256(raw).hexdigest() != SOURCE_HASH: raise ValueError('Expected the unchanged licensed LearnAI pet subset.') data = json.loads(raw) audit(data['records']) return data def audit(records): """Exact source identity/content checks, not a perceptual near-duplicate detector.""" ids, hashes = set(), set() counts = {'training': 0, 'development': 0} for record in records: identity, digest = record['id'], record['sourceImage']['sha256'] if identity in ids or digest in hashes: raise ValueError('Duplicate source identity or source image bytes.') ids.add(identity) hashes.add(digest) split = record['split'] if split not in counts or record['sourceSplit'] != ('trainval' if split == 'training' else 'test'): raise ValueError('Unexpected source partition.') counts[split] += 1 n = record['width'] * record['height'] if not (1 <= record['width'] <= 48 and 1 <= record['height'] <= 48): raise ValueError('Image dimensions out of bounds.') if len(record['rgb']) != 3 * n or len(record['trimap']) != n: raise ValueError('Image and annotation dimensions disagree.') if any(type(v) is not int or not 0 <= v <= 255 for v in record['rgb']): raise ValueError('Invalid RGB values.') if any(type(v) is not int or v not in (1, 2, 3) for v in record['trimap']): raise ValueError('Invalid trimap.') if counts != {'training': 5, 'development': 3}: raise ValueError('Expected five training and three development source images.') return {'sourceImages': counts, 'exactDuplicates': 0, 'nearDuplicates': 'not checked; inspect images and provenance', 'evaluation': 'public repeated development; no untouched test'} def features(record, i, mode): width, height = record['width'], record['height'] return [*(v / 127.5 - 1 for v in record['rgb'][3*i:3*i+3]), 2 * ((i % width) + .5) / width - 1 if mode == 'color-position' else 0, 2 * ((i // width) + .5) / height - 1 if mode == 'color-position' else 0, 1] def prepare(records, mode, split): return [(features(r, i, mode), int(label == 1)) for r in records if r['split'] == split for i, label in enumerate(r['trimap']) if label != 3] def sigmoid(z): return 1 / (1 + math.exp(-z)) if z >= 0 else math.exp(z) / (1 + math.exp(z)) def objective(weights, rows): if not rows: raise ValueError('No classified pixels.') loss, gradient = 0., [0.] * 6 for x, y in rows: z = sum(w * v for w, v in zip(weights, x)) p = sigmoid(z) loss += max(z, 0) - z*y + math.log1p(math.exp(-abs(z))) for j in range(6): gradient[j] += (p-y) * x[j] / len(rows) return loss / len(rows), gradient def validate_model(model): if model.get('mode') not in ('color', 'color-position'): raise ValueError('Unknown feature mode.') weights = model.get('weights') if not isinstance(weights, list) or len(weights) != 6 or any( type(w) not in (int, float) or not math.isfinite(w) or abs(w) > 50 for w in weights): raise ValueError('Expected six bounded finite weights.') if model['mode'] == 'color' and weights[3:5] != [0, 0]: raise ValueError('Position weights must be zero in color mode.') if type(model.get('updates')) is not int or not 0 <= model['updates'] <= 1000: raise ValueError('Updates must be an integer from zero to 1000.') return model def train(records, mode='color', rate=.3, updates=100): if mode not in ('color', 'color-position') or type(updates) is not int or not 1 <= updates <= 1000: raise ValueError('Invalid mode or update count.') if type(rate) not in (int, float) or not math.isfinite(rate) or not 0 < rate <= 1: raise ValueError('Learning rate must be above zero and at most one.') rows = prepare(records, mode, 'training') model = {'mode': mode, 'weights': [0.] * 6, 'updates': 0} initial_loss = objective(model['weights'], rows)[0] for _ in range(updates): _, gradient = objective(model['weights'], rows) model = validate_model({'mode': mode, 'weights': [w-rate*g for w, g in zip(model['weights'], gradient)], 'updates': model['updates']+1}) return model, {'initialTrainingLoss': initial_loss, 'finalTrainingLoss': objective(model['weights'], rows)[0]} def score(labels, predictions): tp = fp = fn = correct = count = excluded = 0 for label, predicted in zip(labels, predictions): if label == 3: excluded += 1 continue pet = label == 1 count += 1 correct += int(pet == predicted) tp += int(pet and predicted) fp += int(not pet and predicted) fn += int(pet and not predicted) union = tp + fp + fn return {'classifiedPixels': count, 'excludedPixels': excluded, 'correctPixels': correct, 'accuracy': correct/count if count else None, 'petIntersection': tp, 'petUnion': union, 'petIoU': tp/union if union else None} def evaluate(records, model, threshold=.5): validate_model(model) if type(threshold) not in (int, float) or not math.isfinite(threshold) or not 0 < threshold < 1: raise ValueError('Threshold must be strictly between zero and one.') output = [] for r in records: prediction = [sigmoid(sum(w*x for w, x in zip(model['weights'], features(r, i, model['mode'])))) >= threshold for i in range(len(r['trimap']))] output.append({'id': r['id'], 'split': r['split'], 'model': score(r['trimap'], prediction), 'alwaysBackground': score(r['trimap'], [False]*len(prediction)), 'alwaysPet': score(r['trimap'], [True]*len(prediction)), 'predictedMask': [int(v) for v in prediction]}) return output def replay(data, artifact): if artifact.get('version') != VERSION or artifact.get('sourceHash') != SOURCE_HASH: raise ValueError('Wrong artifact version or source hash.') # Never trust imported losses, metrics, or assertions about training history. model = validate_model(artifact['model']) return {'version': VERSION, 'sourceHash': SOURCE_HASH, 'model': model, 'threshold': artifact['threshold'], 'audit': audit(data['records']), 'trainingLossRecomputed': objective(model['weights'], prepare(data['records'], model['mode'], 'training'))[0], 'images': evaluate(data['records'], model, artifact['threshold']), 'evidence': 'Replayed supplied baseline; weights do not authenticate training history.', 'independentReview': 'pending', 'freshSourceEvaluation': 'pending'} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--data', type=Path, default=Path(__file__).with_name('subset.json')) parser.add_argument('--mode', choices=['color', 'color-position'], default='color') parser.add_argument('--updates', type=int, default=100) parser.add_argument('--rate', type=float, default=.3) parser.add_argument('--threshold', type=float, default=.5) parser.add_argument('--replay', type=Path, help='Recompute a saved artifact; ignore its claimed scores.') parser.add_argument('--out', type=Path, required=True, help='New output file; existing files are not overwritten.') args = parser.parse_args() data = load_data(args.data) if args.out.exists(): parser.error('Output already exists. Choose a new filename.') if args.replay: with args.replay.open('rb') as stream: raw = stream.read(500001) if len(raw) > 500000: parser.error('Artifact exceeds 500 KB.') result = replay(data, json.loads(raw)) else: model, trace = train(data['records'], args.mode, args.rate, args.updates) result = replay(data, {'version': VERSION, 'sourceHash': SOURCE_HASH, 'model': model, 'threshold': args.threshold}) result['suppliedTrainingRun'] = {**trace, 'rate': args.rate, 'python': platform.python_version()} with args.out.open('x') as stream: json.dump(result, stream, indent=2, allow_nan=False) print(f'Wrote {args.out}; repeated development only. Review and fresh-source evaluation pending.') if __name__ == '__main__': main()