#!/usr/bin/env python3 """ Probe Tip Annotation Tool Shows cropped probe images and lets annotators click on the probe tip. Saves (x, y) in crop pixel coordinates. Usage: python3 annotate_tips.py --data-dir ./crop_cache [--port 5555] The --data-dir should contain: - manifest.json (generated by the pre-crop step) - *.png (cropped images) """ from flask import Flask, Response, request, jsonify, render_template_string, send_file import json import os from datetime import datetime import argparse import threading # HF repo-based persistence (optional, for Spaces deployment) # Use a separate DATASET repo so syncs don't trigger Space rebuilds HF_REPO_ID = os.environ.get('SPACE_ID') # auto-set on HF Spaces HF_DATA_REPO = 'chfeng/probe-tip-annotations-data' # dataset repo for annotations _hf_api = None _save_lock = threading.Lock() def _get_hf_api(): global _hf_api if _hf_api is None and HF_REPO_ID: try: from huggingface_hub import HfApi _hf_api = HfApi() except ImportError: pass return _hf_api def _sync_to_repo(annotator): """Upload annotation file to the HF Space repo for persistence.""" api = _get_hf_api() if not api: return path = os.path.join(ANNOTATION_DIR, f'{annotator}.json') if not os.path.exists(path): return try: api.upload_file( path_or_fileobj=path, path_in_repo=f'{annotator}.json', repo_id=HF_DATA_REPO, repo_type='dataset', commit_message=f'Update {annotator} annotations', ) except Exception as e: print(f'Warning: failed to sync {annotator}.json to repo: {e}') def _load_annotations_from_repo(): """On startup, download any existing annotations from the repo.""" api = _get_hf_api() if not api: return os.makedirs(ANNOTATION_DIR, exist_ok=True) for name in ANNOTATORS: try: path = api.hf_hub_download( repo_id=HF_DATA_REPO, filename=f'{name}.json', repo_type='dataset', ) import shutil shutil.copy(path, os.path.join(ANNOTATION_DIR, f'{name}.json')) print(f' Loaded {name}.json from repo') except Exception: pass app = Flask(__name__) DATA_DIR = './crop_cache' ANNOTATION_DIR = './tip_annotations' ANNOTATORS = ['andrew', 'ayush', 'xuanchen', 'chao'] manifest = {} def load_manifest(): global manifest path = os.path.join(DATA_DIR, 'manifest.json') with open(path) as f: manifest = json.load(f) print(f"Loaded manifest: {manifest['total']} samples") for name in ANNOTATORS: print(f" {name}: {len(manifest['assignments'][name])} images") def get_annotations(annotator): path = os.path.join(ANNOTATION_DIR, f'{annotator}.json') if os.path.exists(path): with open(path) as f: return json.load(f) return {} def save_annotation_file(annotator, data): path = os.path.join(ANNOTATION_DIR, f'{annotator}.json') with open(path, 'w') as f: json.dump(data, f, indent=2) threading.Thread(target=_sync_to_repo, args=(annotator,), daemon=True).start() # ─── Routes ─── @app.route('/') def index(): stats = {} for name in ANNOTATORS: anns = get_annotations(name) total = len(manifest['assignments'][name]) done = sum(1 for v in anns.values() if not v.get('skipped', False)) skipped = sum(1 for v in anns.values() if v.get('skipped', False)) stats[name] = { 'total': total, 'done': done, 'skipped': skipped, 'pct': int(100 * (done + skipped) / total) if total > 0 else 0 } return render_template_string(INDEX_HTML, stats=stats, annotators=ANNOTATORS) @app.route('/annotate/') def annotate(annotator): if annotator not in manifest['assignments']: return "Unknown annotator", 404 return render_template_string(ANNOTATE_HTML, annotator=annotator) @app.route('/api/samples/') def api_samples(annotator): if annotator not in manifest['assignments']: return jsonify({'error': 'Unknown annotator'}), 404 samples = manifest['assignments'][annotator] anns = get_annotations(annotator) return jsonify({ 'samples': samples, 'annotations': anns, 'total': len(samples), 'done': sum(1 for v in anns.values() if not v.get('skipped', False)), 'skipped': sum(1 for v in anns.values() if v.get('skipped', False)) }) @app.route('/api/crop/') def api_crop(stem): path = os.path.join(DATA_DIR, f'{stem}.png') if os.path.exists(path): return send_file(path, mimetype='image/png') return "Not found", 404 @app.route('/api/save', methods=['POST']) def api_save(): data = request.json annotator = data['annotator'] stem = data['stem'] tip_x = data['tip_x'] tip_y = data['tip_y'] meta = manifest['metadata'].get(stem) if not meta: return jsonify({'error': 'unknown image'}), 404 anns = get_annotations(annotator) anns[stem] = { 'tip_x': round(tip_x, 1), 'tip_y': round(tip_y, 1), 'crop_x1': meta['crop_x1'], 'crop_y1': meta['crop_y1'], 'crop_w': meta['crop_w'], 'crop_h': meta['crop_h'], 'img_w': meta['img_w'], 'img_h': meta['img_h'], 'full_x': round(meta['crop_x1'] + tip_x, 1), 'full_y': round(meta['crop_y1'] + tip_y, 1), 'timestamp': datetime.now().isoformat() } save_annotation_file(annotator, anns) done = sum(1 for v in anns.values() if not v.get('skipped', False)) return jsonify({'ok': True, 'done': done}) @app.route('/api/skip', methods=['POST']) def api_skip(): data = request.json annotator = data['annotator'] stem = data['stem'] anns = get_annotations(annotator) anns[stem] = {'skipped': True, 'timestamp': datetime.now().isoformat()} save_annotation_file(annotator, anns) done = sum(1 for v in anns.values() if not v.get('skipped', False)) return jsonify({'ok': True, 'done': done}) @app.route('/api/delete', methods=['POST']) def api_delete(): data = request.json annotator = data['annotator'] stem = data['stem'] anns = get_annotations(annotator) if stem in anns: del anns[stem] save_annotation_file(annotator, anns) done = sum(1 for v in anns.values() if not v.get('skipped', False)) return jsonify({'ok': True, 'done': done}) # ─── Inline HTML templates ─── INDEX_HTML = ''' Probe Tip Annotation

