Use chart interactions in Python¶

Select a region in a chart and use its coordinates in your Python analysis. You can also send results back to the chart—for example, to show saved regions on an annotation track.

The example below does both. Drag across the chart, then release: Python adds the region to a list called annotations, and a mark appears in the lower track. The list stays in memory; restarting the example clears it.

The data is fictional: chrDemo has 1,000 bases. Coordinates are zero-based, with the end excluded: an interval from 10 to 20 contains 10 bases. Overlapping saved regions share one compact track.

The Python connection API used here is experimental.

Try it in a notebook¶

Download these two files into the same folder:

notebook.ipynb — the notebook to open.
component.py — the chart and Python code that handles selections.

Install the dependencies in the Python environment your notebook uses:

pip install "genome-spy-python>=0.4.0" ipywidgets ipykernel

Open the notebook in Jupyter or VS Code and select that environment’s kernel (the Python session running your cells). Run the first code cell and wait for Ready. Drag across the chart and release. The status reports what Python received, and the lower track shows the saved interval.

Run the next cell to inspect your results:

annotations

Rerun that inspection cell after another selection to refresh its output. The status above the list updates automatically.

If it does not connect¶

Make sure you installed the package in the notebook’s Python environment, then restart the kernel and run the cells again. Wait for Ready before selecting a region. Let the setup cell finish: keeping it waiting for a reply can prevent that reply from arriving in VS Code.

Add your own Python work¶

The notebook’s widget provides the connection to Python. A background task keeps listening for selections while you use other cells. Keep that setup when adapting the example.

In component.py, annotate() waits for you to finish dragging, adds the region to annotations, and updates the lower track. Put your own processing after annotations.append(record). In the notebook, you can also process the list in another cell.

Chart and annotation code

"""Shared chart and Python annotation logic for the notebook and web examples."""

import asyncio
from collections.abc import Callable
from typing import Any

import genome_spy as gs
from genome_spy.embed import EmbedResult

# Synthetic intervals on a fictional chromosome; no external data required.
features = (
    gs.Chart(
        [
            {"chrom": "chrDemo", "start": 100, "end": 300},
            {"chrom": "chrDemo", "start": 450, "end": 700},
        ]
    )
    .mark_rect(color="#547aa5")
    .encode(x=gs.Locus("chrom", "start"), x2=gs.Locus("chrom", "end"))
    .properties(height=70, title="Brush a region; release to save in Python")
)

saved = (
    gs.Chart(data={"name": "annotations"})
    .mark_rect(color="#d89632", opacity=0.7)
    .encode(
        x=gs.Locus("chrom", "start"),
        x2=gs.Locus("chrom", "end"),
        tooltip=["name:N", "chrom:N", "start:Q", "end:Q"],
    )
    .properties(height=22, title="Saved annotations")
)

chart = (
    (features & saved)
    .properties(
        width=700,
        assembly="demo",
        genomes={"demo": {"contigs": [{"name": "chrDemo", "size": 1000}]}},
        datasets={"annotations": []},
        scales=gs.scales(
            x=gs.Scale(
                domain=[
                    {"chrom": "chrDemo", "pos": 0},
                    {"chrom": "chrDemo", "pos": 1000},
                ]
            )
        ),
    )
    .add_params(
        gs.selection_interval(
            "brush", encodings=["x"], extent="container", on={"type": "mousedown"}
        )
    )
)


async def annotate(
    api: EmbedResult,
    annotations: list[dict[str, Any]],
    show: Callable[[str], None],
) -> None:
    """Listen until cancelled; save committed brushes and update the chart."""
    events: asyncio.Queue = asyncio.Queue()
    async with asyncio.timeout(30):
        brush = await api.params.get_selection("brush")
        stop = await brush.subscribe(events.put_nowait, {"delivery": "commit"})
    show("Ready — brush a region and release to save it in Python.")
    try:
        while True:
            snapshot = await events.get()
            endpoints = snapshot.get("complexIntervals", {}).get("x")
            if not snapshot.get("active") or not endpoints:
                continue
            left, right = endpoints
            start, end = left["pos"], right["pos"]
            if left["chrom"] != right["chrom"] or not 0 <= start < end:
                show("Choose a nonempty interval within one chromosome.")
                continue
            record = {
                "chrom": left["chrom"],
                "start": start,
                "end": end,
                "name": f"region_{len(annotations) + 1}",
            }
            async with asyncio.timeout(30):
                await api.datasets.set("annotations", [*annotations, record])
            annotations.append(record)
            show(f"Python saved {len(annotations)} region(s): {record}")
    finally:
        async with asyncio.timeout(2):
            await stop()

