from __future__ import annotations
import functools
from collections import defaultdict
from typing import (
TYPE_CHECKING,
ClassVar,
Generic,
Literal,
TypeVar,
Union,
cast,
overload,
)
import higlass_schema as hgs
from pydantic import BaseModel
import higlass._display as display
import higlass._utils as utils
__all__ = [
"EnumTrack",
"HeatmapTrack",
"IndependentViewportProjectionTrack",
"CombinedTrack",
"PluginTrack",
"TrackT",
"View",
"ViewT",
"Viewconf",
"concat",
"hconcat",
"vconcat",
"track",
"view",
"combine",
"divide",
"lock",
]
if TYPE_CHECKING:
from higlass.server import TilesetResource
## Mixins
# Can't figure out a good way to type these classes.
# We ignoring the errors in the parameter signature
# mean we can get good type information within the
# function and return types are inferred for end users.
class _PropertiesMixin:
def properties(
self: utils.ModelT, # type: ignore
inplace: bool = False,
**fields, # type: ignore
) -> utils.ModelT: # type: ignore
"""Configures top-level properties.
Updates top-level properties for a Track, View, or Viewconf. This
is really a convenience to allow method chaining/derived objects
to be created without multiple lines of mutating the classes. For
example,
>>> view = hg.view(hg.track("heatmap"))
>>> updated_view = view.properties(zoomFixed=True)
>>> assert view.uid != updated_view.uid
>>> assert view.zoomFixed is None
>>> assert updated_view.zoomFixed is True
inplace : bool, optional
Whether to modify the existing track in place or return
a new track with the options applied (default: `False`).
**fields : dict
The updated properties for .
Returns
-------
track : A track with the the newly specified track options.
"""
model = self if inplace else utils.copy_unique(self)
for k, v in fields.items():
setattr(model, k, v)
return model
class _OptionsMixin:
def opts(
self: TrackT, # type: ignore
inplace: bool = False,
**options,
) -> TrackT: # type: ignore
"""Configures options for a Track.
A convenience method to update `track.options`.
Parameters
----------
inplace : bool, optional
Whether to modify the existing track in place or return
a new track with the options applied (default: `False`).
**options : dict
Options to pass down to the underlying track object.
Returns
-------
track : A track with the the newly specified track options.
Examples
--------
>>> track = hg.track("heatmap")
>>> derived = track.opts(colorRange=["rgba(255,255,255,1)", "rgba(0,0,0,1)"])
>>> assert track.options is None
>>> assert isinstance(derived.options["colorRange"], list)
"""
track = self if inplace else utils.copy_unique(self)
if track.options is None:
track.options = {}
track.options.update(options)
return track
class _TilesetMixin:
def tileset(
self: TrackT, # type: ignore
tileset: TilesetResource,
inplace: bool = False,
) -> TrackT: # type: ignore
"""Replace or add a tileset to a Track.
A convenience method to update a track with a tileset.
Parameters
----------
tileset : TilesetResource
A tileset resource returned from `hg.server`.
inplace : bool, optional
Whether to modify the existing track in place or return
a new track (default: `False`)
Returns
-------
track : A track with the bound tileset.
"""
track = self if inplace else utils.copy_unique(self)
track.server = tileset.server
track.tilesetUid = tileset.tileset.uid
return track
## Extend higlass-schema classes
[docs]
class EnumTrack(hgs.EnumTrack, _OptionsMixin, _PropertiesMixin, _TilesetMixin):
"""Represents a generic track."""
...
[docs]
class HeatmapTrack(hgs.HeatmapTrack, _OptionsMixin, _PropertiesMixin, _TilesetMixin):
"""Represets a specialized heatmap track."""
...
[docs]
class IndependentViewportProjectionTrack(
hgs.IndependentViewportProjectionTrack,
_OptionsMixin,
_PropertiesMixin,
_TilesetMixin,
):
"""Represents a view-independent viewport projection track."""
...
[docs]
class CombinedTrack(hgs.CombinedTrack, _OptionsMixin, _PropertiesMixin, _TilesetMixin):
"""Represents a track combining multiple tracks."""
...
[docs]
class PluginTrack(hgs.BaseTrack, _OptionsMixin, _PropertiesMixin, _TilesetMixin):
"""Represents an unknown plugin track."""
plugin_url: ClassVar[str]
Track = Union[
HeatmapTrack,
IndependentViewportProjectionTrack,
EnumTrack,
CombinedTrack,
PluginTrack,
]
TrackT = TypeVar("TrackT", bound=Track)
[docs]
class View(hgs.View[TrackT], _PropertiesMixin, Generic[TrackT]):
"""Represets a HiGlass View that is generic over the inner tracks."""
[docs]
def domain(
self,
x: hgs.Domain | None = None,
y: hgs.Domain | None = None,
inplace: bool = False,
):
"""Configures the view x and/or y domain(s).
Parameters
----------
x : tuple[float, float], optional
The x domain for the view (default: `None`)
y : tuple[float, float], optional
The y domain for the view (default: `None`)
inplace : bool, optional
Whether to modify this view inplace. If `False` (default),
a new view is created and returned with the domains
applied (default: `False`)
Returns
-------
view : a View with the updated x & y domains
"""
view = self if inplace else utils.copy_unique(self)
if x is not None:
view.initialXDomain = x
if y is not None:
view.initialYDomain = y
return view
def __or__(self, other: View[TrackT] | Viewconf[TrackT]):
"""Horizontally concatenate with another view or viewconf.
A convenience method for `hg.hconcat`.
Parameters
----------
other : View | Viewconf
The other view or viewconf to combine with.
Returns
-------
viewconf : A combined Viewconf
"""
return hconcat(self, other)
def __truediv__(self, other: View[TrackT] | Viewconf[TrackT]):
"""Vertically concatenate two Views.
A convenience method for `hg.vconcat`.
Parameters
----------
other : View | Viewconf
The other view or viewconf to combine with.
Returns
-------
viewconf : A combined Viewconf
"""
return vconcat(self, other)
[docs]
def clone(self):
"""Clone the the View instance.
Inserts a unique uuid so the view can be uniquely identified in the front end.
Returns
-------
view : A copy of the original view with a new uuid
"""
return utils.copy_unique(self)
[docs]
def viewconf(self, **kwargs):
"""Consumes the current View into a top-level view config.
Returns
-------
viewconf : A top-level HiGlass Viewconf.
"""
return Viewconf[TrackT](views=[self], **kwargs)
def _repr_mimebundle_(self, include=None, exclude=None):
""" "Displays the view in an IPython environment."""
return self.viewconf()._repr_mimebundle_(include, exclude)
[docs]
def project(
self,
view: View,
on: Literal["center", "top", "bottom", "left", "right"] = "center",
inplace: bool = False,
**kwargs,
):
"""Project a different viewport onto this view.
Creates a viewport-projection track whos bounds are synchonrized
with an existing, separate view.
Parameters
----------
view : View
The view with the desired viewport to project.
on : Literal["center", "top", "bottom", "left", "right"], optional
Where to project the viewport on this view (default: "center").
inplace : bool
Whether to modify this view inplace, or alternatively return
a new view (default: `False`).
**kwargs : dict
Additional parameters for the created viewport-projection track.
Returns
-------
view : A view with a new viewport-projection track.
"""
new_view = self if inplace else utils.copy_unique(self)
# projection track type from position
if on == "center":
track_type = "viewport-projection-center"
elif on == "top" or on == "bottom":
track_type = "viewport-projection-horizontal"
elif on == "left" or on == "right":
track_type = "viewport-projection-vertical"
else:
raise ValueError("Not possible")
if getattr(new_view.tracks, on) is None:
setattr(new_view.tracks, on, [])
trk = track(type_=track_type, fromViewUid=view.uid, **kwargs)
getattr(new_view.tracks, on).append(trk)
return new_view
ViewT = TypeVar("ViewT", bound=View)
def gather_plugin_urls(views: list[ViewT]) -> list[str]:
"""Inspect tracks and extract plugin urls to inject into HTML."""
plugin_urls = {}
for view in views:
for _, track in view.tracks:
if isinstance(track, PluginTrack):
plugin_urls[track.type] = track.plugin_url
return list(plugin_urls.values())
[docs]
class Viewconf(hgs.Viewconf[View[TrackT]], _PropertiesMixin, Generic[TrackT]):
"""Represents a top-level viewconfig, or a complete HiGlass visualization."""
def _repr_mimebundle_(self, include=None, exclude=None):
""" "Displays the view config in an IPython environment."""
renderer = display.renderers.get()
plugin_urls = [] if self.views is None else gather_plugin_urls(self.views)
return renderer(self.dict(), plugin_urls=plugin_urls)
[docs]
@classmethod
def from_url(cls, url: str, **kwargs):
"""Load a viewconf via URL and construct a Viewconf.
Makes an HTTP request.
Parameters
----------
url : str
The URL for a JSON HiGlass view config.
**kwargs : dict
Options for the `urllib.Request` instance.
Returns
-------
viewconf : The parsed view config
"""
import urllib.request as urllib
request = urllib.Request(url, **kwargs)
with urllib.urlopen(request) as response:
raw = response.read()
return cls.parse_raw(raw)
[docs]
def locks(
self,
*locks: hgs.Lock | hgs.ValueScaleLock,
zoom: list[hgs.Lock] | hgs.Lock | None = None,
location: list[hgs.Lock] | hgs.Lock | None = None,
value_scale: list[hgs.ValueScaleLock] | hgs.ValueScaleLock | None = None,
inplace: bool = False,
):
"""Specify view locks.
Parameters
----------
*locks : hgs.Lock | hgs.ValueScaleLock, optional
A set of either zoom/location or value scale locks to add. Instances
of `hgs.Lock` will synchronize _both_ zoom and location for the views
included in the lock. If you'd like to synchronize just zoom or location,
use the explicit keyword arguments below.
zoom : hgs.Lock | list[hgs.Lock], optional
A lock or set of locks to synchronize only the zoom.
location : hgs.Lock | list[hgs.Lock], optional
A lock or set of locks to synchronize only the location.
value_scale : hgs.ValueScaleLock | list[hgs.ValueScaleLock], optional
A single or set of value-scale locks.
inplace : bool, optional
Whether to modify the viewconf in place or return a new viewconf
with the locks applied (default: `False`)
Returns
-------
viewconf : A Viewconf with the synchronized views.
Examples
--------
>>> view1 = hg.view(hg.track("heatmap"))
>>> view2 = hg.view(hg.track("heatmap"))
>>> # create an abstract lock for two views
>>> view_lock = hg.lock(view1, view2)
>>> # lock location & zoom
>>> (view1 | view2).lock(view_lock)
>>> # lock only zoom
>>> (view1 | view2).lock(zoom=view_lock)
>>> # lock only location
>>> (view1 | view2).lock(location=view_lock)
"""
conf = self if inplace else utils.copy_unique(self)
zoom = utils.ensure_list(zoom)
location = utils.ensure_list(location)
value_scale = utils.ensure_list(value_scale)
shared_locks: list[hgs.Lock] = []
for lock in locks:
if isinstance(lock, hgs.Lock):
shared_locks.append(lock)
else:
value_scale.append(lock)
zoom.extend(shared_locks)
location.extend(shared_locks)
if conf.zoomLocks is None:
conf.zoomLocks = hgs.ZoomLocks()
for lock in zoom:
assert isinstance(lock.uid, str)
conf.zoomLocks.locksDict[lock.uid] = lock
for vuid, _ in lock:
conf.zoomLocks.locksByViewUid[vuid] = lock.uid
if conf.locationLocks is None:
conf.locationLocks = hgs.LocationLocks()
for lock in location:
assert isinstance(lock.uid, str)
conf.locationLocks.locksDict[lock.uid] = lock
for vuid, _ in lock:
conf.locationLocks.locksByViewUid[vuid] = lock.uid
if conf.valueScaleLocks is None:
conf.valueScaleLocks = hgs.ValueScaleLocks()
for lock in value_scale:
assert isinstance(lock.uid, str)
conf.valueScaleLocks.locksDict[lock.uid] = lock
for vuid, _ in lock:
conf.valueScaleLocks.locksByViewUid[vuid] = lock.uid
return conf
def __or__(self, other: View[TrackT] | Viewconf[TrackT]) -> Viewconf[TrackT]:
"""Horizontally concatenate with another view or viewconf.
A convenience method for `hg.hconcat`.
Parameters
----------
other : View | Viewconf
The other view or viewconf to combine with.
Returns
-------
viewconf : A combined Viewconf
"""
return hconcat(self, other)
def __truediv__(self, other: View[TrackT] | Viewconf[TrackT]) -> Viewconf[TrackT]:
"""Vertically concatenate with another view or viewconf.
A convenience method for `hg.hconcat`.
Parameters
----------
other : View | Viewconf
The other view or viewconf to combine with.
Returns
-------
viewconf : A combined Viewconf
"""
return vconcat(self, other)
TrackTA = TypeVar("TrackTA", bound=Track)
TrackTB = TypeVar("TrackTB", bound=Track)
[docs]
def concat(
method: Literal["horizontal", "vertical"],
a: View[TrackTA] | Viewconf[TrackTA],
b: View[TrackTB] | Viewconf[TrackTB],
) -> Viewconf[TrackTA | TrackTB]:
"""Concatenate two views or separate viewconfs together.
Uses the layout of one view or the total bounds of all views in one
viewconf to offset the other view/viewconf.
Parameters
----------
method : Literal["horizontal", "vertical"]
How to concatenate views/viewconfs.
a : View | Viewconf
A view or viewconf to combine.
b : View | Viewconf
The other view or viewconf to combine.
Returns
-------
viewconf : A combined viewconf containing multiple views.
"""
a = a.viewconf() if isinstance(a, View) else a
assert a.views is not None
b = b.viewconf() if isinstance(b, View) else b
assert b.views is not None
if method == "vertical":
def mapper(view):
return view.layout.y + view.layout.h
field = "y"
elif method == "horizontal":
def mapper(view):
return view.layout.x + view.layout.w
field = "x"
else:
raise ValueError("concat method must be 'vertical' or 'horizontal'.")
# gather views and adjust layout
views = [v.copy(deep=True) for v in b.views]
offset = 0 if a.views is None else max(map(mapper, a.views))
for view in views:
curr = getattr(view.layout, field)
setattr(view.layout, field, curr + offset)
a.views.extend(views) # type: ignore
# merge locks
for lockattr in ["zoomLocks", "valueScaleLocks", "locationLocks"]:
locks = getattr(b, lockattr)
if locks:
if getattr(a, lockattr) is None:
setattr(a, lockattr, locks.copy(deep=True))
else:
getattr(a, lockattr).locksByViewUid.update(locks.locksByViewUid)
getattr(a, lockattr).locksDict.update(locks.locksDict)
return cast(Viewconf[Union[TrackTA, TrackTB]], a)
hconcat = functools.partial(concat, "horizontal")
vconcat = functools.partial(concat, "vertical")
## Top-level functions to easily create tracks,
# TODO: register plugins globally to work here?
class _TrackCreator(BaseModel):
"""Create track instances from their track type.
Used internally by `hg.track` to leverage pydantic's ability to get
the appropriate base model by the track type.
Example:
-------
>>> assert isinstance(_TrackCreator(type="heatmap").__root__, HeatmapTrack)
"""
__root__: Track
@overload
def track(type_: hgs.EnumTrackType, uid: str | None = None, **kwargs) -> EnumTrack: ...
@overload
def track(
type_: Literal["heatmap"], uid: str | None = None, **kwargs
) -> HeatmapTrack: ...
@overload
def track(type_: str, uid: str | None = None, **kwargs) -> PluginTrack: ...
[docs]
def track(
type_: hgs.EnumTrackType | Literal["heatmap"] | str,
uid: str | None = None,
**kwargs,
) -> Track:
"""Create a HiGlass track.
Parameters
----------
type_ : str
The track type to create
uid: str, optional
A unique id for the track. If unspecified (default), a unique id is given
internally.
**kwargs: dict
Other top-level track properties.
Returns
-------
track : An instance of an hg.Track
"""
if uid is None:
uid = utils.uid()
data = dict(type=type_, uid=uid, **kwargs)
return _TrackCreator.parse_obj(data).__root__
[docs]
def view(
*_tracks: TrackT | tuple[TrackT, utils.TrackPosition] | hgs.Tracks[TrackT],
x: int = 0,
y: int = 0,
width: int = 12,
height: int = 6,
tracks: hgs.Tracks[TrackT] | None = None,
layout: hgs.Layout | None = None,
uid: str | None = None,
**kwargs,
) -> View[TrackT]:
"""Create a HiGlass view from multiple tracks.
Parameters
----------
*_tracks : Track | tuple[Track, str] | hgs.Tracks[Track]
The tracks to include in the view. Can be 1) separate track objects
(position inferred), 2) explicit (Track, position) tuples, or 3.) a
`higlass_schema.Tracks` Object specifying all tracks and positions.
x : int, optional
The x position of the view in the HiGlass grid (default: 0). Used to
created a Layout if none is provided.
y : int, optional
The y position of the view in the HiGlass grid (default: 0). Used to
created a Layout if none is provided.
width : int, optional
The width of the view (default: 12). Used to created a Layout if none is
provided.
height : int, optional
The height of the view (default: 6). Used to created a Layout if none is
provided.
layout : hgs.Layout, optional
An explicit layout for the view (default: None). If provided the `x`, `y`,
`width`, and `height` parameters are ignored.
uid: str, optional
A unique id for the view. If unspecified (default), a unique id is automatically
generated internally.
**kwargs: dict
Other top-level view properties.
Returns
-------
view : An instance of an hg.View
"""
if layout is None:
layout = hgs.Layout(x=x, y=y, w=width, h=height)
else:
layout = hgs.Layout(**layout.dict())
if tracks is None:
data = defaultdict(list)
else:
data = defaultdict(list, tracks.dict())
for track in _tracks:
if isinstance(track, hgs.Tracks):
track = track.dict()
for position, track_list in track.items():
data[position].extend(track_list)
else:
if isinstance(track, tuple):
track, position = track
else:
position = utils.track_default_position.get(track.type)
if position is None:
raise ValueError("No default track type")
data[position].append(track)
if uid is None:
uid = utils.uid()
return View[TrackT](
layout=layout,
tracks=hgs.Tracks[TrackT](**data),
uid=uid,
**kwargs,
)
[docs]
def combine(t1: Track, t2: Track, uid: str | None = None, **kwargs) -> CombinedTrack:
"""A utility to create a CombinedTrack from two other tracks.
Parameters
----------
t1 : hg.Track
The first of two tracks to combine.
t2 : hg.Track
The second of two tracks to combine.
uid : str, optional
A unique id for the newly created CombinedTrack.
**kwargs : dict
Additional properties to pass to the CombinedTrack constructor.
Returns
-------
combined_track : An CombinedTrack with two children tracks.
"""
if uid is None:
uid = utils.uid()
if isinstance(t1, CombinedTrack):
copy = CombinedTrack(**t1.dict())
copy.contents.append(t2.__class__(**t2.dict()))
for key, val in kwargs.items():
setattr(copy, key, val)
return copy
return CombinedTrack(
type="combined",
uid=uid,
contents=[track.__class__(**track.dict()) for track in (t1, t2)],
**kwargs,
)
T = TypeVar("T", bound=Union[EnumTrack, HeatmapTrack])
[docs]
def divide(t1: T, t2: T, **kwargs) -> T:
"""A utility to create a divided track.
Tracks must the the same type.
Parameters
----------
t1 : hg.EnumTrack | hg.HeatmapTrack
The first track to divide by the other.
t2 : hg.EnumTrack | hg.HeatmapTrack
The other track.
**kwargs : dict
Additional properties to pass to the newly created track.
Returns
-------
divided_track : An identical track with "divided" data sources.
"""
assert t1.type == t2.type, "divided tracks must be same type"
assert isinstance(t1.tilesetUid, str)
assert isinstance(t1.server, str)
assert isinstance(t2.tilesetUid, str)
assert isinstance(t2.server, str)
copy = utils.copy_unique(t1)
copy.tilesetUid = None
copy.server = None
copy.data = hgs.Data(
type="divided",
children=[
{
"tilesetUid": track.tilesetUid,
"server": track.server,
}
for track in (t1, t2)
],
)
# overrides
for key, val in kwargs.items():
setattr(copy, key, val)
return copy
@overload
def lock(*views: View, uid: str | None = None) -> hgs.Lock:
"""Create an abstract lock linking two or more views.
Parameters
----------
*views : View
Two or more views to synchronize with this lock.
uid : str, optional
An (optional) unique id for this lock.
Returns
-------
lock : An abstract lock linking two or more views.
"""
...
@overload
def lock(
*pairs: tuple[View, Track],
uid: str | None = None,
ignoreOffScreenValues: bool | None = None,
) -> hgs.ValueScaleLock:
"""Create an abstract value-scale lock linking one or more (View, Track) pairs.
Parameters
----------
*pairs : tuple[View, Track]
One or more (View, Track) pairs specifying a Track whos values should
be used to scale all tracks within the View.
uid : str, optional
An (optional) unique id for this lock.
ignoreOffScreenValues : bool, optional
Whether to ignore off screen values when scaling the view.
Returns
-------
lock : A value-scale lock defining how the tracks in a View should be scaled by
by a particular track.
"""
...
[docs]
def lock(*data, uid: str | None = None, **kwargs):
"""Create an abstract lock or value-scale lock.
Overloaded to either return a `hgs.Lock` or `hgs.ValueScaleLock` depending on
the arguments. See `Viewconf.locks` for how to use the created locks in
your top-level view config.
Examples
--------
Create an abstract lock linking two or more separate views.
>>> view1 = hg.view(hg.track("heatmap"))
>>> view2 = hg.view(hg.track("heatmap"))
>>> lock = hg.lock(view1, view2)
Create a value-scale lock, scaling two views by the values of a particular track.
>>> t1 = hg.track("heatmap")
>>> t2 = hg.track("heatmap")
>>> view1 = hg.view(t1, t2)
>>> view2 = hg.view(t1, t2)
>>> value_scale_lock = hg.lock((view1, t2), (view2, t2))
"""
assert len(data) >= 1
if uid is None:
uid = utils.uid()
if isinstance(data[0], View):
lck = hgs.Lock(uid=uid)
for view in data:
assert isinstance(view.uid, str)
setattr(lck, view.uid, (1, 1, 1))
return lck
else:
lck = hgs.ValueScaleLock(uid=uid, **kwargs)
for view, track in data:
assert isinstance(view.uid, str)
assert isinstance(track.uid, str)
vtuid = f"{view.uid}.{track.uid}"
setattr(lck, vtuid, {"track": track.uid, "view": view.uid})
return lck