Spaces:
Sleeping
Sleeping
| 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 | |
| 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 | |
| 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, | |
| ) | |