Blame for app/gapmap_service.py

"""Serves the gap map: cluster the studies around a question, describe them, place the question.

A map with a question is built from that question's neighborhood: the
``gapmap_neighborhood`` studies nearest its vector, fetched by kNN, clustered
and drawn. That is a few hundred documents, so the build takes about a second
and the picture is one a reader can take in. Its coverage says exactly that —
the N nearest studies out of the embedded index — and never more.

Without a question there is nothing to center on, so the map falls back to the
whole embedded index. Covering the whole index means the build is long enough
to be worth watching, so it is written as a stream: documents are scanned in
pages, provisional centroids move as pages arrive, and full-corpus k-means then
runs to convergence, emitting the real intermediate state after every iteration.
Nothing emitted is interpolated or replayed — a partial map is labelled partial,
and only the final event describes the whole corpus. Measured on the live index
(2,096,736 embedded studies): 264 s end to end — roughly 130 s to scan at sixteen
slices and 25 bounded k-means iterations over the whole corpus in the rest,
peaking at 6.9 GB.
"""

from __future__ import annotations

import asyncio
import contextlib
import logging
import zlib
from collections.abc import Awaitable, Callable

import numpy as np

from app.config import Settings, settings
from app.embeddings import QueryEmbedder
from app.gapmap import (
    MAP_VERSION,
    RunningClusters,
    _bucket,
    assign,
    calibrate,
    combine,
    describe_regions,
    find_gaps,
    fit_projection,
    kmeans_steps,
    neighbor_edges,
    normalize,
    place,
    project_with,
)
from app.models import MapArithmetic

logger = logging.getLogger(__name__)

Progress = Callable[[dict], Awaitable[None]]

# Question maps kept in memory beside the corpus map.
MAP_CACHE = 16


def _public(region: dict) -> dict:
    """Region without its centroid: a 384-float vector is not UI payload."""
    return {key: value for key, value in region.items() if key != "centroid"}


def _drawn(identifier: str, stride: int) -> bool:
    """Whether a document is one of the ones drawn on the canvas.

    A browser cannot render two million marks, so the canvas shows a subset —
    chosen by a hash of the document id, which is deterministic, independent of
    the order pages happen to arrive in, and unrelated to anything the map
    labels. Every document is still clustered; only the drawing is thinned.
    """
    return stride <= 1 or zlib.crc32(identifier.encode("utf-8")) % stride == 0


class MapBuild:
    """The corpus as it accumulates: vectors in one matrix, documents beside it.

    The matrix is grown by doubling and handed out as a view, so clustering a
    corpus of millions never holds two copies of it and adding a page costs the
    page, not the corpus. The embedding is dropped from each document once it
    has been copied into the matrix: keeping both is three gigabytes of Python
    floats for nothing.
    """

    def __init__(self, corpus: int, capacity: int = 0):
        self.corpus = corpus
        self.documents: list[dict] = []
        self.dimensions: int | None = None
        self._capacity = max(1024, capacity)
        self._buffer: np.ndarray | None = None
        self._rows = 0

    def _reserve(self, rows: int) -> None:
        assert self.dimensions is not None
        if self._buffer is None:
            self._buffer = np.empty((self._capacity, self.dimensions), dtype=np.float32)
        while self._rows + rows > len(self._buffer):
            grown = np.empty((len(self._buffer) * 2, self.dimensions), dtype=np.float32)
            grown[: self._rows] = self._buffer[: self._rows]
            self._buffer = grown

    def plan(self, rows: int) -> None:
        """Grow to the size the scan expects, in one step rather than by doubling."""
        self._capacity = max(self._capacity, rows)
        if self.dimensions is None or rows <= 0:
            return
        if self._buffer is None or len(self._buffer) < rows:
            grown = np.empty((rows, self.dimensions), dtype=np.float32)
            if self._buffer is not None:
                grown[: self._rows] = self._buffer[: self._rows]
            self._buffer = grown

    def add(self, documents: list[dict]) -> int:
        """Append a page's vectors to the matrix; returns how many were usable."""
        rows, kept = [], []
        for document in documents:
            embedding = document.pop("embedding", None)
            if not isinstance(embedding, (list, tuple)) or not embedding:
                continue
            vector = np.asarray(embedding, dtype=np.float32)
            if not np.isfinite(vector).all() or not vector.any():
                continue
            if self.dimensions is None:
                self.dimensions = len(vector)
            if len(vector) != self.dimensions:
                continue
            rows.append(vector)
            kept.append(document)
        if not rows:
            return 0
        self._reserve(len(rows))
        assert self._buffer is not None
        self._buffer[self._rows : self._rows + len(rows)] = normalize(np.vstack(rows))
        self._rows += len(rows)
        self.documents.extend(kept)
        return len(kept)

    @property
    def vectors(self) -> np.ndarray:
        if self._buffer is None:
            return np.zeros((0, self.dimensions or 1), dtype=np.float32)
        return self._buffer[: self._rows]

    def __len__(self) -> int:
        return self._rows