For named annotations and BED export, use Annotate genomic intervals. The other interactive workflows show gene selection and sequence editing. Their web demos run without Python; download their notebooks to work with the results in Python.

Optional: run the same example as a web app¶

Use this when you want a browser page to send selections to Python outside a notebook. If you only want to share an interactive chart, save it as HTML instead—no server is needed.

Put these files in one folder:

component.py — the same Python code.
server.py — the Python program that serves the page.
index.html — the page shown in your browser.

Install uv if needed. Open a terminal in that folder and run:

uv run --with "genome-spy-python>=0.4.0" --with aiohttp server.py

Keep the command running and open http://127.0.0.1:8080. This address opens the example on your own computer. Drag and release to save a region in Python. Stop the program with Ctrl+C.

Each browser tab has its own Python list. Reloading that tab starts over. Nothing is written to disk.

Python server

"""Local-only web host. Run: uv run --with aiohttp server.py."""

import asyncio
from contextlib import suppress
from importlib.resources import files
import logging
from pathlib import Path

from aiohttp import WSMsgType, web

from component import annotate, chart
from genome_spy.embed import EmbedError, attach_embed

logger = logging.getLogger(__name__)


async def index(request):
    return web.FileResponse(Path(__file__).with_name("index.html"))


async def specification(request):
    return web.json_response(chart.to_dict())


async def javascript(request):
    name = request.match_info["name"]
    if name not in {"genome-spy.js", "embed-bridge.js"}:
        raise web.HTTPNotFound()
    source = files("genome_spy").joinpath("static", name).read_bytes()
    return web.Response(body=source, content_type="text/javascript")


async def websocket(request):
    # Do not let unrelated websites connect to this local Python process.
    if request.headers.get("Origin") != f"http://{request.host}":
        raise web.HTTPForbidden(text="A same-origin browser connection is required.")
    socket = web.WebSocketResponse(heartbeat=20)
    await socket.prepare(request)
    outgoing = asyncio.Queue()
    listeners = []
    annotations = []  # One Python list per connection, never shared between tabs.

    def subscribe(callback):
        listeners.append(callback)
        return lambda: listeners.remove(callback)

    async def send():
        # One writer preserves the order required by attach_embed.
        while True:
            await socket.send_json(await outgoing.get())

    async def receive():
        async for message in socket:
            if message.type == WSMsgType.TEXT:
                for callback in list(listeners):
                    callback(message.json())
            elif message.type == WSMsgType.ERROR:
                raise socket.exception()

    async def run_annotations():
        async with asyncio.timeout(30):
            api = await attach_embed(outgoing.put_nowait, subscribe)
        try:
            await annotate(
                api,
                annotations,
                lambda text: outgoing.put_nowait({"status": text}),
            )
        finally:
            with suppress(TimeoutError, EmbedError):
                async with asyncio.timeout(2):
                    await api.finalize()

    tasks = [asyncio.create_task(fn()) for fn in (send, receive, run_annotations)]
    try:
        done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            task.result()
    except Exception:
        logger.exception("Annotation session failed")
    finally:
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
        await socket.close()
    return socket


def create_app():
    app = web.Application()
    app.add_routes(
        [
            web.get("/", index),
            web.get("/spec.json", specification),
            web.get("/static/{name}", javascript),
            web.get("/ws", websocket),
        ]
    )
    return app


if __name__ == "__main__":
    web.run_app(create_app(), host="127.0.0.1", port=8080)

This is a local teaching example, not a public hosting setup. Publishing an app requires additional security and decisions about storing users’ results.