Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import ast | |
| import math | |
| import re | |
| import time | |
| from dataclasses import dataclass | |
| from app.schemas.composition import ( | |
| AnnotationPanelSpec, | |
| Bounds3D, | |
| ChartEncoding, | |
| ChartPanelSpec, | |
| ChartSpec, | |
| CompositionAssertion, | |
| CompositionPayload, | |
| CompositionSpec, | |
| DatasetSpec, | |
| DerivedValueSpec, | |
| DiagramPanelSpec, | |
| EquationPanelSpec, | |
| ExpressionInstruction, | |
| ExpressionProgram, | |
| GeometryPresentation, | |
| ImplicitGeometrySpec, | |
| InteractionSpec, | |
| MathematicalProperty, | |
| ParameterConstraints, | |
| ParameterSpec, | |
| ParametricGeometrySpec, | |
| ProvenanceSpec, | |
| RegisteredGeometrySpec, | |
| RequirementCoverage, | |
| Resolution3D, | |
| SemanticDiagramEdge, | |
| SemanticDiagramNode, | |
| SemanticDiagramSpec, | |
| ThreeScenePanelSpec, | |
| VisualizationIntentRecord, | |
| VectorFieldGeometrySpec, | |
| ) | |
| from app.schemas.visual_lesson import ExtractedChartDataset | |
| class CompositionCompilationError(ValueError): | |
| code = "composition_compilation_failed" | |
| _FUNCTIONS = {"sin", "cos", "tan", "sqrt", "abs", "exp", "log", "min", "max"} | |
| _VARIABLES = {"x", "y", "z", "u", "v", "pi", "a", "b", "c", "e", "iso_value"} | |
| _BINOPS = { | |
| ast.Add: "add", ast.Sub: "subtract", ast.Mult: "multiply", ast.Div: "divide", ast.Pow: "power", | |
| } | |
| class SafeExpressionParser: | |
| """Turns user-supplied mathematics into nonrecursive, executable-free instructions.""" | |
| def _normalise(source: str) -> str: | |
| source = source.strip().replace("−", "-").replace("×", "*").replace("÷", "/") | |
| source = source.replace("²", "^2").replace("³", "^3").replace("π", "pi").replace("^", "**") | |
| source = re.sub(r"(?<=\d)(?=[a-zA-Z(])|(?<=[xyzuv)])(?=\()", "*", source) | |
| return source | |
| def parse_equation(self, source: str) -> ExpressionProgram: | |
| normalised = self._normalise(source) | |
| if "=" in normalised: | |
| left, right = normalised.split("=", 1) | |
| normalised = f"({left})-({right})" | |
| if len(normalised) > 500: | |
| raise CompositionCompilationError("The supplied expression is too long") | |
| try: | |
| tree = ast.parse(normalised, mode="eval") | |
| except SyntaxError as exc: | |
| raise CompositionCompilationError("The supplied equation could not be parsed") from exc | |
| instructions: list[ExpressionInstruction] = [] | |
| def emit(node: ast.AST) -> str: | |
| instruction_id = f"e{len(instructions)}" | |
| if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): | |
| value = float(node.value) | |
| if not math.isfinite(value) or abs(value) > 1e12: | |
| raise CompositionCompilationError("Expression constants must be finite and bounded") | |
| instructions.append(ExpressionInstruction(instruction_id=instruction_id, op="constant", value=value)) | |
| return instruction_id | |
| if isinstance(node, ast.Name) and node.id in _VARIABLES: | |
| instructions.append(ExpressionInstruction(instruction_id=instruction_id, op="variable", name=node.id)) | |
| return instruction_id | |
| if isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.USub): | |
| arg = emit(node.operand) | |
| instruction_id = f"e{len(instructions)}" | |
| instructions.append(ExpressionInstruction(instruction_id=instruction_id, op="negate", args=[arg])) | |
| return instruction_id | |
| if isinstance(node, ast.BinOp) and type(node.op) in _BINOPS: | |
| left, right = emit(node.left), emit(node.right) | |
| if isinstance(node.op, ast.Pow) and isinstance(node.right, ast.Constant) and abs(float(node.right.value)) > 12: | |
| raise CompositionCompilationError("Expression exponents are limited to an absolute value of 12") | |
| instruction_id = f"e{len(instructions)}" | |
| instructions.append(ExpressionInstruction(instruction_id=instruction_id, op=_BINOPS[type(node.op)], args=[left, right])) | |
| return instruction_id | |
| if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in _FUNCTIONS: | |
| args = [emit(item) for item in node.args] | |
| if node.func.id not in {"min", "max"} and len(args) != 1: | |
| raise CompositionCompilationError(f"{node.func.id} expects one argument") | |
| if node.func.id in {"min", "max"} and len(args) < 2: | |
| raise CompositionCompilationError(f"{node.func.id} expects at least two arguments") | |
| instruction_id = f"e{len(instructions)}" | |
| instructions.append(ExpressionInstruction(instruction_id=instruction_id, op=node.func.id, args=args)) | |
| return instruction_id | |
| raise CompositionCompilationError("The expression contains an unsupported construct") | |
| result_id = emit(tree.body) | |
| if len(instructions) > 128: | |
| raise CompositionCompilationError("The expression exceeds the 128-operation limit") | |
| return ExpressionProgram(instructions=instructions, result_id=result_id) | |
| class CompositionValidator: | |
| MAX_ROWS = 20_000 | |
| MAX_GRID_SAMPLES = 2_100_000 | |
| def validate(self, spec: CompositionSpec) -> None: | |
| parameter_ids = {item.parameter_id for item in spec.parameters} | |
| if len(parameter_ids) != len(spec.parameters): | |
| raise CompositionCompilationError("Composition parameter IDs must be unique") | |
| panel_ids = {panel.panel_id for panel in spec.panels} | |
| if len(panel_ids) != len(spec.panels) or not panel_ids: | |
| raise CompositionCompilationError("Composition panel IDs must be unique and nonempty") | |
| dataset_ids = {dataset.dataset_id for dataset in spec.datasets} | |
| for dataset in spec.datasets: | |
| if len(dataset.rows) > self.MAX_ROWS: | |
| raise CompositionCompilationError("Chart dataset exceeds the interactive row budget") | |
| if any(set(row) - set(dataset.columns) for row in dataset.rows): | |
| raise CompositionCompilationError("Chart rows contain undeclared columns") | |
| for panel in spec.panels: | |
| if set(panel.bindings) - (parameter_ids | {item.value_id for item in spec.derived_values}): | |
| raise CompositionCompilationError(f"Panel {panel.panel_id} references an unknown binding") | |
| if isinstance(panel, ChartPanelSpec) and panel.chart.dataset_id not in dataset_ids: | |
| raise CompositionCompilationError("Chart panel references an unknown dataset") | |
| if isinstance(panel, DiagramPanelSpec) and panel.diagram_type == "semantic": | |
| diagram = panel.semantic | |
| if diagram is None: | |
| raise CompositionCompilationError("Semantic diagram panels require a semantic diagram specification") | |
| if not 1 <= len(diagram.nodes) <= 80 or len(diagram.edges) > 160: | |
| raise CompositionCompilationError("Semantic diagram exceeds the interactive node or edge budget") | |
| node_ids = {node.node_id for node in diagram.nodes} | |
| if len(node_ids) != len(diagram.nodes): | |
| raise CompositionCompilationError("Semantic diagram node IDs must be unique") | |
| if any(edge.source not in node_ids or edge.target not in node_ids for edge in diagram.edges): | |
| raise CompositionCompilationError("Semantic diagram edge references an unknown node") | |
| if isinstance(panel, ChartPanelSpec): | |
| dataset = next(item for item in spec.datasets if item.dataset_id == panel.chart.dataset_id) | |
| available = set(dataset.columns) | |
| for transform in panel.chart.transforms: | |
| if transform.input_field and transform.input_field not in available: | |
| raise CompositionCompilationError("Chart transform references an unknown input field") | |
| if transform.kind == "bin" and not 2 <= transform.bins <= 200: | |
| raise CompositionCompilationError("Chart bin counts must stay between 2 and 200") | |
| if transform.output_field: | |
| available.add(transform.output_field) | |
| if isinstance(panel, ThreeScenePanelSpec) and isinstance(panel.geometry, ImplicitGeometrySpec): | |
| r = panel.geometry.resolution | |
| if min(r.preview, r.refined, r.high_quality) < 8 or max(r.preview, r.refined, r.high_quality) > 128: | |
| raise CompositionCompilationError("Implicit resolution must stay between 8 and 128") | |
| if r.refined ** 3 > self.MAX_GRID_SAMPLES: | |
| raise CompositionCompilationError("Implicit resolution exceeds the scalar-field budget") | |
| self.validate_program(panel.geometry.expression) | |
| if isinstance(panel, ThreeScenePanelSpec) and isinstance(panel.geometry, ParametricGeometrySpec): | |
| if panel.geometry.u_segments < 4 or panel.geometry.v_segments < 4 or panel.geometry.u_segments * panel.geometry.v_segments > 100_000: | |
| raise CompositionCompilationError("Parametric surface resolution exceeds its budget") | |
| self.validate_program(panel.geometry.x_program) | |
| self.validate_program(panel.geometry.y_program) | |
| self.validate_program(panel.geometry.z_program) | |
| if isinstance(panel, ThreeScenePanelSpec) and isinstance(panel.geometry, VectorFieldGeometrySpec): | |
| if not 2 <= panel.geometry.samples_per_axis <= 32: | |
| raise CompositionCompilationError("Vector-field sampling must stay between 2 and 32 per axis") | |
| self.validate_program(panel.geometry.x_program) | |
| self.validate_program(panel.geometry.y_program) | |
| self.validate_program(panel.geometry.z_program) | |
| for value in spec.derived_values: | |
| self.validate_program(value.program) | |
| for interaction in spec.interactions: | |
| if interaction.state_id not in parameter_ids and interaction.kind == "parameter": | |
| raise CompositionCompilationError("Parameter interaction references an unknown parameter") | |
| if set(interaction.target_panel_ids) - panel_ids: | |
| raise CompositionCompilationError("Interaction references an unknown panel") | |
| if spec.request_intent: | |
| covered = {item.requirement for item in spec.requirement_coverage if item.status == "satisfied"} | |
| missing = set(spec.request_intent.hard_requirements) - covered | |
| if missing: | |
| raise CompositionCompilationError( | |
| "Composition does not satisfy required visualization commands: " + ", ".join(sorted(missing)) | |
| ) | |
| if any(item.status != "satisfied" for item in spec.requirement_coverage): | |
| raise CompositionCompilationError("A required visualization command is unsupported or conflicting") | |
| def validate_program(program: ExpressionProgram) -> None: | |
| seen: set[str] = set() | |
| for instruction in program.instructions: | |
| if instruction.instruction_id in seen: | |
| raise CompositionCompilationError("Expression instruction IDs must be unique") | |
| if any(arg not in seen for arg in instruction.args): | |
| raise CompositionCompilationError("Expression instructions must form a forward-only DAG") | |
| if instruction.op == "constant" and (instruction.value is None or not math.isfinite(instruction.value)): | |
| raise CompositionCompilationError("Expression constants must be finite") | |
| if instruction.op == "variable" and instruction.name not in _VARIABLES: | |
| raise CompositionCompilationError("Expression variable is not approved") | |
| seen.add(instruction.instruction_id) | |
| if program.result_id not in seen: | |
| raise CompositionCompilationError("Expression result does not exist") | |
| def validate_checkpoint(self, spec: CompositionSpec, values: dict[str, object]) -> None: | |
| parameters = {item.parameter_id: item for item in spec.parameters} | |
| if set(values) - set(parameters): | |
| raise CompositionCompilationError("Checkpoint contains an unknown parameter") | |
| resolved = {item.parameter_id: item.initial_value for item in spec.parameters} | |
| resolved.update(values) | |
| for parameter_id, value in resolved.items(): | |
| parameter = parameters[parameter_id] | |
| if parameter.value_type == "boolean" and not isinstance(value, bool): | |
| raise CompositionCompilationError(f"{parameter.label} must be true or false") | |
| if parameter.value_type in {"scalar", "integer"}: | |
| if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)): | |
| raise CompositionCompilationError(f"{parameter.label} must be finite") | |
| number = float(value) | |
| c = parameter.constraints | |
| if c.minimum is not None and (number < c.minimum or (c.exclusive_minimum and number == c.minimum)): | |
| raise CompositionCompilationError(f"{parameter.label} is below its allowed range") | |
| if c.maximum is not None and (number > c.maximum or (c.exclusive_maximum and number == c.maximum)): | |
| raise CompositionCompilationError(f"{parameter.label} is above its allowed range") | |
| if parameter.value_type == "enum" and str(value) not in parameter.constraints.allowed_values: | |
| raise CompositionCompilationError(f"{parameter.label} is not an allowed option") | |
| operator = next((p.geometry for p in spec.panels if isinstance(p, ThreeScenePanelSpec) and isinstance(p.geometry, RegisteredGeometrySpec)), None) | |
| if operator and operator.operator == "prolate_spheroid": | |
| a = float(resolved[operator.parameter_bindings["majorSemiAxis"]]) | |
| b = float(resolved[operator.parameter_bindings["equatorialSemiAxis"]]) | |
| if not a > b > 0: | |
| raise CompositionCompilationError("A prolate spheroid requires a > b > 0") | |
| class ChartRendererDefinition: | |
| renderer: str | |
| families: frozenset[str] | |
| CHART_RENDERERS = ( | |
| ChartRendererDefinition("plotly", frozenset({"line", "scatter", "area", "bar", "histogram", "box", "violin", "heatmap", "contour", "error-band"})), | |
| ChartRendererDefinition("d3", frozenset({"force-graph", "tree", "sankey", "hierarchy", "chord", "treemap", "packed-circles"})), | |
| ChartRendererDefinition("svg", frozenset({"heatmap"})), | |
| ) | |
| def chart_renderer_for(spec: ChartSpec) -> str: | |
| candidates = [item.renderer for item in CHART_RENDERERS if spec.family in item.families] | |
| if spec.renderer_preference != "automatic" and spec.renderer_preference in candidates: | |
| return spec.renderer_preference | |
| if not candidates: | |
| raise CompositionCompilationError(f"No verified renderer supports {spec.family}") | |
| return candidates[0] | |
| class CompositionCompiler: | |
| GEOMETRY_TERMS = { | |
| "prolate_spheroid": ("prolate spheroid", "prolate ellipsoid"), | |
| "mobius_strip": ("möbius", "mobius"), | |
| "torus": ("torus", "toroid", "doughnut surface"), | |
| "ellipsoid": ("ellipsoid", "spheroid"), | |
| "sphere": ("sphere", "spherical surface"), | |
| } | |
| def __init__(self) -> None: | |
| self.parser = SafeExpressionParser() | |
| self.validator = CompositionValidator() | |
| def supports(cls, prompt: str) -> bool: | |
| lowered = prompt.lower() | |
| if any(term in lowered for terms in cls.GEOMETRY_TERMS.values() for term in terms): | |
| return True | |
| if any(term in lowered for term in ("line chart", "scatter plot", "bar chart", "heatmap", "force graph", "force-directed", "parabola", "quadratic curve", "quadratic function")): | |
| return True | |
| if all(re.search(rf"\b{name}\s*=", lowered) for name in ("x", "y", "z")) and any(variable in lowered for variable in ("u", "v")): | |
| return True | |
| return "=" in prompt and any(variable in lowered for variable in ("x", "y", "z")) | |
| def _common_parameters(prompt: str = "") -> list[ParameterSpec]: | |
| lowered = prompt.lower() | |
| opacity_match = re.search(r"\bopacity\s*(?:=|of|to)?\s*(0(?:\.\d+)?|1(?:\.0+)?)", lowered) | |
| opacity = float(opacity_match.group(1)) if opacity_match else 0.88 | |
| return [ | |
| ParameterSpec(parameter_id="opacity", label="Opacity", value_type="scalar", initial_value=opacity, constraints=ParameterConstraints(minimum=0.1, maximum=1), recompute="presentation-only"), | |
| ParameterSpec(parameter_id="wireframe", label="Wireframe", value_type="boolean", initial_value="wireframe" in lowered and not any(term in lowered for term in ("no wireframe", "without wireframe", "wireframe off")), recompute="presentation-only"), | |
| ParameterSpec(parameter_id="show_axes", label="Axes", value_type="boolean", initial_value=not any(term in lowered for term in ("hide axes", "without axes", "no axes", "axes off")), recompute="presentation-only"), | |
| ParameterSpec(parameter_id="spin", label="Spin", value_type="boolean", initial_value=any(term in lowered for term in ("spin", "rotate automatically", "auto-rotate", "auto rotate")), recompute="presentation-only"), | |
| ] | |
| def build(self, project_id: str, prompt: str, extracted: ExtractedChartDataset | None = None) -> CompositionPayload: | |
| lowered = prompt.lower() | |
| if extracted is not None or any(term in lowered for term in ("line chart", "scatter plot", "bar chart", "heatmap", "force graph", "force-directed", "parabola", "quadratic curve", "quadratic function")): | |
| spec = self._chart(project_id, prompt, extracted=extracted) | |
| elif all(re.search(rf"\b{name}\s*=", lowered) for name in ("x", "y", "z")) and any(variable in lowered for variable in ("u", "v")): | |
| spec = self._parametric(project_id, prompt) | |
| elif "=" in prompt and any(variable in lowered for variable in ("x", "y", "z")): | |
| spec = self._implicit(project_id, prompt) | |
| else: | |
| operator = next((name for name, terms in self.GEOMETRY_TERMS.items() if any(term in lowered for term in terms)), "") | |
| if not operator: | |
| raise CompositionCompilationError("The requested composition has no registered deterministic operator") | |
| spec = self._registered(project_id, prompt, operator) | |
| self.validator.validate(spec) | |
| return CompositionPayload(spec=spec, warnings=[]) | |
| def build_planned(self, project_id: str, prompt: str, kind: str, target: str, extracted: ExtractedChartDataset | None = None) -> CompositionPayload: | |
| if kind == "registered_geometry" and target in self.GEOMETRY_TERMS: | |
| spec = self._registered(project_id, prompt, target) | |
| elif kind == "chart" and target in {"line", "scatter", "bar", "heatmap", "force-graph"}: | |
| spec = self._chart(project_id, prompt, target, extracted=extracted) | |
| else: | |
| raise CompositionCompilationError("No registered composition operator can answer this request") | |
| self.validator.validate(spec) | |
| return CompositionPayload(spec=spec, warnings=[]) | |
| _REQUIREMENT_STOPWORDS = { | |
| "the", "a", "an", "and", "or", "with", "for", "of", "to", "in", "on", "at", "by", | |
| "show", "add", "include", "make", "generate", "create", "render", "featuring", | |
| "should", "must", "that", "this", "each", "every", "have", "has", "can", "using", | |
| } | |
| def _significant_words(cls, text: str) -> set[str]: | |
| words = re.findall(r"[a-z][a-z0-9]{2,}", text.lower()) | |
| return {word for word in words if word not in cls._REQUIREMENT_STOPWORDS} | |
| def _facet_text(cls, facet) -> str: | |
| parts = [facet.title, facet.purpose] | |
| parts.extend(node.label for node in facet.nodes) | |
| parts.extend(edge.label for edge in facet.edges if edge.label) | |
| return " ".join(parts) | |
| def build_semantic(self, project_id: str, prompt: str, intent, plan) -> CompositionPayload: | |
| """Compile an agent-selected semantic plan into app-owned diagram panels.""" | |
| panels: list[DiagramPanelSpec] = [] | |
| for order, facet in enumerate(plan.facets): | |
| nodes = [SemanticDiagramNode(**node.model_dump()) for node in facet.nodes] | |
| edges = [SemanticDiagramEdge(**edge.model_dump()) for edge in facet.edges] | |
| semantic = SemanticDiagramSpec( | |
| diagram_id=facet.facet_id, | |
| title=facet.title, | |
| layout=facet.layout, | |
| nodes=nodes, | |
| edges=edges, | |
| ) | |
| panels.append(DiagramPanelSpec( | |
| panel_id=facet.facet_id, | |
| diagram_type="semantic", | |
| title=facet.title, | |
| semantic=semantic, | |
| order=order, | |
| )) | |
| # A semantic panel has no authority to prove raw natural-language commands such | |
| # as "3D", "animation", or "plot". Those modes are resolved before this | |
| # compiler runs; keeping phrase-overlap coverage here created false acceptance | |
| # and false rejection alike. | |
| coverage: list[RequirementCoverage] = [] | |
| spec = CompositionSpec( | |
| project_id=project_id, | |
| prompt=prompt, | |
| title=plan.title, | |
| answer_markdown=plan.answer_markdown, | |
| parameters=[], | |
| panels=panels, | |
| interactions=[], | |
| assumptions=["Diagram geometry is schematic; nodes and edges represent the requested mechanism rather than measured spatial coordinates."], | |
| provenance=ProvenanceSpec(interpretation=plan.interpretation, research_used=False), | |
| request_intent=VisualizationIntentRecord( | |
| specificity=intent.specificity, | |
| interpretation=intent.interpretation, | |
| requested_views=intent.requested_views, | |
| hard_requirements=[], | |
| requested_interactions=intent.requested_interactions, | |
| ), | |
| requirement_coverage=coverage, | |
| created_at=time.time(), | |
| ) | |
| self.validator.validate(spec) | |
| return CompositionPayload(spec=spec, warnings=[]) | |
| def _registered(self, project_id: str, prompt: str, operator: str) -> CompositionSpec: | |
| parameters = self._common_parameters(prompt) | |
| bindings: dict[str, str] | |
| derived: list[DerivedValueSpec] = [] | |
| panels: list = [] | |
| properties: list[MathematicalProperty] = [] | |
| equations: list[str] = [] | |
| title = operator.replace("_", " ").title() | |
| assumptions = ["Dimensions are illustrative unless the prompt supplies explicit values."] | |
| if operator == "prolate_spheroid": | |
| a_match = re.search(r"\ba\s*=\s*(\d+(?:\.\d+)?)", prompt, flags=re.I) | |
| b_match = re.search(r"\bb\s*=\s*(\d+(?:\.\d+)?)", prompt, flags=re.I) | |
| axes_match = re.search(r"\b(?:semi[- ]?axes|axes)\s+(\d+(?:\.\d+)?)\s+(?:and|by|,|×|x)\s*(\d+(?:\.\d+)?)", prompt, flags=re.I) | |
| a_value = float(a_match.group(1)) if a_match else float(axes_match.group(1)) if axes_match else 2.0 | |
| b_value = float(b_match.group(1)) if b_match else float(axes_match.group(2)) if axes_match else 1.0 | |
| if a_value <= b_value: | |
| raise CompositionCompilationError("A prolate spheroid requires the major semi-axis a to be greater than b") | |
| parameters = [ | |
| ParameterSpec(parameter_id="a", label="Major semi-axis a", value_type="scalar", initial_value=a_value, constraints=ParameterConstraints(minimum=0.05, maximum=max(10, a_value), exclusive_minimum=True), recompute="geometry-transform", persistence="lesson"), | |
| ParameterSpec(parameter_id="b", label="Equatorial semi-axis b", value_type="scalar", initial_value=b_value, constraints=ParameterConstraints(minimum=0.05, maximum=max(10, b_value), exclusive_minimum=True), recompute="geometry-transform", persistence="lesson"), | |
| *parameters, | |
| ] | |
| bindings = {"majorSemiAxis": "a", "equatorialSemiAxis": "b"} | |
| derived = [ | |
| DerivedValueSpec(value_id="c", label="Focal distance", program=self.parser.parse_equation("sqrt(a^2-b^2)")), | |
| DerivedValueSpec(value_id="e", label="Eccentricity", program=self.parser.parse_equation("sqrt(a^2-b^2)/a")), | |
| DerivedValueSpec(value_id="volume", label="Volume", program=self.parser.parse_equation("4*pi*a*b^2/3")), | |
| ] | |
| equations = ["x^2/b^2 + y^2/b^2 + z^2/a^2 = 1", "c = sqrt(a^2-b^2)", "e = c/a", "V = 4*pi*a*b^2/3"] | |
| properties = [ | |
| MathematicalProperty(name="surface type", value="closed quadric", status="registered", method="Canonical prolate-spheroid operator"), | |
| MathematicalProperty(name="rotational symmetry", value=True, status="registered", method="Revolution about the major axis"), | |
| ] | |
| panels = [ | |
| ThreeScenePanelSpec(panel_id="shape", geometry=RegisteredGeometrySpec(operator="prolate_spheroid", parameter_bindings=bindings), bindings=["a", "b", "c", "e", "opacity", "wireframe", "show_axes", "spin"], order=0), | |
| DiagramPanelSpec(panel_id="cross-section", diagram_type="ellipse_cross_section", bindings=["a", "b", "c"], order=1), | |
| EquationPanelSpec(panel_id="equations", equations=equations, bindings=["a", "b", "c", "e", "volume"], order=2), | |
| ] | |
| else: | |
| parameters.insert(0, ParameterSpec(parameter_id="scale", label="Scale", value_type="scalar", initial_value=1.5, constraints=ParameterConstraints(minimum=0.1, maximum=8, exclusive_minimum=True), recompute="geometry-transform", persistence="lesson")) | |
| bindings = {"scale": "scale"} | |
| if operator == "ellipsoid": | |
| parameters[0:1] = [ | |
| ParameterSpec(parameter_id="a", label="Axis a", value_type="scalar", initial_value=2.0, constraints=ParameterConstraints(minimum=0.1, maximum=8), recompute="geometry-transform", persistence="lesson"), | |
| ParameterSpec(parameter_id="b", label="Axis b", value_type="scalar", initial_value=1.4, constraints=ParameterConstraints(minimum=0.1, maximum=8), recompute="geometry-transform", persistence="lesson"), | |
| ParameterSpec(parameter_id="c", label="Axis c", value_type="scalar", initial_value=1.0, constraints=ParameterConstraints(minimum=0.1, maximum=8), recompute="geometry-transform", persistence="lesson"), | |
| ] | |
| bindings = {"xSemiAxis": "a", "ySemiAxis": "b", "zSemiAxis": "c"} | |
| panels = [ThreeScenePanelSpec(panel_id="shape", geometry=RegisteredGeometrySpec(operator=operator, parameter_bindings=bindings), bindings=[*bindings.values(), "opacity", "wireframe", "show_axes", "spin"], order=0)] | |
| properties = [MathematicalProperty(name="operator", value=operator, status="registered", method="App-owned geometry registry")] | |
| panel_ids = [panel.panel_id for panel in panels] | |
| interactions = [InteractionSpec(interaction_id=f"change-{item.parameter_id}", kind="parameter", state_id=item.parameter_id, target_panel_ids=panel_ids) for item in parameters] | |
| return CompositionSpec( | |
| project_id=project_id, prompt=prompt, title=title, | |
| answer_markdown=f"I interpreted your request as an interactive **{title.lower()}**. Its controls update every bound view from the same verified parameter state.", | |
| parameters=parameters, derived_values=derived, panels=panels, interactions=interactions, | |
| assumptions=assumptions, provenance=ProvenanceSpec(interpretation=f"Registered analytic operator: {operator}", operator_sources=[operator]), | |
| assertions=[CompositionAssertion(assertion_id="finite-derived", kind="finite", target_id="derived")], | |
| mathematical_properties=properties, created_at=time.time(), | |
| ) | |
| def _implicit(self, project_id: str, prompt: str) -> CompositionSpec: | |
| match = re.search(r"([^\n.;]+=[^\n.;]+)", prompt) | |
| equation = (match.group(1) if match else prompt).strip() | |
| equation = re.sub(r"^(?:visuali[sz]e|plot|show)\s+", "", equation, flags=re.I) | |
| program = self.parser.parse_equation(equation) | |
| parameters = [ | |
| ParameterSpec(parameter_id="iso_value", label="Iso-value", value_type="scalar", initial_value=0.0, constraints=ParameterConstraints(minimum=-100, maximum=100), recompute="remesh", persistence="lesson"), | |
| ParameterSpec(parameter_id="high_quality", label="96³ quality", value_type="boolean", initial_value=False, recompute="field-resample", persistence="session"), | |
| *self._common_parameters(prompt), | |
| ] | |
| geometry = ImplicitGeometrySpec( | |
| expression=program, bounds=Bounds3D(x=(-2, 2), y=(-2, 2), z=(-2, 2)), resolution=Resolution3D(), | |
| presentation=GeometryPresentation(show_bounding_box=True), | |
| ) | |
| panel = ThreeScenePanelSpec(panel_id="implicit-surface", geometry=geometry, bindings=[item.parameter_id for item in parameters], order=0) | |
| return CompositionSpec( | |
| project_id=project_id, prompt=prompt, title="Implicit surface", answer_markdown=f"I rendered the zero level set of `{equation}` inside the finite domain x, y, z ∈ [-2, 2]. The domain and sampling resolution affect the visible approximation.", | |
| parameters=parameters, panels=[panel], interactions=[InteractionSpec(interaction_id=f"change-{item.parameter_id}", kind="parameter", state_id=item.parameter_id, target_panel_ids=[panel.panel_id]) for item in parameters], | |
| assumptions=["The finite plotting domain is x, y, z ∈ [-2, 2].", "The surface is a marching-cubes approximation, not exact symbolic geometry."], | |
| provenance=ProvenanceSpec(interpretation="User-supplied implicit equation", research_used=False), | |
| assertions=[CompositionAssertion(assertion_id="mesh-budget", kind="mesh_budget", target_id="implicit-surface")], | |
| mathematical_properties=[MathematicalProperty(name="topology", value=None, status="unknown", caveat="Topology is not inferred from the intended equation or a finite-resolution mesh.")], created_at=time.time(), | |
| ) | |
| def _parametric(self, project_id: str, prompt: str) -> CompositionSpec: | |
| expressions: dict[str, str] = {} | |
| for name in ("x", "y", "z"): | |
| match = re.search(rf"\b{name}\s*=\s*([^;\n]+)", prompt, flags=re.I) | |
| if not match: | |
| raise CompositionCompilationError("Parametric surfaces require explicit x(u,v), y(u,v), and z(u,v) expressions separated by semicolons") | |
| expressions[name] = match.group(1).strip() | |
| geometry = ParametricGeometrySpec( | |
| x_program=self.parser.parse_equation(expressions["x"]), | |
| y_program=self.parser.parse_equation(expressions["y"]), | |
| z_program=self.parser.parse_equation(expressions["z"]), | |
| u_domain=(0, 2 * math.pi), v_domain=(0, math.pi), u_segments=80, v_segments=40, | |
| ) | |
| parameters = self._common_parameters(prompt) | |
| panel = ThreeScenePanelSpec(panel_id="parametric-surface", geometry=geometry, bindings=[item.parameter_id for item in parameters]) | |
| return CompositionSpec( | |
| project_id=project_id, prompt=prompt, title="Parametric surface", | |
| answer_markdown="I rendered the supplied parametric surface over the disclosed finite u/v domain. The mesh resolution controls its numerical approximation.", | |
| parameters=parameters, panels=[panel], interactions=[InteractionSpec(interaction_id=f"change-{item.parameter_id}", kind="parameter", state_id=item.parameter_id, target_panel_ids=[panel.panel_id]) for item in parameters], | |
| assumptions=["The parameter domain is u ∈ [0, 2π], v ∈ [0, π] unless a registered operator provides another domain."], | |
| provenance=ProvenanceSpec(interpretation="User-supplied parametric equations"), | |
| mathematical_properties=[MathematicalProperty(name="topology", value=None, status="unknown", caveat="No topological property is inferred from sampled parametric geometry.")], created_at=time.time(), | |
| ) | |
| def _is_number(text: str) -> bool: | |
| try: | |
| float(text) | |
| return True | |
| except ValueError: | |
| return False | |
| def _chart(self, project_id: str, prompt: str, family_override: str = "", extracted: ExtractedChartDataset | None = None) -> CompositionSpec: | |
| lowered = prompt.lower() | |
| keyword_family = "force-graph" if "force" in lowered else "heatmap" if "heatmap" in lowered else "scatter" if "scatter" in lowered else "bar" if "bar" in lowered else "line" | |
| has_grounded_data = extracted is not None and extracted.has_data and bool(extracted.rows) | |
| if family_override: | |
| family = family_override | |
| elif keyword_family == "force-graph": | |
| # An explicit network-graph request never fits label/value rows -> keep the deterministic path. | |
| family = "force-graph" | |
| elif has_grounded_data: | |
| # Trust the data-informed choice over a keyword guess that just defaults to "line". | |
| family = extracted.chart_family | |
| else: | |
| family = keyword_family | |
| is_parabola = any(term in lowered for term in ("parabola", "quadratic curve", "quadratic function")) | |
| grounded_rows = extracted.rows if (has_grounded_data and family != "force-graph") else [] | |
| x_label, y_label = "x", "y = x²" if is_parabola else "y" | |
| if grounded_rows: | |
| illustrative = False | |
| x_is_numeric = all(self._is_number(row.label) for row in grounded_rows) | |
| rows = [{"x": float(row.label) if x_is_numeric else row.label, "y": row.value} for row in grounded_rows] | |
| columns = ["x", "y"] | |
| encodings = [ | |
| ChartEncoding(channel="x", field="x", value_type="quantitative" if x_is_numeric else "nominal"), | |
| ChartEncoding(channel="y", field="y"), | |
| ] | |
| x_label = extracted.x_label or x_label | |
| y_label = extracted.y_label or y_label | |
| else: | |
| pairs = [(float(a), float(b)) for a, b in re.findall(r"\(?\s*(-?\d+(?:\.\d+)?)\s*[, :]\s*(-?\d+(?:\.\d+)?)\s*\)?", prompt)][:200] | |
| illustrative = not pairs | |
| pairs = pairs or ([(float(x), float(x * x)) for x in range(-5, 6)] if is_parabola else [(float(x), float(x * x)) for x in range(6)]) | |
| if family == "force-graph": | |
| named_edges = re.findall(r"\b([A-Za-z][\w-]*)\s*(?:->|--|—|–)\s*([A-Za-z][\w-]*)\b", prompt) | |
| rows = ([{"source": source, "target": target, "weight": 1.0} for source, target in named_edges] | |
| if named_edges else [{"source": f"N{int(x)}", "target": f"N{int(y)}", "weight": 1.0} for x, y in pairs]) | |
| illustrative = not named_edges and illustrative | |
| columns = ["source", "target", "weight"] | |
| encodings = [ChartEncoding(channel="source", field="source", value_type="nominal"), ChartEncoding(channel="target", field="target", value_type="nominal"), ChartEncoding(channel="weight", field="weight")] | |
| else: | |
| rows = [{"x": x, "y": y} for x, y in pairs] | |
| columns = ["x", "y"] | |
| encodings = [ChartEncoding(channel="x", field="x"), ChartEncoding(channel="y", field="y")] | |
| dataset = DatasetSpec(dataset_id="data", columns=columns, rows=rows) | |
| chart = ChartSpec(chart_id="chart", family=family, dataset_id="data", encodings=encodings, x_label=x_label, y_label=y_label, interactions=["zoom", "hover", "select-datum", "linked-selection"]) | |
| renderer = chart_renderer_for(chart) | |
| if grounded_rows: | |
| answer_markdown = f"I interpreted this as a **{family.replace('-', ' ')}** rendered through the verified {renderer.title()} adapter, using the values transcribed from your selected context." | |
| elif is_parabola: | |
| answer_markdown = "I rendered the standard parabola **y = x²** as an interactive 2D plot." | |
| else: | |
| answer_markdown = f"I interpreted this as a **{family.replace('-', ' ')}** and rendered it through the verified {renderer.title()} adapter." + (" The displayed values are illustrative because no explicit dataset was supplied." if illustrative else "") | |
| return CompositionSpec( | |
| project_id=project_id, prompt=prompt, title="Parabola" if is_parabola and not grounded_rows else f"{family.replace('-', ' ').title()}", | |
| answer_markdown=answer_markdown, | |
| parameters=[], datasets=[dataset], panels=[ChartPanelSpec(panel_id="chart", chart=chart, order=0)], | |
| assumptions=["The chart dataset is illustrative." ] if illustrative else [], | |
| provenance=ProvenanceSpec( | |
| interpretation=f"Semantic chart family: {family}" + (" (grounded in selected context)" if grounded_rows else ""), | |
| operator_sources=[renderer], | |
| ), | |
| assertions=[CompositionAssertion(assertion_id="dataset-shape", kind="dataset_shape", target_id="data")], created_at=time.time(), | |
| ) | |