loom-client

Python WebSocket client for the Loom Genome Browser. Provides a fully async, typed interface for programmatically controlling a running Loom instance -- navigating loci, managing tracks and regions of interest, exporting views, and subscribing to browser events.

Installation

pip install loom-client

Or install from source:

git clone https://github.com/riyavsinha/loom-client.git
cd loom-client
pip install -e .

Requires Python >= 3.11.

Quick start

Option 1: Connect to a Loom instance

Use this when your code initiates the WebSocket connection to a running Loom server.

import asyncio
from loom_client import LoomClient

async def main():
    client = LoomClient("ws://localhost:8080")
    await client.connect()

    # Get current browser state
    state = await client.get_browser_state()
    print(state.locus_string)   # e.g. "chr1:1,000-2,000"
    print(state.zoom_level)     # e.g. ZoomLevel.GENE

    # Navigate to a locus
    await client.navigate(locus="chr17:7565097-7590856")  # TP53

    # Zoom out 2x
    await client.navigate(zoom="out", factor=2.0)

    await client.close()

asyncio.run(main())

Option 2: Use an existing WebSocket connection

Use this when the WebSocket is already established — for example, a frontend-initiated connection handled by your backend, or when bridging through a proxy.

from loom_client import LoomClient

async def handle_connection(websocket):
    """Example: called by your WebSocket server when a client connects."""
    client = LoomClient.from_websocket(websocket)
    client.start_listening()

    state = await client.get_browser_state()
    print(state.locus_string)

    await client.close()

API reference

Connection

# Option 1: Initiate a new connection
client = LoomClient(url)   # Create client
await client.connect()     # Open WebSocket + start listener

# Option 2: Reuse an existing WebSocket
client = LoomClient.from_websocket(ws)  # Wrap existing connection
client.start_listening()                # Start listener

await client.close()       # Tear down

Commands

All command methods are async and return typed Pydantic models.

get_browser_state(record=None) -> ProjectedState

Returns the current viewport state including locus, zoom level, loaded tracks, ROIs, and annotations.

state = await client.get_browser_state()
for track in state.tracks:
    print(f"{track.name} ({track.type})")

navigate(locus=None, zoom=None, factor=None) -> bool

Move the viewport to a genomic locus or zoom in/out.

await client.navigate(locus="chr1:1000-2000")
await client.navigate(zoom="in", factor=3.0)

modify_tracks(actions) -> ModifyTracksResult

Add, remove, find, or update tracks in a single batch.

from loom_client import AddTrackAction, RemoveTrackAction, TrackSessionConfig, DataSourceConfig, WireTrackSelector

# Add a BigWig track
result = await client.modify_tracks([
    AddTrackAction(config=TrackSessionConfig(
        type="wig",
        name="H3K27ac",
        data_source=DataSourceConfig(
            type="bigwig",
            url="https://example.com/h3k27ac.bw",
        ),
    ))
])

# Remove by name pattern
result = await client.modify_tracks([
    RemoveTrackAction(selector=WireTrackSelector(name_regex="H3K.*"))
])

query_features(track_id, summarize=None) -> QueryFeaturesResult

Query features for a loaded track.

result = await client.query_features("track-id-123", summarize=True)
print(result.feature_count)

set_layout(tracks, locus=None) -> SetLayoutResult

Replace the entire track layout at once.

result = await client.set_layout(
    tracks=[
        TrackSessionConfig(type="ruler"),
        TrackSessionConfig(type="sequence"),
        TrackSessionConfig(
            type="annotation",
            name="GENCODE",
            data_source=DataSourceConfig(type="gencode", genome="hg38"),
        ),
    ],
    locus="chr1:1000-5000",
)

export_view(format, width=None) -> str | SessionConfig

Export the current view as SVG, PNG (returned as strings), or a session config object.

svg = await client.export_view("svg", width=1200)
session = await client.export_view("session")  # returns SessionConfig

capture(track_ids=None, locus=None, format=None, width=None, include_axis=None) -> CaptureResult

Image of just a fragment — a subset of tracks over a locus — re-rendered off-screen without disturbing the live view. Unlike export_view (the whole browser), this is small enough to hand to a vision model (e.g. inspect the motifs in one dynseq contribution track). png (default) needs a DOM-attached browser; svg works headlessly.

shot = await client.capture(
    track_ids=["dynseq-contrib"],       # omit for all tracks
    locus="chr1:1,000,000-1,000,300",   # omit for the current view
    format="png", width=800,
)
# shot.image → 'data:image/png;base64,…'  ·  shot.width / shot.height / shot.track_ids

