from __future__ import annotations import math import re from app.schemas.visual_lesson import ( CompiledDifferentialEquation, DifferentialEquationFeature, DifferentialEquationParameters, DifferentialEquationSample, DifferentialEquationSpec, DirectionFieldPoint, ) class DifferentialEquationCompilationError(ValueError): pass class DifferentialEquationCompiler: SAMPLE_COUNT = 401 @staticmethod def _number(text: str, patterns: tuple[str, ...]) -> float | None: for pattern in patterns: match = re.search(pattern, text, flags=re.IGNORECASE) if match: return float(match.group(1).replace(" ", "")) return None def parse_prompt(self, prompt: str) -> tuple[DifferentialEquationSpec, list[str]]: text = re.sub(r"\s+", " ", prompt.strip()).lower().replace("−", "-") warnings: list[str] = [] t_end = self._number(text, (r"\bt\s*(?:=|from)?\s*0\s*(?:\.\.|to)\s*(-?\d+(?:\.\d+)?)", r"\buntil\s+t\s*=\s*(\d+(?:\.\d+)?)")) if any(term in text for term in ("logistic", "carrying capacity", "population saturation")): family = "logistic" r = self._number(text, (r"\br\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"growth rate\s*(?:of|=|:)\s*(-?\d+(?:\.\d+)?)")) capacity = self._number(text, (r"\bk\s*[:=]\s*(\d+(?:\.\d+)?)", r"carrying capacity\s*(?:of|=|:)\s*(\d+(?:\.\d+)?)")) initial = self._number(text, (r"\by\s*\(\s*0\s*\)\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"\by0\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"initial (?:value|population)\s*(?:of|=|:)\s*(-?\d+(?:\.\d+)?)")) r = 1.0 if r is None else r capacity = 100.0 if capacity is None else capacity initial = 5.0 if initial is None else initial if not 0 < r <= 20: raise DifferentialEquationCompilationError("Logistic growth rate r must be greater than 0 and at most 20") if not 0 < capacity <= 1e9: raise DifferentialEquationCompilationError("Carrying capacity K must be greater than 0") if not 0 < initial <= capacity * 10: raise DifferentialEquationCompilationError("The initial value y(0) must be positive and no more than 10K") t_end = t_end if t_end is not None else max(5.0, min(20.0, 6.0 / r)) parameters = DifferentialEquationParameters(growth_rate=r, carrying_capacity=capacity, initial_value=initial) equation = r"\frac{dy}{dt}=r y\left(1-\frac{y}{K}\right)" interpretation = f"Logistic growth with r={r:g}, K={capacity:g}, and y(0)={initial:g}." views = ["trajectory", "direction_field"] title = "Logistic growth" elif any(term in text for term in ("oscillator", "damped", "spring", "phase portrait", "pendulum")): family = "damped_oscillator" is_pendulum = "pendulum" in text omega = self._number(text, (r"(?:omega|ω|natural frequency|w0)\s*(?:_?0)?\s*[:=]\s*(\d+(?:\.\d+)?)",)) if omega is None and is_pendulum: length = self._number(text, (r"length\s*(?:of|=|:)?\s*(\d+(?:\.\d+)?)\s*m", r"\bl\s*[:=]\s*(\d+(?:\.\d+)?)")) gravity = self._number(text, (r"gravity\s*(?:of|=|:)?\s*(\d+(?:\.\d+)?)", r"\bg\s*[:=]\s*(\d+(?:\.\d+)?)")) length = 1.0 if length is None else length gravity = 9.8 if gravity is None else gravity if not 0.01 <= length <= 100: raise DifferentialEquationCompilationError("Pendulum length must be between 0.01 and 100 m") if not 0.1 <= gravity <= 100: raise DifferentialEquationCompilationError("Gravity must be between 0.1 and 100 m/s^2") omega = math.sqrt(gravity / length) zeta = self._number(text, (r"(?:zeta|ζ|damping ratio)\s*[:=]\s*(\d+(?:\.\d+)?)",)) initial = self._number(text, (r"\bx\s*\(\s*0\s*\)\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"\bx0\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"initial angle\s*(?:of|=|:)?\s*(-?\d+(?:\.\d+)?)")) velocity = self._number(text, (r"\bv\s*\(\s*0\s*\)\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"\bv0\s*[:=]\s*(-?\d+(?:\.\d+)?)")) omega = 1.0 if omega is None else omega zeta = (0.05 if is_pendulum else 0.15) if zeta is None else zeta initial = (0.35 if is_pendulum else 1.0) if initial is None else initial velocity = 0.0 if velocity is None else velocity if not 0.05 <= omega <= 100 or not 0 <= zeta <= 5: raise DifferentialEquationCompilationError("Oscillator frequency must be 0.05–100 and damping ratio must be 0–5") if abs(initial) + abs(velocity) == 0: raise DifferentialEquationCompilationError("At least one oscillator initial condition must be nonzero") t_end = t_end if t_end is not None else min(30.0, max(8.0, 6.0 * math.pi / omega)) parameters = DifferentialEquationParameters(initial_value=initial, initial_velocity=velocity, natural_frequency=omega, damping_ratio=zeta) equation = r"\theta''+2\zeta\omega_0 \theta'+\omega_0^2\theta=0" if is_pendulum else r"x''+2\zeta\omega_0 x'+\omega_0^2x=0" if is_pendulum: interpretation = f"Simple gravity pendulum (small-angle approximation) with ω₀={omega:g} rad/s, ζ={zeta:g}, θ(0)={initial:g} rad, and θ'(0)={velocity:g} rad/s." title = "Simple pendulum" else: interpretation = f"Damped harmonic motion with ω₀={omega:g}, ζ={zeta:g}, x(0)={initial:g}, and v(0)={velocity:g}." title = "Damped oscillator" views = ["trajectory", "phase_portrait"] else: family = "linear_first_order" coefficient = self._number(text, (r"\ba\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"y\s*['’]\s*=\s*(-?\d+(?:\.\d+)?)\s*y")) forcing = self._number(text, (r"\bb\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"y\s*['’]\s*=\s*-?\d+(?:\.\d+)?\s*y\s*([+-]\s*\d+(?:\.\d+)?)")) initial = self._number(text, (r"\by\s*\(\s*0\s*\)\s*[:=]\s*(-?\d+(?:\.\d+)?)", r"\by0\s*[:=]\s*(-?\d+(?:\.\d+)?)")) coefficient = 1.0 if coefficient is None else coefficient forcing = 0.0 if forcing is None else forcing initial = 1.0 if initial is None else initial if abs(coefficient) > 20 or abs(forcing) > 1e9 or abs(initial) > 1e9: raise DifferentialEquationCompilationError("Linear-equation parameters exceed the supported range") t_end = t_end if t_end is not None else (5.0 if coefficient > 0 else 10.0) parameters = DifferentialEquationParameters(linear_coefficient=coefficient, forcing=forcing, initial_value=initial) equation = r"\frac{dy}{dt}=ay+b" interpretation = f"Linear first-order growth with a={coefficient:g}, b={forcing:g}, and y(0)={initial:g}." views = ["trajectory", "direction_field"] title = "Linear first-order equation" if t_end is None or not 0.1 <= t_end <= 100: raise DifferentialEquationCompilationError("The displayed time interval must end between 0.1 and 100") spec = DifferentialEquationSpec( project_id="", prompt=prompt, title=title, family=family, interpretation=interpretation, equation_latex=equation, parameters=parameters, t_end=t_end, sample_count=self.SAMPLE_COUNT, enabled_views=views, assumptions=[ "Parameters are constant over the displayed time interval.", "The model is continuous and deterministic; no measurement noise or external perturbation is included.", ], created_at=0.0, ) return spec, warnings @staticmethod def _rhs(spec: DifferentialEquationSpec, primary: float, secondary: float = 0.0) -> tuple[float, float]: p = spec.parameters if spec.family == "logistic": return p.growth_rate * primary * (1.0 - primary / p.carrying_capacity), 0.0 if spec.family == "linear_first_order": return p.linear_coefficient * primary + p.forcing, 0.0 return secondary, -2.0 * p.damping_ratio * p.natural_frequency * secondary - p.natural_frequency ** 2 * primary def _samples(self, spec: DifferentialEquationSpec) -> list[DifferentialEquationSample]: count = spec.sample_count dt = (spec.t_end - spec.t_start) / (count - 1) p = spec.parameters samples: list[DifferentialEquationSample] = [] if spec.family == "logistic": ratio = (p.carrying_capacity - p.initial_value) / p.initial_value for index in range(count): t = spec.t_start + index * dt value = p.carrying_capacity / (1.0 + ratio * math.exp(-p.growth_rate * t)) derivative, _ = self._rhs(spec, value) samples.append(DifferentialEquationSample(sample_index=index, t=t, primary=value, derivative=derivative)) return samples if spec.family == "linear_first_order": for index in range(count): t = spec.t_start + index * dt if abs(p.linear_coefficient) < 1e-12: value = p.initial_value + p.forcing * t else: equilibrium = -p.forcing / p.linear_coefficient value = equilibrium + (p.initial_value - equilibrium) * math.exp(p.linear_coefficient * t) derivative, _ = self._rhs(spec, value) samples.append(DifferentialEquationSample(sample_index=index, t=t, primary=value, derivative=derivative)) return samples x, velocity = p.initial_value, p.initial_velocity for index in range(count): t = spec.t_start + index * dt dx, _ = self._rhs(spec, x, velocity) samples.append(DifferentialEquationSample(sample_index=index, t=t, primary=x, derivative=dx, secondary=velocity)) if index == count - 1: break k1x, k1v = self._rhs(spec, x, velocity) k2x, k2v = self._rhs(spec, x + dt * k1x / 2, velocity + dt * k1v / 2) k3x, k3v = self._rhs(spec, x + dt * k2x / 2, velocity + dt * k2v / 2) k4x, k4v = self._rhs(spec, x + dt * k3x, velocity + dt * k3v) x += dt * (k1x + 2 * k2x + 2 * k3x + k4x) / 6 velocity += dt * (k1v + 2 * k2v + 2 * k3v + k4v) / 6 return samples def _direction_field(self, spec: DifferentialEquationSpec, samples: list[DifferentialEquationSample]) -> list[DirectionFieldPoint]: if spec.family == "damped_oscillator": return [] values = [sample.primary for sample in samples] low, high = min(values), max(values) padding = max((high - low) * 0.18, max(abs(low), abs(high), 1.0) * 0.08) if spec.family == "logistic": low = min(0.0, low - padding) high = max(spec.parameters.carrying_capacity * 1.08, high + padding) else: low -= padding high += padding points: list[DirectionFieldPoint] = [] for ti in range(13): t = spec.t_start + (spec.t_end - spec.t_start) * ti / 12 for yi in range(11): state = low + (high - low) * yi / 10 slope, _ = self._rhs(spec, state) points.append(DirectionFieldPoint(t=t, state=state, slope=slope)) return points def _features(self, spec: DifferentialEquationSpec) -> list[DifferentialEquationFeature]: p = spec.parameters features = [DifferentialEquationFeature(feature_id="initial", label="Initial state", value=p.initial_value, kind="initial_state")] if spec.family == "logistic": features.extend([ DifferentialEquationFeature(feature_id="equilibrium-zero", label="Unstable equilibrium", value=0.0, kind="equilibrium"), DifferentialEquationFeature(feature_id="carrying-capacity", label="Carrying capacity K", value=p.carrying_capacity, kind="equilibrium"), ]) ratio = (p.carrying_capacity - p.initial_value) / p.initial_value if ratio > 0: t_inflection = math.log(ratio) / p.growth_rate if spec.t_start <= t_inflection <= spec.t_end: features.append(DifferentialEquationFeature(feature_id="inflection", label="Maximum growth rate", t=t_inflection, value=p.carrying_capacity / 2, kind="inflection")) elif spec.family == "linear_first_order" and abs(p.linear_coefficient) > 1e-12: features.append(DifferentialEquationFeature(feature_id="equilibrium", label="Equilibrium", value=-p.forcing / p.linear_coefficient, kind="equilibrium")) return features def compile_spec(self, spec: DifferentialEquationSpec) -> CompiledDifferentialEquation: samples = self._samples(spec) if len(samples) != spec.sample_count or not all(math.isfinite(value) for sample in samples for value in (sample.t, sample.primary, sample.derivative, sample.secondary)): raise DifferentialEquationCompilationError("The differential-equation solution produced invalid values") if abs(samples[0].primary - spec.parameters.initial_value) > 1e-9: raise DifferentialEquationCompilationError("The compiled solution does not satisfy the initial value") if spec.family == "damped_oscillator" and abs(samples[0].secondary - spec.parameters.initial_velocity) > 1e-9: raise DifferentialEquationCompilationError("The compiled solution does not satisfy the initial velocity") for sample in samples: expected, _ = self._rhs(spec, sample.primary, sample.secondary) if abs(sample.derivative - expected) > 1e-8 * max(abs(expected), 1.0): raise DifferentialEquationCompilationError("The derivative assertion failed") return CompiledDifferentialEquation( samples=samples, direction_field=self._direction_field(spec, samples), features=self._features(spec), assertions_passed=True, )