Skip to content

Add higlass.tilesets.register decorator #172

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
manzt merged 8 commits into from
Feb 12, 2025
Merged

Add higlass.tilesets.register decorator #172

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
a6dfeda
Add `higlass.tilesets.register` decorator
manzt Feb 11, 2025
478b49f
Re-export tilesets
manzt Feb 11, 2025
ee77453
Automatic uid
manzt Feb 11, 2025
54676b2
Go through registry to get identifier
manzt Feb 11, 2025
64288c7
Make uids implicit in higlass-python
manzt Feb 12, 2025
faffc61
Update tests
manzt Feb 12, 2025
e3abe80
Compatability with 3.8
manzt Feb 12, 2025
f9fefd0
make protocol position only
manzt Feb 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 2 additions & 10 deletions src/higlass/__init__.py
View file
Open in desktop
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
View file
Open in desktop
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
View file
Open in desktop
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