""" Schema Registry Provides a centralized registry for managing annotation schema types. This module serves as the single source of truth for available annotation types and their generators, replacing the hardcoded dictionary in front_end.py. Usage: from potato.server_utils.schemas.registry import schema_registry # Get a schema generator generator = schema_registry.get("radio") # Generate layout for an annotation scheme html, keybindings = schema_registry.generate(annotation_scheme_dict) # List all available schemas schemas = schema_registry.list_schemas() """ from dataclasses import dataclass, field from typing import Callable, Dict, List, Tuple, Any, Optional import logging logger = logging.getLogger(__name__) @dataclass class SchemaDefinition: """ Defines metadata and generator for an annotation schema type. Attributes: name: Unique identifier for the schema type (e.g., "radio", "multiselect") generator: Callable that generates HTML and keybindings for this schema type required_fields: List of required configuration fields optional_fields: List of optional configuration fields supports_keybindings: Whether this schema type supports keyboard shortcuts description: Human-readable description of the schema type """ name: str generator: Callable[[Dict[str, Any]], Tuple[str, List[Tuple[str, str]]]] required_fields: List[str] = field(default_factory=list) optional_fields: List[str] = field(default_factory=list) supports_keybindings: bool = True description: str = "" class SchemaRegistry: """ Centralized registry for annotation schema types. Provides methods to register, retrieve, and list schema types, as well as generate layouts from annotation scheme configurations. """ def __init__(self): self._schemas: Dict[str, SchemaDefinition] = {} logger.debug("SchemaRegistry initialized") def register(self, schema: SchemaDefinition) -> None: """ Register a new schema type. Args: schema: SchemaDefinition to register Raises: ValueError: If a schema with the same name is already registered """ if schema.name in self._schemas: raise ValueError(f"Schema '{schema.name}' is already registered") self._schemas[schema.name] = schema logger.debug(f"Registered schema: {schema.name}") def get(self, name: str) -> Optional[SchemaDefinition]: """ Get a schema definition by name. Args: name: The schema type name Returns: SchemaDefinition if found, None otherwise """ return self._schemas.get(name) def get_generator(self, name: str) -> Optional[Callable]: """ Get the generator function for a schema type. Args: name: The schema type name Returns: Generator callable if found, None otherwise """ schema = self.get(name) return schema.generator if schema else None def generate(self, annotation_scheme: Dict[str, Any]) -> Tuple[str, List[Tuple[str, str]]]: """ Generate HTML and keybindings for an annotation scheme. Args: annotation_scheme: Configuration dictionary with 'annotation_type' key Returns: Tuple of (html_string, keybindings_list) Raises: ValueError: If annotation_type is missing or not supported """ import json as json_module annotation_type = annotation_scheme.get("annotation_type") if not annotation_type: raise ValueError("annotation_scheme must have 'annotation_type' field") schema = self.get(annotation_type) if not schema: supported = ", ".join(sorted(self._schemas.keys())) raise ValueError( f"Unsupported annotation type: '{annotation_type}'. " f"Supported types are: {supported}" ) logger.debug(f"Generating layout for annotation type: {annotation_type}") html, keybindings = schema.generator(annotation_scheme) # Wrap HTML with display_logic attributes if present html = self._wrap_with_display_logic(html, annotation_scheme) return html, keybindings def _wrap_with_display_logic(self, html: str, annotation_scheme: Dict[str, Any]) -> str: """ Wrap the generated HTML with display_logic attributes and container if needed. If the annotation scheme has display_logic, wraps the HTML in a container div with data-display-logic attribute containing the serialized logic, and applies the initial hidden state. Args: html: The generated HTML from the schema generator annotation_scheme: The annotation scheme configuration Returns: The HTML, possibly wrapped with display_logic container """ import json as json_module from html import escape display_logic = annotation_scheme.get("display_logic") if not display_logic: return html schema_name = annotation_scheme.get("name", "") # Serialize display_logic to JSON for the data attribute display_logic_json = json_module.dumps(display_logic) # Escape for HTML attribute (single quotes in JSON are safe) display_logic_escaped = escape(display_logic_json) # Wrap in a container with display logic attributes # Initially hidden - the frontend JS will evaluate and show if conditions are met wrapped_html = f'''
{html}
''' return wrapped_html def list_schemas(self) -> List[Dict[str, Any]]: """ List all registered schemas with their metadata. Returns: List of dictionaries containing schema metadata """ return [ { "name": schema.name, "description": schema.description, "required_fields": schema.required_fields, "optional_fields": schema.optional_fields, "supports_keybindings": schema.supports_keybindings, } for schema in sorted(self._schemas.values(), key=lambda s: s.name) ] def is_registered(self, name: str) -> bool: """ Check if a schema type is registered. Args: name: The schema type name Returns: True if registered, False otherwise """ return name in self._schemas def get_supported_types(self) -> List[str]: """ Get a list of all supported annotation types. Returns: Sorted list of annotation type names """ return sorted(self._schemas.keys()) # Global registry instance schema_registry = SchemaRegistry() def _register_builtin_schemas(): """ Register all built-in annotation schema types. Called automatically when this module is imported. """ from .radio import generate_radio_layout from .multiselect import generate_multiselect_layout from .multirate import generate_multirate_layout from .likert import generate_likert_layout from .textbox import generate_textbox_layout from .number import generate_number_layout from .slider import generate_slider_layout from .span import generate_span_layout from .span_link import generate_span_link_layout from .select import generate_select_layout from .pure_display import generate_pure_display_layout from .video import generate_video_layout from .image_annotation import generate_image_annotation_layout from .audio_annotation import generate_audio_annotation_layout from .video_annotation import generate_video_annotation_layout from .pairwise import generate_pairwise_layout from .coreference import generate_coreference_layout from .tree_annotation import generate_tree_annotation_layout from .triage import generate_triage_layout from .event_annotation import generate_event_annotation_layout from .tiered_annotation import generate_tiered_annotation_layout from .bws import generate_bws_layout from .soft_label import generate_soft_label_layout from .confidence import generate_confidence_layout from .constant_sum import generate_constant_sum_layout from .semantic_differential import generate_semantic_differential_layout from .ranking import generate_ranking_layout from .range_slider import generate_range_slider_layout from .hierarchical_multiselect import generate_hierarchical_multiselect_layout from .vas import generate_vas_layout from .extractive_qa import generate_extractive_qa_layout from .rubric_eval import generate_rubric_eval_layout from .text_edit import generate_text_edit_layout from .error_span import generate_error_span_layout from .card_sort import generate_card_sort_layout from .conjoint import generate_conjoint_layout from .trajectory_eval import generate_trajectory_eval_layout from .trajectory_edit import generate_trajectory_edit_layout from .process_reward import generate_process_reward_layout from .code_review import generate_code_review_layout schemas = [ SchemaDefinition( name="radio", generator=generate_radio_layout, required_fields=["name", "description", "labels"], optional_fields=["horizontal", "label_requirement", "sequential_key_binding", "has_free_response", "option_randomization", "dynamic_options", "dynamic_options_field"], supports_keybindings=True, description="Single-choice radio button selection" ), SchemaDefinition( name="multiselect", generator=generate_multiselect_layout, required_fields=["name", "description", "labels"], optional_fields=["display_config", "label_requirement", "sequential_key_binding", "video_as_label", "has_free_response", "option_randomization", "dynamic_options", "dynamic_options_field"], supports_keybindings=True, description="Multiple-choice checkbox selection" ), SchemaDefinition( name="multirate", generator=generate_multirate_layout, required_fields=["name", "description", "options", "labels"], optional_fields=["label_requirement"], supports_keybindings=False, description="Rate multiple items on a scale" ), SchemaDefinition( name="likert", generator=generate_likert_layout, required_fields=["name", "description", "min_label", "max_label", "size"], optional_fields=["label_requirement"], supports_keybindings=True, description="Likert scale rating" ), SchemaDefinition( name="text", generator=generate_textbox_layout, required_fields=["name", "description"], optional_fields=["label_requirement", "placeholder", "rows"], supports_keybindings=False, description="Free-form text input" ), SchemaDefinition( name="number", generator=generate_number_layout, required_fields=["name", "description"], optional_fields=["min", "max", "step", "label_requirement"], supports_keybindings=False, description="Numeric input field" ), SchemaDefinition( name="slider", generator=generate_slider_layout, required_fields=["name", "description", "min_value", "max_value", "starting_value"], optional_fields=["step", "label_requirement"], supports_keybindings=False, description="Slider for selecting a value in a range" ), SchemaDefinition( name="span", generator=generate_span_layout, required_fields=["name", "description", "labels"], optional_fields=["sequential_key_binding", "bad_text_label", "title", "allow_discontinuous", "entity_linking", "show_span_labels"], supports_keybindings=True, description="Text span annotation/highlighting with optional entity linking to knowledge bases" ), SchemaDefinition( name="select", generator=generate_select_layout, required_fields=["name", "description", "labels"], optional_fields=["label_requirement", "option_randomization", "dynamic_options", "dynamic_options_field"], supports_keybindings=False, description="Dropdown selection" ), SchemaDefinition( name="pure_display", generator=generate_pure_display_layout, required_fields=["name", "description"], optional_fields=["labels", "allow_html"], supports_keybindings=False, description="Display-only content (instructions, headers)" ), SchemaDefinition( name="video", generator=generate_video_layout, required_fields=["name", "description", "video_path"], optional_fields=["autoplay", "loop", "muted", "controls", "custom_css", "fallback_text", "additional_sources"], supports_keybindings=False, description="Video player display" ), SchemaDefinition( name="image_annotation", generator=generate_image_annotation_layout, required_fields=["name", "description", "tools", "labels"], optional_fields=["zoom_enabled", "pan_enabled", "min_annotations", "max_annotations", "freeform_brush_size", "freeform_simplify"], supports_keybindings=True, description="Image annotation with bounding boxes, polygons, freeform drawing, and landmarks" ), SchemaDefinition( name="audio_annotation", generator=generate_audio_annotation_layout, required_fields=["name", "description"], optional_fields=["mode", "labels", "segment_schemes", "min_segments", "max_segments", "zoom_enabled", "playback_rate_control"], supports_keybindings=True, description="Audio segmentation and annotation with waveform visualization" ), SchemaDefinition( name="video_annotation", generator=generate_video_annotation_layout, required_fields=["name", "description"], optional_fields=["mode", "labels", "segment_schemes", "min_segments", "max_segments", "timeline_height", "overview_height", "zoom_enabled", "playback_rate_control", "frame_stepping", "show_timecode", "video_fps", "tracking_options"], supports_keybindings=True, description="Video annotation with temporal segments, frame classification, keyframes, and object tracking" ), SchemaDefinition( name="span_link", generator=generate_span_link_layout, required_fields=["name", "description", "link_types", "span_schema"], optional_fields=["visual_display"], supports_keybindings=False, description="Create relationships/links between spans (e.g., PERSON works_for ORGANIZATION)" ), SchemaDefinition( name="pairwise", generator=generate_pairwise_layout, required_fields=["name", "description"], optional_fields=["mode", "items_key", "items", "show_labels", "labels", "allow_tie", "tie_label", "sequential_key_binding", "scale", "label_requirement"], supports_keybindings=True, description="Pairwise comparison of two items (binary selection or scale rating)" ), SchemaDefinition( name="coreference", generator=generate_coreference_layout, required_fields=["name", "description", "span_schema"], optional_fields=["entity_types", "allow_singletons", "visual_display"], supports_keybindings=False, description="Coreference chain annotation for grouping mentions of the same entity" ), SchemaDefinition( name="tree_annotation", generator=generate_tree_annotation_layout, required_fields=["name", "description"], optional_fields=["node_scheme", "path_selection", "branch_comparison"], supports_keybindings=False, description="Annotation of conversation tree nodes with path selection" ), SchemaDefinition( name="triage", generator=generate_triage_layout, required_fields=["name", "description"], optional_fields=["accept_label", "reject_label", "skip_label", "auto_advance", "show_progress", "accept_key", "reject_key", "skip_key"], supports_keybindings=True, description="Binary accept/reject triage for rapid data curation" ), SchemaDefinition( name="event_annotation", generator=generate_event_annotation_layout, required_fields=["name", "description", "event_types", "span_schema"], optional_fields=["visual_display"], supports_keybindings=False, description="N-ary event annotation with triggers and typed arguments" ), SchemaDefinition( name="tiered_annotation", generator=generate_tiered_annotation_layout, required_fields=["name", "description", "tiers", "source_field"], optional_fields=["media_type", "tier_height", "show_tier_labels", "collapsed_tiers", "zoom_enabled", "playback_rate_control", "overview_height"], supports_keybindings=True, description="Hierarchical multi-tier annotation for audio/video (ELAN-style)" ), SchemaDefinition( name="bws", generator=generate_bws_layout, required_fields=["name", "description"], optional_fields=["best_description", "worst_description", "tuple_size", "sequential_key_binding", "label_requirement"], supports_keybindings=True, description="Best-Worst Scaling: select the best and worst item from a set" ), SchemaDefinition( name="soft_label", generator=generate_soft_label_layout, required_fields=["name", "description", "labels"], optional_fields=["total", "min_per_label", "show_distribution_chart"], supports_keybindings=False, description="Probability distribution across labels via constrained sliders" ), SchemaDefinition( name="confidence", generator=generate_confidence_layout, required_fields=["name", "description"], optional_fields=["target_schema", "scale_type", "scale_points", "labels", "min_value", "max_value", "step", "left_label", "right_label"], supports_keybindings=False, description="Confidence rating meta-annotation for any primary annotation" ), SchemaDefinition( name="constant_sum", generator=generate_constant_sum_layout, required_fields=["name", "description", "labels"], optional_fields=["total_points", "min_per_item", "input_type"], supports_keybindings=False, description="Allocate a fixed budget of points across categories" ), SchemaDefinition( name="semantic_differential", generator=generate_semantic_differential_layout, required_fields=["name", "description", "pairs"], optional_fields=["scale_points"], supports_keybindings=False, description="Bipolar adjective scales for measuring connotative meaning" ), SchemaDefinition( name="ranking", generator=generate_ranking_layout, required_fields=["name", "description", "labels"], optional_fields=["allow_ties"], supports_keybindings=False, description="Drag-and-drop ranking of items by preference or relevance" ), SchemaDefinition( name="range_slider", generator=generate_range_slider_layout, required_fields=["name", "description"], optional_fields=["min_value", "max_value", "step", "left_label", "right_label", "show_values"], supports_keybindings=False, description="Dual-thumb slider for selecting an acceptable range" ), SchemaDefinition( name="hierarchical_multiselect", generator=generate_hierarchical_multiselect_layout, required_fields=["name", "description", "taxonomy"], optional_fields=["auto_select_children", "auto_select_parent", "show_search", "max_selections"], supports_keybindings=False, description="Hierarchical tree-structured multi-label selection" ), SchemaDefinition( name="vas", generator=generate_vas_layout, required_fields=["name", "description"], optional_fields=["left_label", "right_label", "min_value", "max_value", "show_value"], supports_keybindings=False, description="Continuous visual analog scale for fine-grained magnitude estimation" ), SchemaDefinition( name="extractive_qa", generator=generate_extractive_qa_layout, required_fields=["name", "description"], optional_fields=["question_field", "passage_field", "allow_unanswerable", "highlight_color"], supports_keybindings=False, description="SQuAD-style extractive question answering with answer span highlighting" ), SchemaDefinition( name="rubric_eval", generator=generate_rubric_eval_layout, required_fields=["name", "description", "criteria"], optional_fields=["scale_points", "scale_labels", "show_overall"], supports_keybindings=False, description="Multi-criteria rubric evaluation grid for LLM and text quality assessment" ), SchemaDefinition( name="text_edit", generator=generate_text_edit_layout, required_fields=["name", "description"], optional_fields=["source_field", "show_diff", "show_edit_distance", "allow_reset"], supports_keybindings=False, description="Inline text editing with diff tracking for post-editing and correction tasks" ), SchemaDefinition( name="error_span", generator=generate_error_span_layout, required_fields=["name", "description", "error_types"], optional_fields=["severities", "show_score", "max_score"], supports_keybindings=False, description="MQM-style error span annotation with typed severity for quality evaluation" ), SchemaDefinition( name="card_sort", generator=generate_card_sort_layout, required_fields=["name", "description"], optional_fields=["mode", "groups", "items_field", "allow_empty_groups", "allow_multiple"], supports_keybindings=False, description="Drag-and-drop card sorting into predefined or user-created groups" ), SchemaDefinition( name="conjoint", generator=generate_conjoint_layout, required_fields=["name", "description"], optional_fields=["profiles_per_set", "attributes", "show_none_option", "profiles_field"], supports_keybindings=False, description="Discrete choice conjoint analysis with side-by-side profile comparison" ), SchemaDefinition( name="trajectory_eval", generator=generate_trajectory_eval_layout, required_fields=["name", "description"], optional_fields=["steps_key", "step_text_key", "correctness_options", "error_types", "severities", "show_score", "max_score"], supports_keybindings=False, description="Per-step trajectory evaluation with error taxonomy and severity scoring" ), SchemaDefinition( name="process_reward", generator=generate_process_reward_layout, required_fields=["name", "description"], optional_fields=["steps_key", "step_text_key", "mode"], supports_keybindings=False, description="Binary per-step process reward signals for PRM training" ), SchemaDefinition( name="trajectory_edit", generator=generate_trajectory_edit_layout, required_fields=["name", "description"], optional_fields=["steps_key", "step_text_key", "editable_fields", "show_diff", "show_edit_distance", "allow_reset", "require_reason_on_edit", "edit_final_answer", "final_answer_key"], supports_keybindings=False, description="Per-step trajectory correction/editing for SFT/DPO training data" ), SchemaDefinition( name="code_review", generator=generate_code_review_layout, required_fields=["name", "description"], optional_fields=["comment_categories", "verdict_options", "file_rating_dimensions"], supports_keybindings=False, description="GitHub PR-style code review with inline comments and file ratings" ), ] for schema in schemas: schema_registry.register(schema) logger.debug(f"Registered {len(schemas)} built-in schemas") # Auto-register built-in schemas on import _register_builtin_schemas()