manage_rois(action) -> ROI result

Manage regions of interest. The action type determines the operation.

from loom_client import AddROIAction, ROI, ListROIAction, FindROIAtLocusAction

# Add a region of interest
result = await client.manage_rois(AddROIAction(
    roi=ROI(id="roi-1", chr="chr17", start=7565097, end=7590856,
            name="TP53", color="rgba(255,0,0,0.3)"),
    set_name="Genes of interest",
))

# List all ROIs
result = await client.manage_rois(ListROIAction())

# Find ROIs overlapping a locus
result = await client.manage_rois(
    FindROIAtLocusAction(chr="chr17", start=7500000, end=7600000)
)

annotate(action) -> annotation result

Draw and manage the user/agent annotation layer — track-scoped box / point / freehand markup, distinct from ROIs (cross-track highlights). The server generates id/createdAt, so pass an AnnotationInput (only chr/start/end/trackIds are required; kind defaults to box, author to user).

from loom_client import (
    AddAnnotationAction, AnnotationInput, UpdateAnnotationAction,
    ListAnnotationAction, AnnotationFilter,
)

# Agent draws a glowing box to call attention to a region on one track
result = await client.annotate(AddAnnotationAction(
    annotation=AnnotationInput(
        kind="box", chr="chr17", start=7674200, end=7674260,
        track_ids=["h3k27ac"], text="likely active enhancer",
        author="agent", emphasis="glow",   # 'pulse' (size) | 'glow'/'flash' (color)
    ),
))
annotation_id = result["annotation"]["id"]

# Toggle emphasis later (e.g. a history stepper highlighting the current step's mark)
await client.annotate(UpdateAnnotationAction(
    annotation_id=annotation_id, changes={"emphasis": "pulse"}))

# List agent-authored annotations
result = await client.annotate(ListAnnotationAction(
    filter=AnnotationFilter(author="agent")))

# "Circle this area": an ellipse outline reads as a callout, not a BED-like box.
# outline=stroke-only (no fill); corner_radius 0..1 (1 = ellipse). box-only.
await client.annotate(AddAnnotationAction(
    annotation=AnnotationInput(
        kind="box", chr="chr17", start=7674200, end=7674260,
        track_ids=["h3k27ac"], author="agent",
        outline=True, corner_radius=1, text="candidate enhancer",
    ),
))

from loom_client import ResolveFeaturesAnnotationAction

# "What did the user circle?" — expand a freehand/box mark into the concrete
# features its span overlaps within its own tracks: "peaks A, B, C", not just a region.
result = await client.annotate(
    ResolveFeaturesAnnotationAction(annotation_id=annotation_id))
for r in result["resolved"]:
    for f in r["features"]:
        print(f["trackId"], f["chr"], f["start"], f["end"], f.get("name"))
# Omit annotation_id to resolve all (add filter=AnnotationFilter(...) to scope).
# Resolved on-demand against live feature caches — never stale, never stored on the
# annotation. Capped at max_features (default 200); "truncated": true flags a cut list.

Actions: AddAnnotationAction, RemoveAnnotationAction, UpdateAnnotationAction, ClearAnnotationAction, ListAnnotationAction, GetVisibleAnnotationAction, ResolveFeaturesAnnotationAction. A subscribed client receives annotationadded when a human draws, and every get_browser_state includes an annotations list of what's visible.

subscribe_events(events) -> SubscribeEventsResult

Subscribe to server-pushed events. Use ["*"] for all events or [] to unsubscribe.

await client.subscribe_events(["locuschange", "trackadded", "dataloaded"])

Event handling

Register callbacks for browser events:

def on_locus_change(data):
    locus = data["locus"]
    print(f"Moved to {locus['chr']}:{locus['start']}-{locus['end']}")

client.on("locuschange", on_locus_change)

# Wildcard handler receives (event_name, data)
client.on("*", lambda name, data: print(f"Event: {name}"))

# Unregister
client.off("locuschange", on_locus_change)

Available events: locuschange, trackadded, trackremoved, dataloaded, dataerror, rendererror, trackclick, trackhover, trackcontextmenu, trackorderchanged, roiadded, roiremoved, roichanged, roiclick, roicontextmenu

Error handling

Server errors raise CommandException:

from loom_client import CommandException

try:
    await client.navigate(locus="invalid")
except CommandException as e:
    print(e.code, e.message)

Development

pip install -e ".[dev]"
pytest

License

MIT