"""``calculus-series@v1`` π* — Taylor / Maclaurin series canonicalizer.

Domain: ``calculus``. Computes the truncated Taylor series of
``f`` around ``x = x0`` to ``n`` terms via ``sp.series``. The
``O(x**n)`` remainder is dropped — that's what makes the series
finite-byte canonical.

Why this matters
----------------
Ticket #000030 §2.2 Phase 5. Series complete the calculus-primitive
quartet (derivative · integral · limit · series) on the SymPy
substrate. All four ride ``algebra-symbolic@v1``'s
``sp.expand + sp.srepr`` normalizer for the closed-form output, so
the four kernels compose cleanly.

Input shape (JSON, UTF-8)
-------------------------
::

    {"f": "<expression-text>",
     "x": "<variable-name>",
     "x0": "<sympify-able-point>",
     "n": <positive-int>,
     "timeout_seconds"?: <float>}

- ``f`` — the function to expand.
- ``x`` — the variable.
- ``x0`` — the expansion point. ``"0"`` gives a Maclaurin series.
- ``n`` — number of terms (the ``n`` in SymPy's ``series(..., n=n)``).
  Must be a positive integer.
- ``timeout_seconds`` — optional, default :data:`DEFAULT_TIMEOUT`.

Output canonical bytes
----------------------
Two paths:

1. **Series with closed-form coefficients** — the truncated
   polynomial after ``.removeO()``, fed through
   ``sp.expand + sp.srepr``. Two equivalent series collapse.
2. **Timeout** — bytes are
   ``b"unevaluated:" + sp.srepr(<original f, x0, n>)``. Mirrors
   ``calculus-integral@v1`` / ``calculus-limit@v1`` sentinel
   discipline.

Optional dependency
-------------------
Same as ``algebra-symbolic@v1``: SymPy via the ``[math]`` extras.

Source: ticket #000030 §2.2 Phase 5 — landed 2026-05-09.
"""

from __future__ import annotations

import concurrent.futures
import json
from dataclasses import dataclass

from arborist.pi_star.protocol import PiStarError
from arborist.pi_star.registry import register


try:
    import sympy as sp  # type: ignore[import-not-found]
except ImportError:  # pragma: no cover — optional extra
    sp = None  # type: ignore[assignment]


DEFAULT_TIMEOUT = 30.0


_SYMPY_REQUIRED_MSG = (
    "calculus-series@v1 requires sympy; install with "
    "`pip install 'arborist[math]'`"
)


def _sentinel(f_expr, x_sym, x0_expr, n: int) -> bytes:
    # Build a stable sentinel from inputs (we don't have a SymPy
    # "unevaluated series" class). Hash-able by srepr of the tuple.
    payload = f"Series({sp.srepr(f_expr)}, {sp.srepr(x_sym)}, {sp.srepr(x0_expr)}, {n})"
    return b"unevaluated:" + payload.encode("utf-8")


@dataclass
class CalculusSeriesV1:
    name: str = "calculus-series"
    version: str = "v1"
    domain: str = "calculus"

    def canonicalize(self, raw: bytes) -> bytes:
        if sp is None:
            raise PiStarError(_SYMPY_REQUIRED_MSG)
        if not isinstance(raw, (bytes, bytearray)):
            raise PiStarError(
                "calculus-series@v1 expects bytes; got "
                f"{type(raw).__name__}"
            )
        try:
            text = raw.decode("utf-8", errors="surrogatepass")
        except UnicodeDecodeError as exc:  # pragma: no cover
            raise PiStarError(f"input not valid UTF-8: {exc}") from exc

        try:
            obj = json.loads(text)
        except json.JSONDecodeError as exc:
            raise PiStarError(
                f"calculus-series@v1 input is not valid JSON: {exc}"
            ) from exc
        if not isinstance(obj, dict):
            raise PiStarError(
                "calculus-series@v1 input must be a JSON object"
            )
        for required in ("f", "x", "x0", "n"):
            if required not in obj:
                raise PiStarError(
                    f"calculus-series@v1 missing required field "
                    f"{required!r}"
                )

        f_text = obj["f"]
        x_name = obj["x"]
        x0_text = obj["x0"]
        n_terms = obj["n"]
        timeout_s = float(obj.get("timeout_seconds", DEFAULT_TIMEOUT))

        if not isinstance(f_text, str) or not f_text.strip():
            raise PiStarError("calculus-series@v1 f must be a non-empty string")
        if not isinstance(x_name, str) or not x_name.strip():
            raise PiStarError("calculus-series@v1 x must be a non-empty string")
        # JSON booleans are subclasses of int in Python; reject explicitly.
        if isinstance(n_terms, bool) or not isinstance(n_terms, int) or n_terms < 1:
            raise PiStarError(
                f"calculus-series@v1 n must be a positive integer; got "
                f"{n_terms!r}"
            )

        try:
            x_sym = sp.Symbol(x_name)
            f_expr = sp.sympify(f_text, locals={x_name: x_sym})
            x0_expr = sp.sympify(x0_text)
        except (sp.SympifyError, SyntaxError, TypeError) as exc:
            raise PiStarError(
                f"calculus-series@v1 cannot parse: {exc}"
            ) from exc

        def _compute():
            return sp.series(f_expr, x_sym, x0_expr, n_terms).removeO()

        try:
            with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
                future = ex.submit(_compute)
                try:
                    truncated = future.result(timeout=timeout_s)
                except concurrent.futures.TimeoutError:
                    return _sentinel(f_expr, x_sym, x0_expr, n_terms)
        except (sp.SympifyError, ValueError, TypeError) as exc:
            raise PiStarError(
                f"calculus-series@v1 evaluation failed: {exc}"
            ) from exc

        try:
            expanded = sp.expand(truncated)
        except Exception as exc:  # pragma: no cover
            raise PiStarError(
                f"calculus-series@v1 cannot expand result: {exc}"
            ) from exc
        return sp.srepr(expanded).encode("utf-8")


if sp is not None:
    register(CalculusSeriesV1())