class GapMapService:
    def __init__(self, repository, config: Settings = settings):
        self.repo = repository
        self.config = config
        self.embedder = QueryEmbedder(config)
        # One cached map per question (keyed by its vector) plus the corpus map.
        self._maps: dict[bytes | None, dict] = {}
        self._lock = asyncio.Lock()
        self._subscribers: set[asyncio.Queue] = set()
        self._tasks: dict[bytes | None, asyncio.Task] = {}
        # The plane the current build is drawing on, so a streamed question can be
        # placed on the same one before the build finishes.
        self._basis: tuple[np.ndarray, np.ndarray] | None = None

    async def embed(self, text: str) -> list[float] | None:
        return await self.embedder.embed(text)

    # ---------------------------------------------------------------- building

    async def _publish(self, payload: dict) -> None:
        for queue in list(self._subscribers):
            if queue.full():  # a slow reader must not stall the build
                continue
            queue.put_nowait(payload)

    @property
    def _built(self) -> dict | None:
        """The cached corpus map, if it has been built."""
        return self._maps.get(None)

    async def build(self, refresh: bool = False, vector: np.ndarray | None = None) -> dict:
        """The cached map for this question (or the corpus), building it once if needed.

        The build is a task of its own rather than a coroutine of whichever
        request arrived first: a viewer who closes the tab after twenty seconds
        must not throw away a clustering run that other viewers, and the cache,
        are waiting for.
        """
        key = None if vector is None else np.asarray(vector, dtype=np.float32).tobytes()
        if key in self._maps and not refresh:
            return self._maps[key]
        async with self._lock:
            if key in self._maps and not refresh:
                return self._maps[key]
            task = self._tasks.get(key)
            if task is None or task.done():
                task = asyncio.create_task(self._build(vector))
                self._tasks[key] = task
        built = await asyncio.shield(task)
        if len(self._maps) >= MAP_CACHE and key not in self._maps:
            for stale in [k for k in self._maps if k is not None][: len(self._maps) - MAP_CACHE + 1]:
                del self._maps[stale]
        self._maps[key] = built
        return built

    async def _build(self, vector: np.ndarray | None) -> dict:
        if vector is None:
            return await self._build_corpus()
        return await self._build_neighborhood(vector)

    async def _build_neighborhood(self, vector: np.ndarray) -> dict:
        """Fetch the studies nearest the question, cluster them, describe the result."""
        counting = asyncio.create_task(self.repo.count_embedded())
        size = self.config.gapmap_neighborhood
        query = normalize(np.asarray(vector, dtype=np.float32)[None, :])[0]
        documents = await self.repo.knn_studies(query.tolist(), size)
        build = MapBuild(0, capacity=size)
        build.add(documents)
        stride = max(1, len(build) // max(1, self.config.gapmap_points))
        drawn = [
            row
            for row in range(len(build))
            if _drawn(str(build.documents[row].get("id", row)), stride)
        ]
        basis: tuple[np.ndarray, np.ndarray] | None = None
        if len(build):
            basis = await asyncio.to_thread(fit_projection, build.vectors)
            self._basis = basis
        corpus = max(0, await counting)
        build.corpus = max(corpus, len(build))
        scope = {"scope": "neighborhood", "neighborhood": size}
        centroids = None
        labels = np.zeros(len(build), dtype=np.int64)
        iteration = 0
        emitted = 0
        if len(build):
            steps = kmeans_steps(
                build.vectors,
                self.config.gapmap_regions,
                seed=self.config.gapmap_seed,
                iterations=self.config.gapmap_iterations,
            )
            while True:
                step = await asyncio.to_thread(next, steps, None)
                if step is None:
                    break
                centroids, labels = step
                iteration += 1
                if basis is not None:
                    emitted = await self._emit_partial(
                        build,
                        centroids,
                        basis,
                        drawn,
                        emitted,
                        len(build),
                        iteration=iteration,
                        labels=labels,
                        extra={"coverage": scope},
                    )
        return await self._finish(build, centroids, labels, basis, drawn, iteration, scope)

    async def _build_corpus(self) -> dict:
        """Scan the embedded corpus, cluster it, and describe the result."""
        # Counted beside the scan rather than before it: the count is a query of
        # its own over two million documents, and nothing can be drawn while it
        # is the only thing running.
        counting = asyncio.create_task(self.repo.count_embedded())
        limit = max(0, self.config.gapmap_scan_limit)
        # Until the corpus size is known the scan cannot thin the drawn subset to
        # fit the canvas, so the first pages are drawn whole up to half the point
        # budget; the stride is set properly as soon as the count lands.
        stride = max(1, limit // max(1, self.config.gapmap_points)) if limit else 1
        early = self.config.gapmap_points // 2
        build = MapBuild(0, capacity=limit)
        target = limit
        clusters = RunningClusters(self.config.gapmap_regions, self.config.gapmap_seed)
        drawn: list[int] = []  # row indices of the documents the canvas shows
        basis: tuple[np.ndarray, np.ndarray] | None = None
        emitted = 0
        ticked = 0.0
        loop = asyncio.get_running_loop()

        async for batch in self.repo.scan_embedded(
            batch_size=self.config.gapmap_batch,
            slices=self.config.gapmap_slices,
            limit=limit,
        ):
            start = len(build)
            build.add(batch)
            for row in range(start, len(build)):
                if not build.corpus and len(drawn) >= early:
                    break
                if _drawn(str(build.documents[row].get("id", row)), stride):
                    drawn.append(row)
            if not build.corpus and counting.done():
                corpus = max(0, counting.result())
                build.corpus = corpus
                target = min(corpus, limit) if limit else corpus
                stride = max(1, target // max(1, self.config.gapmap_points))
                build.plan(target)
            await asyncio.to_thread(clusters.update, build.vectors[start:])
            if basis is None and len(build) >= self.config.gapmap_projection_sample:
                basis = await asyncio.to_thread(
                    fit_projection, build.vectors[: self.config.gapmap_projection_sample]
                )
                self._basis = basis
            # One frame per interval, not one per page: at eleven thousand
            # documents a second the pages are far faster than anything a canvas
            # can show, and every frame resends where each drawn point now sits.
            if (
                basis is not None
                and clusters.centroids is not None
                and loop.time() - ticked >= self.config.gapmap_tick_seconds
            ):
                ticked = loop.time()
                emitted = await self._emit_partial(
                    build, clusters.centroids, basis, drawn, emitted, target
                )

        corpus = max(0, await counting)
        # A live index grows while it is read, so a scan can finish holding more
        # documents than the count taken at its start. The number actually read
        # is the one that was measured; the count is only the older of the two.
        corpus = max(corpus, len(build))
        build.corpus = corpus
        if basis is None and len(build):
            basis = await asyncio.to_thread(
                fit_projection, build.vectors[: self.config.gapmap_projection_sample]
            )
            self._basis = basis

        centroids = clusters.centroids
        labels = np.zeros(len(build), dtype=np.int64)
        iteration = 0
        if len(build):
            # Stepped from a thread one iteration at a time: each step is a pass
            # over the whole corpus, so running the generator inline would block
            # the event loop for exactly as long as the frames it is producing.
            steps = kmeans_steps(
                build.vectors,
                self.config.gapmap_regions,
                seed=self.config.gapmap_seed,
                iterations=self.config.gapmap_iterations,
                centroids=centroids,
            )
            while True:
                step = await asyncio.to_thread(next, steps, None)
                if step is None:
                    break
                centroids, labels = step
                iteration += 1
                if basis is not None:
                    emitted = await self._emit_partial(
                        build,
                        centroids,
                        basis,
                        drawn,
                        emitted,
                        target,
                        iteration=iteration,
                        labels=labels,
                    )

        return await self._finish(
            build, centroids, labels, basis, drawn, iteration, {"scope": "corpus"}
        )

    async def _finish(
        self,
        build: MapBuild,
        centroids: np.ndarray | None,
        labels: np.ndarray,
        basis: tuple[np.ndarray, np.ndarray] | None,
        drawn: list[int],
        iteration: int,
        scope: dict,
    ) -> dict:
        """Describe a clustered build: regions, gaps, drawn points, edges, coverage."""
        corpus = build.corpus
        clusters_used = len(centroids) if centroids is not None else 0
        regions = await asyncio.to_thread(
            describe_regions, build.documents, build.vectors, labels, clusters_used
        )
        gaps = await asyncio.to_thread(
            find_gaps, build.documents, regions, vectors=build.vectors
        )
        points, kept = self._points(build, regions, labels, basis, drawn)
        edges = await asyncio.to_thread(self._edges, build, kept)
        if basis is not None and regions:
            centre, plane = basis
            positions = project_with(
                centre,
                plane,
                np.vstack([region["centroid"] for region in regions]).astype(np.float32),
            )
            for region, (x, y) in zip(regions, positions):
                region["x"], region["y"] = float(x), float(y)
        built = {
            "documents": build.documents,
            "vectors": build.vectors,
            "regions": regions,
            "gaps": gaps,
            "points": points,
            "edges": edges,
            "basis": basis,
            "coverage": {
                "clustered": len(build),
                "corpus": corpus,
                "regions": len(regions),
                "drawn": len(points),
                "complete": True,
                **scope,
            },
            "version": MAP_VERSION,
        }
        logger.info(
            "Gap map (%s) built over %s of %s embedded studies in %s regions after %s iterations",
            scope.get("scope"),
            len(build),
            corpus,
            len(regions),
            iteration,
        )
        await self._publish({"event": "built"})
        return built

    async def _emit_partial(
        self,
        build: MapBuild,
        centroids: np.ndarray,
        basis: tuple[np.ndarray, np.ndarray],
        drawn: list[int],
        emitted: int,
        target: int,
        iteration: int | None = None,
        labels: np.ndarray | None = None,
        publish: Progress | None = None,
        extra: dict | None = None,
    ) -> int:
        """Publish the map as it stands: new points, where the regions are now.

        Only the points the canvas has not seen are sent in full; the ones it
        already holds are recoloured through ``pointRegions``, because a region
        a point belongs to changes on every k-means iteration while its position
        does not. Frames go to ``publish`` when given (one viewer's own build)
        and to every subscriber otherwise (the shared corpus build).
        """
        if publish is None and not self._subscribers:
            return emitted
        centre, plane = basis
        rows = np.array(drawn, dtype=np.int64)
        if not len(rows):
            return emitted
        if labels is None:
            member = assign(build.vectors[rows], centroids)
        else:
            member = labels[rows]
        fresh: list[dict] = []
        edges: list[list[float]] = []
        if len(rows) > emitted:
            new_rows = rows[emitted:]
            coords = project_with(centre, plane, build.vectors[new_rows])
            edges = await asyncio.to_thread(self._edges, build, rows.tolist(), emitted)
            for offset, row in enumerate(new_rows.tolist()):
                document = build.documents[row]
                title = str(document.get("title") or "")
                fresh.append(
                    {
                        "id": str(document.get("id", "")),
                        "x": float(coords[offset, 0]),
                        "y": float(coords[offset, 1]),
                        "region": int(member[emitted + offset]),
                        "bucket": _bucket(document),
                        "title": title if len(title) <= 120 else f"{title[:119]}…",
                        "year": document.get("year") if isinstance(document.get("year"), int) else None,
                    }
                )
        positions = project_with(centre, plane, centroids)
        counts = np.bincount(member, minlength=len(centroids))
        await (publish or self._publish)(
            {
                **(extra or {}),
                "event": "progress",
                "stage": "clustering" if iteration else "scanning",
                "iteration": iteration,
                "coverage": {
                    "clustered": len(build),
                    "corpus": build.corpus,
                    "regions": int((counts > 0).sum()),
                    "drawn": len(rows),
                    "target": target,
                    "complete": False,
                    **(extra or {}).get("coverage", {}),
                },
                "points": fresh,
                "edges": edges,
                "pointRegions": [int(value) for value in member.tolist()],
                "regions": [
                    {
                        "id": index,
                        "size": int(counts[index]),
                        "x": float(positions[index, 0]),
                        "y": float(positions[index, 1]),
                    }
                    for index in range(len(centroids))
                    if counts[index]
                ],
            }
        )
        return len(rows)

    def _points(
        self,
        build: MapBuild,
        regions: list[dict],
        labels: np.ndarray,
        basis: tuple[np.ndarray, np.ndarray] | None,
        drawn: list[int],
    ) -> tuple[list[dict], list[int]]:
        """2-D positions for the drawn subset, each carrying its region.

        Also returns the rows drawn, in point order, so edges can be indexed
        the same way the canvas indexes points.
        """
        if basis is None or not regions or not drawn:
            return [], []
        centre, plane = basis
        known = {region["id"] for region in regions}
        kept = [row for row in drawn if int(labels[row]) in known]
        rows = np.array(kept, dtype=np.int64)
        coords = project_with(centre, plane, build.vectors[rows])
        points = []
        for offset, row in enumerate(rows.tolist()):
            region = int(labels[row])
            document = build.documents[row]
            title = str(document.get("title") or "")
            points.append(
                {
                    "id": str(document.get("id", "")),
                    "x": float(coords[offset, 0]),
                    "y": float(coords[offset, 1]),
                    "region": region,
                    "bucket": _bucket(document),
                    "title": title if len(title) <= 120 else f"{title[:119]}…",
                    "year": document.get("year") if isinstance(document.get("year"), int) else None,
                }
            )
        return points, kept

    def _edges(self, build: MapBuild, drawn: list[int], start: int = 0) -> list[list[float]]:
        """Similarity edges among drawn points, indexed by position in ``drawn``.

        ``start`` is how many of them already have their edges, so a streamed
        frame only pays for the rows it adds. Each edge is ``[i, j, cosine]``.
        """
        if len(drawn) < 2:
            return []
        found = neighbor_edges(
            build.vectors[np.array(drawn, dtype=np.int64)],
            start=start,
            neighbors=self.config.gapmap_edge_neighbors,
            floor=self.config.gapmap_edge_floor,
        )
        return [[i, j, round(cosine, 3)] for i, j, cosine in found]

    # ---------------------------------------------------------------- serving

    def _decorate(
        self, placement: dict, vector: np.ndarray | list[float], built: dict
    ) -> dict:
        """Attach the same-plane 2-D point and strip the internal centroid."""
        placement["region"] = _public(placement["region"]) if placement["region"] else None
        if built["basis"] is not None:
            mean, basis = built["basis"]
            coords = project_with(
                mean, basis, normalize(np.asarray(vector, dtype=np.float32)[None, :])
            )
            placement["point"] = {"x": float(coords[0, 0]), "y": float(coords[0, 1])}
        return placement

    async def _expression_vector(
        self, expression: MapArithmetic, warnings: list[str]
    ) -> np.ndarray | None:
        """The unit vector of ``start − remove + add``: embed each phrase, combine."""
        terms = [expression.start, *expression.remove, *expression.add]
        vectors = await asyncio.gather(*(self.embed(term) for term in terms))
        if any(vector is None for vector in vectors):
            warnings.append("Embeddings are disabled, so the expression could not be placed.")
            return None
        n_remove = len(expression.remove)
        combined = combine(
            [vectors[0], *vectors[1 + n_remove :]], vectors[1 : 1 + n_remove]
        )
        if combined is None:
            warnings.append(
                "The expression cancels itself out; the combined vector has no direction."
            )
        return combined

    async def _place_expression(
        self, expression: MapArithmetic, combined: np.ndarray | list[float], built: dict
    ) -> dict:
        """Place the combined vector like any other point on the map.

        What comes back are the real papers nearest it, so the honesty of the
        answer does not depend on how cleanly the analogy worked.
        """
        placement = await asyncio.to_thread(
            place,
            combined,
            built["documents"],
            built["regions"],
            built["gaps"],
            vectors=built["vectors"],
        )
        placement["expression"] = {
            "start": expression.start,
            "remove": expression.remove,
            "add": expression.add,
        }
        return self._decorate(placement, combined, built)

    async def assess(
        self,
        idea: str | None = None,
        arithmetic: MapArithmetic | None = None,
        cutoff_year: int | None = None,
        refresh: bool = False,
    ) -> dict:
        warnings: list[str] = []
        vector: np.ndarray | list[float] | None = None
        if arithmetic is not None:
            vector = await self._expression_vector(arithmetic, warnings)
        elif idea:
            vector = await self.embed(idea)
            if vector is None:
                warnings.append("Embeddings are disabled, so the idea could not be placed.")
        built = await self.build(refresh=refresh, vector=vector)
        coverage = built["coverage"]
        if coverage.get("scope") == "neighborhood":
            warnings.append(
                f"The map shows the {coverage['clustered']} embedded studies nearest this "
                f"question, out of {coverage['corpus']} in the index: the question's "
                "neighborhood, not the whole index."
            )
        elif coverage["corpus"] > coverage["clustered"]:
            warnings.append(
                f"The map clusters {coverage['clustered']} of {coverage['corpus']} "
                "embedded studies, not the whole index."
            )
        if coverage["drawn"] < coverage["clustered"]:
            warnings.append(
                f"All {coverage['clustered']} clustered studies shape the regions; the canvas "
                f"draws {coverage['drawn']} of them."
            )
        placement = None
        if vector is not None and arithmetic is not None:
            placement = await self._place_expression(arithmetic, vector, built)
        elif vector is not None:
            placement = await asyncio.to_thread(
                place,
                vector,
                built["documents"],
                built["regions"],
                built["gaps"],
                vectors=built["vectors"],
            )
            placement = self._decorate(placement, vector, built)
        calibration = None
        if cutoff_year is not None:
            calibration = await asyncio.to_thread(
                calibrate,
                built["documents"],
                cutoff_year,
                self.config.gapmap_regions,
                self.config.gapmap_seed,
                vectors=built["vectors"],
            )
        return {
            "version": built["version"],
            "coverage": coverage,
            "regions": [_public(region) for region in built["regions"]],
            "gaps": built["gaps"],
            "points": built["points"],
            "edges": built.get("edges", []),
            "placement": placement,
            "calibration": calibration,
            "warnings": warnings,
        }

    async def stream(
        self,
        idea: str | None = None,
        arithmetic: MapArithmetic | None = None,
        cutoff_year: int | None = None,
        refresh: bool = False,
        progress: Progress | None = None,
    ) -> dict:
        """``assess`` with the build published as it happens.

        A cached map produces no progress events: replaying a finished build as
        though it were happening would be theatre, and the caller can tell the
        difference because the first thing it receives is the finished result.
        """
        if progress is None:
            return await self.assess(idea, arithmetic, cutoff_year, refresh)
        queue: asyncio.Queue = asyncio.Queue(maxsize=64)
        self._subscribers.add(queue)
        # Embedded here rather than in the build: the build describes the corpus
        # and is shared by every viewer, while the question belongs to this one.
        asking = asyncio.create_task(self._ask(idea, arithmetic))
        pump = asyncio.create_task(self._pump(queue, progress, asking))
        try:
            return await self.assess(idea, arithmetic, cutoff_year, refresh)
        finally:
            self._subscribers.discard(queue)
            asking.cancel()
            queue.put_nowait(None)
            with contextlib.suppress(asyncio.CancelledError, Exception):
                await pump

    async def _ask(
        self, idea: str | None, arithmetic: MapArithmetic | None
    ) -> np.ndarray | None:
        """The unit vector of this viewer's question, if there is one to embed."""
        if arithmetic is not None:
            vector = await self._expression_vector(arithmetic, [])
        elif idea:
            vector = await self.embed(idea)
        else:
            return None
        if vector is None:
            return None
        return normalize(np.asarray(vector, dtype=np.float32)[None, :])[0]

    async def _pump(
        self,
        queue: asyncio.Queue,
        progress: Progress,
        asking: asyncio.Task | None = None,
    ) -> None:
        while True:
            payload = await queue.get()
            if payload is None or payload.get("event") == "built":
                return
            # The question is placed on the plane the build is drawing on, so it
            # is on the map from the first frame instead of only at the end. It
            # is where the question sits, not a claim about what surrounds it.
            if (
                asking is not None
                and asking.done()
                and not asking.cancelled()
                and asking.exception() is None
                and self._basis is not None
            ):
                vector = asking.result()
                if vector is not None:
                    centre, plane = self._basis
                    coords = project_with(centre, plane, vector[None, :])
                    payload = {
                        **payload,
                        "placement": {"x": float(coords[0, 0]), "y": float(coords[0, 1])},
                    }
            await progress({key: value for key, value in payload.items() if key != "event"})