632 lines
20 KiB
Python
632 lines
20 KiB
Python
# piker: trading gear for hackers
|
|
# Copyright (C) Tyler Goodlet (in stewardship for pikers)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
'''
|
|
Charting overlay helpers.
|
|
|
|
'''
|
|
from typing import Callable, Optional
|
|
|
|
from pyqtgraph.Qt.QtCore import (
|
|
# QObject,
|
|
# Signal,
|
|
Qt,
|
|
# QEvent,
|
|
)
|
|
|
|
from pyqtgraph.graphicsItems.AxisItem import AxisItem
|
|
from pyqtgraph.graphicsItems.ViewBox import ViewBox
|
|
from pyqtgraph.graphicsItems.GraphicsWidget import GraphicsWidget
|
|
from pyqtgraph.graphicsItems.PlotItem.PlotItem import PlotItem
|
|
from pyqtgraph.Qt.QtCore import QObject, Signal, QEvent
|
|
from pyqtgraph.Qt.QtWidgets import QGraphicsGridLayout, QGraphicsLinearLayout
|
|
|
|
from ._interaction import ChartView
|
|
|
|
__all__ = ["PlotItemOverlay"]
|
|
|
|
|
|
# Define the layout "position" indices as to be passed
|
|
# to a ``QtWidgets.QGraphicsGridlayout.addItem()`` call:
|
|
# https://doc.qt.io/qt-5/qgraphicsgridlayout.html#addItem
|
|
# This was pulled from the internals of ``PlotItem.setAxisItem()``.
|
|
_axes_layout_indices: dict[str] = {
|
|
# row incremented axes
|
|
'top': (1, 1),
|
|
'bottom': (3, 1),
|
|
|
|
# view is @ (2, 1)
|
|
|
|
# column incremented axes
|
|
'left': (2, 0),
|
|
'right': (2, 2),
|
|
}
|
|
# NOTE: To clarify this indexing, ``PlotItem.__init__()`` makes a grid
|
|
# with dimensions 4x3 and puts the ``ViewBox`` at postiion (2, 1) (aka
|
|
# row=2, col=1) in the grid layout since row (0, 1) is reserved for
|
|
# a title label and row 1 is for any potential "top" axis. Column 1
|
|
# is the "middle" (since 3 columns) and is where the plot/vb is placed.
|
|
|
|
|
|
class ComposedGridLayout:
|
|
'''
|
|
List-like interface to managing a sequence of overlayed
|
|
``PlotItem``s in the form:
|
|
|
|
| | | | | top0 | | | | |
|
|
| | | | | top1 | | | | |
|
|
| | | | | ... | | | | |
|
|
| | | | | topN | | | | |
|
|
| lN | ... | l1 | l0 | ViewBox | r0 | r1 | ... | rN |
|
|
| | | | | bottom0 | | | | |
|
|
| | | | | bottom1 | | | | |
|
|
| | | | | ... | | | | |
|
|
| | | | | bottomN | | | | |
|
|
|
|
Where the index ``i`` in the sequence specifies the index
|
|
``<axis_name>i`` in the layout.
|
|
|
|
The ``item: PlotItem`` passed to the constructor's grid layout is
|
|
used verbatim as the "main plot" who's view box is give precedence
|
|
for input handling. The main plot's axes are removed from it's
|
|
layout and placed in the surrounding exterior layouts to allow for
|
|
re-ordering if desired.
|
|
|
|
'''
|
|
def __init__(
|
|
self,
|
|
item: PlotItem,
|
|
grid: QGraphicsGridLayout,
|
|
reverse: bool = False, # insert items to the "center"
|
|
|
|
) -> None:
|
|
self.items: list[PlotItem] = []
|
|
# self.grid = grid
|
|
self.reverse = reverse
|
|
|
|
# TODO: use a ``bidict`` here?
|
|
self._pi2axes: dict[
|
|
int,
|
|
dict[str, AxisItem],
|
|
] = {}
|
|
|
|
self._axes2pi: dict[
|
|
AxisItem,
|
|
dict[str, PlotItem],
|
|
] = {}
|
|
|
|
# TODO: better name?
|
|
# construct surrounding layouts for placing outer axes and
|
|
# their legends and title labels.
|
|
self.sides: dict[
|
|
str,
|
|
tuple[QGraphicsLinearLayout, list[AxisItem]]
|
|
] = {}
|
|
|
|
for name, pos in _axes_layout_indices.items():
|
|
layout = QGraphicsLinearLayout()
|
|
self.sides[name] = (layout, [])
|
|
|
|
layout.setContentsMargins(0, 0, 0, 0)
|
|
layout.setSpacing(0)
|
|
|
|
if name in ('top', 'bottom'):
|
|
orient = Qt.Vertical
|
|
elif name in ('left', 'right'):
|
|
orient = Qt.Horizontal
|
|
|
|
layout.setOrientation(orient)
|
|
|
|
self.insert(0, item)
|
|
|
|
# insert surrounding linear layouts into the parent pi's layout
|
|
# such that additional axes can be appended arbitrarily without
|
|
# having to expand or resize the parent's grid layout.
|
|
for name, (linlayout, axes) in self.sides.items():
|
|
|
|
# TODO: do we need this?
|
|
# axis should have been removed during insert above
|
|
index = _axes_layout_indices[name]
|
|
axis = item.layout.itemAt(*index)
|
|
if axis and axis.isVisible():
|
|
assert linlayout.itemAt(0) is axis
|
|
|
|
# item.layout.removeItem(axis)
|
|
item.layout.addItem(linlayout, *index)
|
|
layout = item.layout.itemAt(*index)
|
|
assert layout is linlayout
|
|
|
|
def _register_item(
|
|
self,
|
|
index: int,
|
|
plotitem: PlotItem,
|
|
|
|
) -> None:
|
|
for name, axis_info in plotitem.axes.items():
|
|
axis = axis_info['item']
|
|
# register this plot's (maybe re-placed) axes for lookup.
|
|
self._pi2axes.setdefault(index, {})[name] = axis
|
|
self._axes2pi.setdefault(index, {})[name] = plotitem
|
|
|
|
# enter plot into list for index tracking
|
|
self.items.insert(index, plotitem)
|
|
|
|
def insert(
|
|
self,
|
|
index: int,
|
|
plotitem: PlotItem,
|
|
|
|
) -> (int, int):
|
|
'''
|
|
Place item at index by inserting all axes into the grid
|
|
at list-order appropriate position.
|
|
|
|
'''
|
|
if index < 0:
|
|
raise ValueError('`insert()` only supports an index >= 0')
|
|
|
|
# add plot's axes in sequence to the embedded linear layouts
|
|
# for each "side" thus avoiding graphics collisions.
|
|
for name, axis_info in plotitem.axes.copy().items():
|
|
linlayout, axes = self.sides[name]
|
|
axis = axis_info['item']
|
|
|
|
if axis in axes:
|
|
# TODO: re-order using ``.pop()`` ?
|
|
ValueError(f'{axis} is already in {name} layout!?')
|
|
|
|
# linking sanity
|
|
axis_view = axis.linkedView()
|
|
assert axis_view is plotitem.vb
|
|
|
|
if (
|
|
not axis.isVisible()
|
|
|
|
# XXX: we never skip moving the axes for the *first*
|
|
# plotitem inserted (even if not shown) since we need to
|
|
# move all the hidden axes into linear sub-layouts for
|
|
# that "central" plot in the overlay. Also if we don't
|
|
# do it there's weird geomoetry calc offsets that make
|
|
# view coords slightly off somehow .. smh
|
|
and not len(self.items) == 0
|
|
):
|
|
continue
|
|
|
|
# XXX: Remove old axis? No, turns out we don't need this?
|
|
# DON'T unlink it since we the original ``ViewBox``
|
|
# to still drive it B)
|
|
# popped = plotitem.removeAxis(name, unlink=False)
|
|
# assert axis is popped
|
|
|
|
# invert insert index for layouts which are
|
|
# not-left-to-right, top-to-bottom insert oriented
|
|
if name in ('top', 'left'):
|
|
index = min(len(axes) - index, 0)
|
|
assert index >= 0
|
|
|
|
linlayout.insertItem(index, axis)
|
|
axes.insert(index, axis)
|
|
|
|
self._register_item(index, plotitem)
|
|
|
|
return index
|
|
|
|
def append(
|
|
self,
|
|
item: PlotItem,
|
|
|
|
) -> (int, int):
|
|
'''
|
|
Append item's axes at indexes which put its axes "outside"
|
|
previously overlayed entries.
|
|
|
|
'''
|
|
# for left and bottom axes we have to first remove
|
|
# items and re-insert to maintain a list-order.
|
|
return self.insert(len(self.items), item)
|
|
|
|
def get_axis(
|
|
self,
|
|
plot: PlotItem,
|
|
name: str,
|
|
|
|
) -> AxisItem:
|
|
'''
|
|
Retrieve the named axis for overlayed ``plot``.
|
|
|
|
'''
|
|
index = self.items.index(plot)
|
|
return self._pi2axes[index][name]
|
|
|
|
def pop(
|
|
self,
|
|
item: PlotItem,
|
|
|
|
) -> PlotItem:
|
|
'''
|
|
Remove item and restack all axes in list-order.
|
|
|
|
'''
|
|
raise NotImplementedError
|
|
|
|
|
|
# Unimplemented features TODO:
|
|
# - 'A' (autobtn) should relay to all views
|
|
# - context menu single handler + relay?
|
|
# - layout unwind and re-pack for 'left' and 'top' axes
|
|
# - add labels to layout if detected in source ``PlotItem``
|
|
|
|
# UX nice-to-have TODO:
|
|
# - optional "focussed" view box support for view boxes
|
|
# that have custom input handlers (eg. you might want to
|
|
# scale the view to some "focussed" data view and have overlayed
|
|
# viewboxes only respond to relayed events.)
|
|
# - figure out how to deal with menu raise events for multi-viewboxes.
|
|
# (we might want to add a different menu which specs the name of the
|
|
# view box currently being handled?
|
|
# - allow selection of a particular view box by interacting with its
|
|
# axis?
|
|
|
|
|
|
# TODO: we might want to enabled some kind of manual flag to disable
|
|
# this method wrapping during type creation? As example a user could
|
|
# definitively decide **not** to enable broadcasting support by
|
|
# setting something like ``ViewBox.disable_relays = True``?
|
|
def mk_relay_method(
|
|
|
|
signame: str,
|
|
slot: Callable[
|
|
[ViewBox,
|
|
'QEvent',
|
|
Optional[AxisItem]],
|
|
None,
|
|
],
|
|
|
|
) -> Callable[
|
|
[
|
|
ViewBox,
|
|
# lol, there isn't really a generic type thanks
|
|
# to the rewrite of Qt's event system XD
|
|
'QEvent',
|
|
|
|
'Optional[AxisItem]',
|
|
'Optional[ViewBox]', # the ``relayed_from`` arg we provide
|
|
],
|
|
None,
|
|
]:
|
|
|
|
def maybe_broadcast(
|
|
vb: 'ViewBox',
|
|
ev: 'QEvent',
|
|
axis: 'Optional[int]' = None,
|
|
relayed_from: 'ViewBox' = None,
|
|
|
|
) -> None:
|
|
'''
|
|
(soon to be) Decorator which makes an event handler
|
|
"broadcastable" to overlayed ``GraphicsWidget``s.
|
|
|
|
Adds relay signals based on the decorated handler's name
|
|
and conducts a signal broadcast of the relay signal if there
|
|
are consumers registered.
|
|
|
|
'''
|
|
# When no relay source has been set just bypass all
|
|
# the broadcast machinery.
|
|
if vb.event_relay_source is None:
|
|
ev.accept()
|
|
return slot(
|
|
vb,
|
|
ev,
|
|
axis=axis,
|
|
)
|
|
|
|
if relayed_from:
|
|
assert axis is None
|
|
|
|
# this is a relayed event and should be ignored (so it does not
|
|
# halt/short circuit the graphicscene loop). Further the
|
|
# surrounding handler for this signal must be allowed to execute
|
|
# and get processed by **this consumer**.
|
|
print(f'{vb.name} rx relayed from {relayed_from.name}')
|
|
ev.ignore()
|
|
|
|
return slot(
|
|
vb,
|
|
ev,
|
|
axis=axis,
|
|
)
|
|
|
|
if axis is not None:
|
|
print(f'{vb.name} handling axis event:\n{str(ev)}')
|
|
ev.accept()
|
|
return slot(
|
|
vb,
|
|
ev,
|
|
axis=axis,
|
|
)
|
|
|
|
elif (
|
|
relayed_from is None
|
|
and vb.event_relay_source is vb # we are the broadcaster
|
|
and axis is None
|
|
):
|
|
# Broadcast case: this is a source event which will be
|
|
# relayed to attached consumers and accepted after all
|
|
# consumers complete their own handling followed by this
|
|
# routine's processing. Sequence is,
|
|
# - pre-relay to all consumers *first* - ``.emit()`` blocks
|
|
# until all downstream relay handlers have run.
|
|
# - run the source handler for **this** event and accept
|
|
# the event
|
|
|
|
# Access the "bound signal" that is created
|
|
# on the widget type as part of instantiation.
|
|
signal = getattr(vb, signame)
|
|
# print(f'{vb.name} emitting {signame}')
|
|
|
|
# TODO/NOTE: we could also just bypass a "relay" signal
|
|
# entirely and instead call the handlers manually in
|
|
# a loop? This probably is a lot simpler and also doesn't
|
|
# have any downside, and allows not touching target widget
|
|
# internals.
|
|
signal.emit(
|
|
ev,
|
|
axis,
|
|
# passing this demarks a broadcasted/relayed event
|
|
vb,
|
|
)
|
|
# accept event so no more relays are fired.
|
|
ev.accept()
|
|
|
|
# call underlying wrapped method with an extra
|
|
# ``relayed_from`` value to denote that this is a relayed
|
|
# event handling case.
|
|
return slot(
|
|
vb,
|
|
ev,
|
|
axis=axis,
|
|
)
|
|
|
|
return maybe_broadcast
|
|
|
|
|
|
# XXX: :( can't define signals **after** class compile time
|
|
# so this is not really useful.
|
|
# def mk_relay_signal(
|
|
# func,
|
|
# name: str = None,
|
|
|
|
# ) -> Signal:
|
|
# (
|
|
# args,
|
|
# varargs,
|
|
# varkw,
|
|
# defaults,
|
|
# kwonlyargs,
|
|
# kwonlydefaults,
|
|
# annotations
|
|
# ) = inspect.getfullargspec(func)
|
|
|
|
# # XXX: generate a relay signal with 1 extra
|
|
# # argument for a ``relayed_from`` kwarg. Since
|
|
# # ``'self'`` is already ignored by signals we just need
|
|
# # to count the arguments since we're adding only 1 (and
|
|
# # ``args`` will capture that).
|
|
# numargs = len(args + list(defaults))
|
|
# signal = Signal(*tuple(numargs * [object]))
|
|
# signame = name or func.__name__ + 'Relay'
|
|
# return signame, signal
|
|
|
|
|
|
def enable_relays(
|
|
widget: GraphicsWidget,
|
|
handler_names: list[str],
|
|
|
|
) -> list[Signal]:
|
|
'''
|
|
Method override helper which enables relay of a particular
|
|
``Signal`` from some chosen broadcaster widget to a set of
|
|
consumer widgets which should operate their event handlers normally
|
|
but instead of signals "relayed" from the broadcaster.
|
|
|
|
Mostly useful for overlaying widgets that handle user input
|
|
that you want to overlay graphically. The target ``widget`` type must
|
|
define ``QtCore.Signal``s each with a `'Relay'` suffix for each
|
|
name provided in ``handler_names: list[str]``.
|
|
|
|
'''
|
|
signals = []
|
|
for name in handler_names:
|
|
handler = getattr(widget, name)
|
|
signame = name + 'Relay'
|
|
# ensure the target widget defines a relay signal
|
|
relay = getattr(widget, signame)
|
|
widget.relays[signame] = name
|
|
signals.append(relay)
|
|
method = mk_relay_method(signame, handler)
|
|
setattr(widget, name, method)
|
|
|
|
return signals
|
|
|
|
|
|
enable_relays(
|
|
ChartView,
|
|
['wheelEvent', 'mouseDragEvent']
|
|
)
|
|
|
|
|
|
class PlotItemOverlay:
|
|
'''
|
|
A composite for managing overlaid ``PlotItem`` instances such that
|
|
you can make multiple graphics appear on the same graph with
|
|
separate (non-colliding) axes apply ``ViewBox`` signal broadcasting
|
|
such that all overlaid items respond to input simultaneously.
|
|
|
|
'''
|
|
def __init__(
|
|
self,
|
|
root_plotitem: PlotItem
|
|
|
|
) -> None:
|
|
|
|
self.root_plotitem: PlotItem = root_plotitem
|
|
|
|
vb = root_plotitem.vb
|
|
vb.event_relay_source = vb # TODO: maybe change name?
|
|
vb.setZValue(1000) # XXX: critical for scene layering/relaying
|
|
|
|
self.overlays: list[PlotItem] = []
|
|
from piker.ui._overlay import ComposedGridLayout
|
|
self.layout = ComposedGridLayout(
|
|
root_plotitem,
|
|
root_plotitem.layout,
|
|
)
|
|
self._relays: dict[str, Signal] = {}
|
|
|
|
def add_plotitem(
|
|
self,
|
|
plotitem: PlotItem,
|
|
index: Optional[int] = None,
|
|
|
|
# TODO: we could also put the ``ViewBox.XAxis``
|
|
# style enum here?
|
|
# (0,), # link x
|
|
# (1,), # link y
|
|
# (0, 1), # link both
|
|
link_axes: tuple[int] = (),
|
|
|
|
) -> None:
|
|
|
|
index = index or 0
|
|
root = self.root_plotitem
|
|
# layout: QGraphicsGridLayout = root.layout
|
|
self.overlays.insert(index, plotitem)
|
|
vb: ViewBox = plotitem.vb
|
|
|
|
# mark this consumer overlay as ready to expect relayed events
|
|
# from the root plotitem.
|
|
vb.event_relay_source = root.vb
|
|
|
|
# TODO: some sane way to allow menu event broadcast XD
|
|
# vb.setMenuEnabled(False)
|
|
|
|
# TODO: inside the `maybe_broadcast()` (soon to be) decorator
|
|
# we need have checks that consumers have been attached to
|
|
# these relay signals.
|
|
if link_axes != (0, 1):
|
|
|
|
# wire up relay signals
|
|
for relay_signal_name, handler_name in vb.relays.items():
|
|
# print(handler_name)
|
|
# XXX: Signal class attrs are bound after instantiation
|
|
# of the defining type, so we need to access that bound
|
|
# version here.
|
|
signal = getattr(root.vb, relay_signal_name)
|
|
handler = getattr(vb, handler_name)
|
|
signal.connect(handler)
|
|
|
|
# link dim-axes to root if requested by user.
|
|
# TODO: solve more-then-wanted scaled panning on click drag
|
|
# which seems to be due to broadcast. So we probably need to
|
|
# disable broadcast when axes are linked in a particular
|
|
# dimension?
|
|
for dim in link_axes:
|
|
# link x and y axes to new view box such that the top level
|
|
# viewbox propagates to the root (and whatever other
|
|
# plotitem overlays that have been added).
|
|
vb.linkView(dim, root.vb)
|
|
|
|
# make overlaid viewbox impossible to focus since the top
|
|
# level should handle all input and relay to overlays.
|
|
# NOTE: this was solved with the `setZValue()` above!
|
|
|
|
# TODO: we will probably want to add a "focus" api such that
|
|
# a new "top level" ``PlotItem`` can be selected dynamically
|
|
# (and presumably the axes dynamically sorted to match).
|
|
vb.setFlag(
|
|
vb.GraphicsItemFlag.ItemIsFocusable,
|
|
False
|
|
)
|
|
vb.setFocusPolicy(Qt.NoFocus)
|
|
|
|
# append-compose into the layout all axes from this plot
|
|
self.layout.insert(index, plotitem)
|
|
|
|
plotitem.setGeometry(root.vb.sceneBoundingRect())
|
|
|
|
def size_to_viewbox(vb: 'ViewBox'):
|
|
plotitem.setGeometry(vb.sceneBoundingRect())
|
|
|
|
root.vb.sigResized.connect(size_to_viewbox)
|
|
|
|
# ensure the overlayed view is redrawn on each cycle
|
|
root.scene().sigPrepareForPaint.connect(vb.prepareForPaint)
|
|
|
|
# focus state sanity
|
|
vb.clearFocus()
|
|
assert not vb.focusWidget()
|
|
root.vb.setFocus()
|
|
assert root.vb.focusWidget()
|
|
|
|
# XXX: do we need this? Why would you build then destroy?
|
|
def remove_plotitem(self, plotItem: PlotItem) -> None:
|
|
'''
|
|
Remove this ``PlotItem`` from the overlayed set making not shown
|
|
and unable to accept input.
|
|
|
|
'''
|
|
...
|
|
|
|
# TODO: i think this would be super hot B)
|
|
def focus_item(self, plotitem: PlotItem) -> PlotItem:
|
|
'''
|
|
Apply focus to a contained PlotItem thus making it the "top level"
|
|
item in the overlay able to accept peripheral's input from the user
|
|
and responsible for zoom and panning control via its ``ViewBox``.
|
|
|
|
'''
|
|
...
|
|
|
|
def get_axis(
|
|
self,
|
|
plot: PlotItem,
|
|
name: str,
|
|
|
|
) -> AxisItem:
|
|
'''
|
|
Retrieve the named axis for overlayed ``plot``.
|
|
|
|
'''
|
|
return self.layout.get_axis(plot, name)
|
|
|
|
# TODO: i guess we need this if you want to detach existing plots
|
|
# dynamically? XXX: untested as of now.
|
|
def _disconnect_all(
|
|
self,
|
|
plotitem: PlotItem,
|
|
) -> list[Signal]:
|
|
'''
|
|
Disconnects all signals related to this widget for the given chart.
|
|
|
|
'''
|
|
disconnected = []
|
|
for pi, sig in self._relays.items():
|
|
QObject.disconnect(sig)
|
|
disconnected.append(sig)
|
|
|
|
return disconnected
|