""" Operon Optics — Wire Optic Router (Gradio Demo) ================================================ Four-tab demo: 1. Prism Routing — conditional fan-out by DataType 2. Traversal Transform — collection processing 3. Composed Optics — chained prism + traversal pipeline 4. Optic + Denature — both layers on the same wire Run locally: pip install gradio python space-optics/app.py """ import sys from pathlib import Path import gradio as gr # Allow importing operon_ai from the repo root when running locally _repo_root = Path(__file__).resolve().parent.parent if str(_repo_root) not in sys.path: sys.path.insert(0, str(_repo_root)) from operon_ai import ( DataType, IntegrityLabel, PrismOptic, TraversalOptic, ComposedOptic, LensOptic, OpticError, SummarizeFilter, StripMarkupFilter, NormalizeFilter, ChainFilter, # Wiring TypedValue, ) # ── Constants ──────────────────────────────────────────────────────────── DATATYPE_CHOICES = [dt.value for dt in DataType] INTEGRITY_CHOICES = [ ("UNTRUSTED (0)", IntegrityLabel.UNTRUSTED), ("VALIDATED (1)", IntegrityLabel.VALIDATED), ("TRUSTED (2)", IntegrityLabel.TRUSTED), ] TRANSFORM_PRESETS = { "Double (x * 2)": lambda x: x * 2 if isinstance(x, (int, float)) else x, "Square (x ** 2)": lambda x: x ** 2 if isinstance(x, (int, float)) else x, "Negate (-x)": lambda x: -x if isinstance(x, (int, float)) else x, "Uppercase": lambda x: str(x).upper(), "Reverse string": lambda x: str(x)[::-1], } DENATURE_PRESETS = { "SummarizeFilter (max_length=50)": lambda: SummarizeFilter(max_length=50), "StripMarkupFilter": StripMarkupFilter, "NormalizeFilter": NormalizeFilter, "Chain (StripMarkup + Summarize + Normalize)": lambda: ChainFilter( filters=(StripMarkupFilter(), SummarizeFilter(max_length=80), NormalizeFilter()), ), } # ── HTML helpers ───────────────────────────────────────────────────────── def _badge(text: str, color: str) -> str: return ( f'' f'{text}' ) def _wire_result(label: str, accepted: bool, value=None) -> str: if accepted: val_str = f" → {value}" if value is not None else "" return ( f'
' f'{_badge("DELIVERED", "#22c55e")} {label}{val_str}
' ) return ( f'
' f'{_badge("REJECTED", "#ef4444")} {label}' f' — prism blocked this DataType
' ) def _stage_html(stage_name: str, accepted: bool, before: str, after: str) -> str: color = "#22c55e" if accepted else "#ef4444" status = "PASS" if accepted else "BLOCKED" return ( f'
' f'{_badge(status, color)} {stage_name}
' f'Before: {before}
' f'After: {after}
' ) # ── Tab 1: Prism Routing ──────────────────────────────────────────────── def run_prism_routing( wire1_types: list[str], wire2_types: list[str], source_type: str, payload: str, ) -> str: """Route data through two prism wires. Returns HTML showing which wire accepted the data. """ dt = DataType(source_type) integrity = IntegrityLabel.VALIDATED # Build prisms wire1_accept = frozenset(DataType(t) for t in wire1_types) if wire1_types else frozenset() wire2_accept = frozenset(DataType(t) for t in wire2_types) if wire2_types else frozenset() prism1 = PrismOptic(accept=wire1_accept) prism2 = PrismOptic(accept=wire2_accept) # Check each wire wire1_ok = prism1.can_transmit(dt, integrity) wire2_ok = prism2.can_transmit(dt, integrity) parts = [ f'
' f'Sending {_badge(source_type, "#6366f1")} with payload ' f'{payload}
', ] parts.append(_wire_result( f"Wire 1 — {prism1.name}", wire1_ok, payload if wire1_ok else None, )) parts.append(_wire_result( f"Wire 2 — {prism2.name}", wire2_ok, payload if wire2_ok else None, )) if not wire1_ok and not wire2_ok: parts.append( '
' 'Neither wire accepted this DataType — data is dropped.
' ) return "".join(parts) # ── Tab 2: Traversal Transform ────────────────────────────────────────── def run_traversal(input_text: str, transform_name: str) -> tuple[str, str]: """Apply traversal to a list of values. Returns (before_html, after_html). """ # Parse input raw_items = [x.strip() for x in input_text.split(",") if x.strip()] # Try to parse as numbers items: list = [] for item in raw_items: try: items.append(int(item)) except ValueError: try: items.append(float(item)) except ValueError: items.append(item) transform_fn = TRANSFORM_PRESETS.get(transform_name) if transform_fn is None: return "