Probe Tip Annotation Tool

Click on the probe tip in each cropped image to annotate its (x, y) position

{% set total_done = stats.values()|map(attribute='done')|sum %} {% set total_all = stats.values()|map(attribute='total')|sum %} {% set total_skipped = stats.values()|map(attribute='skipped')|sum %}
Total: {{ total_all }} images
Annotated: {{ total_done }}
Skipped: {{ total_skipped }}
Remaining: {{ total_all - total_done - total_skipped }}
{% for name in annotators %} {% set s = stats[name] %}

{{ name }}

Progress: {{ s.pct }}%
Annotated: {{ s.done }} / {{ s.total }}
Skipped: {{ s.skipped }}
Remaining: {{ s.total - s.done - s.skipped }}
Start Annotating
{% endfor %}
''' ANNOTATE_HTML = ''' Annotate – {{ annotator }}
Home

{{ annotator }}

0 / 0 done
Loading images...
''' if __name__ == '__main__': parser = argparse.ArgumentParser(description='Probe Tip Annotation Tool') parser.add_argument('--port', type=int, default=5555) parser.add_argument('--data-dir', type=str, default='./crop_cache', help='Directory with manifest.json and crop PNGs') parser.add_argument('--annotation-dir', type=str, default='./tip_annotations', help='Directory to save annotation JSON files') args = parser.parse_args() DATA_DIR = os.path.abspath(args.data_dir) ANNOTATION_DIR = os.path.abspath(args.annotation_dir) os.makedirs(ANNOTATION_DIR, exist_ok=True) load_manifest() _load_annotations_from_repo() print(f"\nStarting server on http://0.0.0.0:{args.port}") print(f"Data: {DATA_DIR}") print(f"Annotations: {ANNOTATION_DIR}") print(f"HF Repo: {HF_REPO_ID or 'none (local mode)'}") app.run(host='0.0.0.0', port=args.port, debug=False)