Skip to content

Commit

Permalink
Add higlass.tilesets.register decorator (#172)
Browse files Browse the repository at this point in the history
* Add `higlass.tilesets.register` decorator

Refactors tileset implementations by introducing the
`higlass.tilesets.register` decorator.  When applied to a class
implementing `TilesetProtocol`, this decorator adds a `track()`  method,
enabling track creation while automatically registering the tileset
instance  in the global `TilesetRegistry`.
  • Loading branch information
manzt authored Feb 12, 2025
1 parent b1d8fb4 commit bff6880
Show file tree
Hide file tree
Showing 7 changed files with 253 additions and 1,202 deletions.
12 changes: 2 additions & 10 deletions src/higlass/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,15 +10,7 @@

from higlass_schema import *

import higlass.tilesets
from higlass._tileset_registry import create_jupyter_track_helper
import higlass.tilesets as tilesets
from higlass.api import *
from higlass.fuse import fuse
from higlass.tilesets import remote

bigwig = create_jupyter_track_helper(higlass.tilesets.bigwig)
multivec = create_jupyter_track_helper(higlass.tilesets.multivec)
cooler = create_jupyter_track_helper(higlass.tilesets.cooler)
hitile = create_jupyter_track_helper(higlass.tilesets.hitile)
bed2ddb = create_jupyter_track_helper(higlass.tilesets.bed2ddb)
beddb = create_jupyter_track_helper(higlass.tilesets.beddb)
from higlass.tilesets import bed2ddb, beddb, bigwig, cooler, hitile, multivec, remote
79 changes: 22 additions & 57 deletions src/higlass/_tileset_registry.py
Original file line number Diff line number Diff line change
@@ -1,38 +1,44 @@
from __future__ import annotations

import functools
import typing
import weakref
from dataclasses import dataclass

from typing_extensions import ParamSpec
__all__ = ["TilesetInfo", "TilesetProtocol", "TilesetRegistry"]

import higlass.api
from higlass._utils import datatype_default_track

if typing.TYPE_CHECKING:
from higlass.tilesets import LocalTileset, TrackType
class Transform(typing.TypedDict):
name: str
value: str


__all__ = [
"JupyterTrackHelper",
"TilesetRegistry",
"create_jupyter_track_helper",
]
class TilesetInfo(typing.TypedDict):
resolutions: tuple[int, ...]
transforms: list[Transform]
max_pos: list[int]
min_pos: list[int]
chromsizes: list[tuple[str, int]]


class TilesetProtocol(typing.Protocol):
def tiles(self, tile_ids: typing.Sequence[str], /) -> list[dict]: ...

def info(self) -> TilesetInfo: ...


class TilesetRegistry:
_registry: weakref.WeakValueDictionary[str, LocalTileset] = (
_registry: weakref.WeakValueDictionary[str, TilesetProtocol] = (
weakref.WeakValueDictionary()
)

@classmethod
def add(cls, tileset: LocalTileset) -> None:
def add(cls, tileset: TilesetProtocol) -> str:
"""Register a tileset with a given ID."""
cls._registry[tileset.uid] = tileset
uid = f"hg_{id(tileset):x}"
cls._registry[uid] = tileset
return uid

@classmethod
def get(cls, tileset_id: str) -> LocalTileset:
def get(cls, tileset_id: str) -> TilesetProtocol:
"""Retrieve a tileset by its ID, or None if it no longer exists."""
tileset = cls._registry.get(tileset_id)
if tileset is None:
Expand All @@ -42,44 +48,3 @@ def get(cls, tileset_id: str) -> LocalTileset:
@classmethod
def clear(cls) -> None:
cls._registry.clear()


@dataclass(frozen=True)
class JupyterTrackHelper:
tileset: LocalTileset

def track(self, type_: TrackType | None = None, **kwargs):
# use default track based on datatype if available
if type_ is None:
if getattr(self.tileset, "datatype", None) is None:
raise ValueError("No default track for tileset")
else:
type_ = typing.cast(
TrackType, datatype_default_track[self.tileset.datatype]
)
track = higlass.api.track(
type_=type_,
server="jupyter",
tilesetUid=self.tileset.uid,
**kwargs,
)
if self.tileset.name:
track.opts(name=self.tileset.name, inplace=True)
return track


_P = ParamSpec("_P")


def create_jupyter_track_helper(
tileset_fn: typing.Callable[_P, LocalTileset],
) -> typing.Callable[_P, JupyterTrackHelper]:
"""Create a top-level helper function that adds the tileset to the factory."""

@functools.wraps(tileset_fn)
def wrapper(*args: typing.Any, **kwargs: typing.Any) -> JupyterTrackHelper:
tileset = tileset_fn(*args, **kwargs)
TilesetRegistry.add(tileset)
return JupyterTrackHelper(tileset)

return wrapper # type: ignore
5 changes: 5 additions & 0 deletions src/higlass/_widget.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import functools
import itertools
import json
import logging
import os
import pathlib
import typing
Expand All @@ -17,6 +18,8 @@

__all__ = ["HiGlassWidget"]

logger = logging.getLogger("higlass.widget")


class TilesetInfo(pydantic.BaseModel):
"""A tileset_info request payload."""
Expand Down Expand Up @@ -59,8 +62,10 @@ def get_instance(cls):

def _handle_custom_message(self, widget, msg, buffers):
message = CustomMessage(**msg)
logger.debug("handle_custom_message: %s", message)

def respond_with(payload: object):
logger.debug("handle_custom_message::respond_with: %s", message.id)
self.send({"id": message.id, "payload": payload})

def process_message():
Expand Down
Loading

0 comments on commit bff6880

Please sign in to comment.