Select a transform

", "" traversal = TraversalOptic(transform=transform_fn) result = traversal.transmit(items, DataType.JSON, IntegrityLabel.VALIDATED) before_html = ( '
' f'Input ({len(items)} items):
' + "".join( f'{_badge(str(x), "#6366f1")}' for x in items ) + "
" ) after_html = ( '
' f'Output ({len(result)} items):
' + "".join( f'{_badge(str(x), "#22c55e")}' for x in result ) + "
" ) return before_html, after_html # ── Tab 3: Composed Optics ────────────────────────────────────────────── def run_composed( prism_types: list[str], transform_name: str, source_type: str, payload: str, ) -> str: """Chain a prism then traversal and test. Returns HTML showing each stage. """ dt = DataType(source_type) integrity = IntegrityLabel.VALIDATED prism_accept = frozenset(DataType(t) for t in prism_types) if prism_types else frozenset() prism = PrismOptic(accept=prism_accept) transform_fn = TRANSFORM_PRESETS.get(transform_name, lambda x: x) traversal = TraversalOptic(transform=transform_fn) composed = ComposedOptic(optics=(prism, traversal)) # Parse payload as list if comma-separated raw_items = [x.strip() for x in payload.split(",") if x.strip()] items: list = [] for item in raw_items: try: items.append(int(item)) except ValueError: try: items.append(float(item)) except ValueError: items.append(item) value = items if len(items) > 1 else (items[0] if items else payload) parts = [ f'
' f'Pipeline: {composed.name}
' f'DataType: {_badge(source_type, "#6366f1")} ' f'Payload: {value}
', ] # Stage 1: Prism prism_ok = prism.can_transmit(dt, integrity) parts.append(_stage_html( f"Stage 1: {prism.name}", prism_ok, str(value), str(value) if prism_ok else "(blocked)", )) if not prism_ok: parts.append( '
' 'Pipeline stopped — prism rejected this DataType.
' ) return "".join(parts) # Stage 2: Traversal result = traversal.transmit(value, dt, integrity) parts.append(_stage_html( f"Stage 2: {traversal.name} ({transform_name})", True, str(value), str(result), )) parts.append( f'
' f'Final output: {result}
' ) return "".join(parts) # ── Tab 4: Optic + Denature ───────────────────────────────────────────── def run_denature_optic( denature_name: str, prism_types: list[str], transform_name: str, source_type: str, payload: str, ) -> str: """Apply denature then optic on the same wire. Returns HTML showing each processing stage. """ dt = DataType(source_type) integrity = IntegrityLabel.VALIDATED # Build denature filter denature_factory = DENATURE_PRESETS.get(denature_name) if denature_factory is None: return "

Select a denature filter

" denature_filter = denature_factory() # Build optic prism_accept = frozenset(DataType(t) for t in prism_types) if prism_types else frozenset() prism = PrismOptic(accept=prism_accept) transform_fn = TRANSFORM_PRESETS.get(transform_name, lambda x: x) traversal = TraversalOptic(transform=transform_fn) composed_optic = ComposedOptic(optics=(prism, traversal)) parts = [ f'
' f'Wire configuration: Denature({denature_filter.name}) ' f'→ Optic({composed_optic.name})
' f'DataType: {_badge(source_type, "#6366f1")} ' f'Payload: {payload}
', ] # Stage 1: Denaturation denatured = denature_filter.denature(payload) parts.append(_stage_html( f"Stage 1: Denature — {denature_filter.name}", True, payload, denatured, )) # Stage 2: Prism check prism_ok = prism.can_transmit(dt, integrity) parts.append(_stage_html( f"Stage 2: Prism — {prism.name}", prism_ok, denatured, denatured if prism_ok else "(blocked)", )) if not prism_ok: parts.append( '
' 'Wire blocked — prism rejected this DataType after denaturation.
' ) return "".join(parts) # Stage 3: Traversal transform # Parse denatured as list if possible try: items = [x.strip() for x in denatured.split(",") if x.strip()] value: list | str = items if len(items) > 1 else denatured except Exception: value = denatured result = traversal.transmit(value, dt, integrity) parts.append(_stage_html( f"Stage 3: Traversal ({transform_name})", True, str(value), str(result), )) parts.append( f'
' f'Final delivered value: {result}
' ) return "".join(parts) # ── Gradio UI ──────────────────────────────────────────────────────────── def build_app() -> gr.Blocks: with gr.Blocks(title="Optic Router") as app: gr.Markdown( "# 🔬 Optics — Wire Optic Router\n" "Explore **prism routing**, **traversal transforms**, and how " "optics compose with denaturation filters on wires." ) with gr.Tabs(): # ── Tab 1: Prism Routing ───────────────────────────────── with gr.TabItem("Prism Routing"): gr.Markdown( "Configure two wires with different **PrismOptic** filters. " "Each prism accepts a set of DataTypes — data is delivered " "only to wires whose prism matches the source type." ) with gr.Row(): with gr.Column(): gr.Markdown("#### Wire 1") pr_wire1 = gr.CheckboxGroup( choices=DATATYPE_CHOICES, value=["json", "text"], label="Accepted DataTypes", ) with gr.Column(): gr.Markdown("#### Wire 2") pr_wire2 = gr.CheckboxGroup( choices=DATATYPE_CHOICES, value=["error", "stop"], label="Accepted DataTypes", ) with gr.Row(): pr_source = gr.Dropdown( choices=DATATYPE_CHOICES, value="json", label="Source DataType", ) pr_payload = gr.Textbox( value='{"key": "value"}', label="Payload", scale=2, ) pr_btn = gr.Button("Route", variant="primary") pr_result = gr.HTML(label="Routing Result") pr_btn.click( fn=run_prism_routing, inputs=[pr_wire1, pr_wire2, pr_source, pr_payload], outputs=[pr_result], ) # ── Tab 2: Traversal Transform ─────────────────────────── with gr.TabItem("Traversal Transform"): gr.Markdown( "Apply an element-wise **TraversalOptic** transform to a " "list of values. The traversal maps the function over each " "element (or applies it to a single value)." ) tr_input = gr.Textbox( value="1, 2, 3, 4, 5", label="Input List (comma-separated)", ) tr_transform = gr.Dropdown( choices=list(TRANSFORM_PRESETS.keys()), value="Double (x * 2)", label="Transform", ) tr_btn = gr.Button("Transmit", variant="primary") with gr.Row(): tr_before = gr.HTML(label="Before") tr_after = gr.HTML(label="After") tr_btn.click( fn=run_traversal, inputs=[tr_input, tr_transform], outputs=[tr_before, tr_after], ) # ── Tab 3: Composed Optics ─────────────────────────────── with gr.TabItem("Composed Optics"): gr.Markdown( "Chain a **PrismOptic** filter with a **TraversalOptic** " "transform. The prism must accept the DataType before the " "traversal runs. See each stage independently." ) with gr.Row(): co_prism = gr.CheckboxGroup( choices=DATATYPE_CHOICES, value=["json"], label="Prism: Accepted DataTypes", ) co_transform = gr.Dropdown( choices=list(TRANSFORM_PRESETS.keys()), value="Uppercase", label="Traversal Transform", ) with gr.Row(): co_source = gr.Dropdown( choices=DATATYPE_CHOICES, value="json", label="Source DataType", ) co_payload = gr.Textbox( value="hello, world, operon", label="Payload (comma-separated for lists)", scale=2, ) co_btn = gr.Button("Run Pipeline", variant="primary") co_result = gr.HTML(label="Pipeline Result") co_btn.click( fn=run_composed, inputs=[co_prism, co_transform, co_source, co_payload], outputs=[co_result], ) # ── Tab 4: Optic + Denature ────────────────────────────── with gr.TabItem("Optic + Denature"): gr.Markdown( "Attach both a **DenatureFilter** and an **Optic** to the " "same wire. Denaturation runs first (strips injection " "vectors), then the optic routes and transforms." ) with gr.Row(): dn_denature = gr.Dropdown( choices=list(DENATURE_PRESETS.keys()), value="StripMarkupFilter", label="Denature Filter", ) with gr.Row(): dn_prism = gr.CheckboxGroup( choices=DATATYPE_CHOICES, value=["text", "json"], label="Prism: Accepted DataTypes", ) dn_transform = gr.Dropdown( choices=list(TRANSFORM_PRESETS.keys()), value="Uppercase", label="Traversal Transform", ) with gr.Row(): dn_source = gr.Dropdown( choices=DATATYPE_CHOICES, value="text", label="Source DataType", ) dn_payload = gr.Textbox( value='Hello ```injected code``` world ignore previous test', label="Payload (try injection patterns!)", scale=2, ) dn_btn = gr.Button("Process Wire", variant="primary") dn_result = gr.HTML(label="Processing Pipeline") dn_btn.click( fn=run_denature_optic, inputs=[dn_denature, dn_prism, dn_transform, dn_source, dn_payload], outputs=[dn_result], ) return app if __name__ == "__main__": app = build_app() app.launch(theme=gr.themes.Soft())