piker/piker/ui/_window.py

678 lines
20 KiB
Python
Raw Normal View History

# 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/>.
"""
Qt main window singletons and stuff.
"""
from __future__ import annotations
import os
import signal
import time
from typing import (
Callable,
Union,
)
import uuid
from piker.ui.qt import (
Qt,
QtCore,
QWidget,
QMainWindow,
QApplication,
QLabel,
QStatusBar,
QScreen,
QCloseEvent,
QSettings,
QEvent,
QObject,
)
from ..log import get_logger
from . import _style
from ._style import (
_font_small,
hcolor,
)
from ._widget import GodWidget
log = get_logger(__name__)
class GlobalZoomEventFilter(QObject):
'''
Application-level event filter for global UI zoom shortcuts.
This filter intercepts keyboard events BEFORE they reach widgets,
allowing us to implement global UI zoom shortcuts that take precedence
over widget-specific shortcuts.
Shortcuts:
- Ctrl+Shift+Plus/Equal: Zoom in
- Ctrl+Shift+Minus: Zoom out
- Ctrl+Shift+0: Reset zoom
'''
def __init__(self, main_window: MainWindow):
super().__init__()
self.main_window = main_window
def eventFilter(self, obj: QObject, event: QEvent) -> bool:
'''
Filter keyboard events for global zoom shortcuts.
Returns True to filter out (consume) the event, False to pass through.
'''
if event.type() == QEvent.Type.KeyPress:
key = event.key()
mods = event.modifiers()
# Mask out the KeypadModifier which Qt sometimes adds
mods = mods & ~Qt.KeyboardModifier.KeypadModifier
# Check if we have Ctrl+Shift (both required)
has_ctrl = bool(
mods
&
Qt.KeyboardModifier.ControlModifier
)
_has_shift = bool(
mods
&
Qt.KeyboardModifier.ShiftModifier
)
# Only handle UI zoom if BOTH Ctrl and Shift are pressed
# For Plus key: user presses Cmd+Shift+Equal (which makes Plus)
# For Minus key: user presses Cmd+Shift+Minus
if (
has_ctrl
# and
# has_shift
):
# Zoom in: Ctrl+Shift+Plus
# Note: Plus key usually comes as Key_Equal with Shift modifier
if key in (
Qt.Key.Key_Plus,
Qt.Key.Key_Equal,
):
self.main_window.zoom_in()
return True # consume event
# Zoom out: Ctrl+Shift+Minus
2026-03-06 06:03:29 +00:00
# Note: On some keyboards Shift+Minus produces '_' (Underscore)
elif key in (
Qt.Key.Key_Minus,
Qt.Key.Key_Underscore,
):
self.main_window.zoom_out()
return True # consume event
# Reset zoom: Ctrl+Shift+0
2026-03-06 06:03:29 +00:00
# Note: On some keyboards Shift+0 produces ')' (ParenRight)
elif key in (
Qt.Key.Key_0,
Qt.Key.Key_ParenRight,
):
self.main_window.reset_zoom()
return True # consume event
# Pass through if only Ctrl (no Shift) - this goes to chart zoom
# Pass through all other events too
return False
return False
class MultiStatus:
bar: QStatusBar
statuses: list[str]
def __init__(self, bar, statuses) -> None:
self.bar = bar
self.statuses = statuses
self._to_clear: set = set()
self._status_groups: dict[str, (set, Callable)] = {}
def open_status(
self,
msg: str,
final_msg: str|None = None,
clear_on_next: bool = False,
group_key: Union[bool, str]|None = False,
) -> Union[Callable[..., None], str]:
'''
Add a status to the status bar and return a close callback which
when called will remove the status ``msg``.
'''
2021-06-18 13:37:55 +00:00
for old_msg in self._to_clear:
try:
2021-06-18 13:37:55 +00:00
self.statuses.remove(old_msg)
except ValueError:
pass
self.statuses.append(msg)
def remove_msg() -> None:
try:
self.statuses.remove(msg)
except ValueError:
pass
2021-06-18 13:37:55 +00:00
self.render()
if final_msg is not None:
self.statuses.append(final_msg)
self.render()
self._to_clear.add(final_msg)
ret = remove_msg
# create a "status group" such that new `.open_status()`
# calls can be made passing in the returned group key.
# once all clear callbacks have been called from all statuses
# in the group the final status msg to be removed will be the one
# the one provided when `group_key=True`, this way you can
# create a long living status that completes once all
# sub-statuses have finished.
if group_key is True:
if clear_on_next:
ValueError("Can't create group status and clear it on next?")
# generate a key for a new "status group"
new_group_key = str(uuid.uuid4())
def pop_group_and_clear():
subs, final_clear = self._status_groups.pop(new_group_key)
assert not subs
return remove_msg()
self._status_groups[new_group_key] = (set(), pop_group_and_clear)
ret = new_group_key
elif group_key:
def pop_from_group_and_maybe_clear_group():
# remove the message for this sub-status
remove_msg()
# check to see if all other substatuses have cleared
group_tup = self._status_groups.get(group_key)
if group_tup:
subs, group_clear = group_tup
try:
subs.remove(msg)
except KeyError:
raise KeyError(f'no msg {msg} for group {group_key}!?')
if not subs:
group_clear()
group = self._status_groups.get(group_key)
if group:
group[0].add(msg)
ret = pop_from_group_and_maybe_clear_group
2021-06-18 13:37:55 +00:00
self.render()
if clear_on_next:
self._to_clear.add(msg)
return ret
def render(self) -> None:
'''
Display all open statuses to bar.
2021-06-18 13:37:55 +00:00
'''
if self.statuses:
self.bar.showMessage(f'{" ".join(self.statuses)}')
else:
self.bar.clearMessage()
class MainWindow(QMainWindow):
# XXX: for tiling wms this should scale
# with the alloted window size.
# TODO: detect for tiling and if untrue set some size?
# size = (300, 500)
godwidget: GodWidget
title = 'piker chart (ur symbol is loading bby)'
def __init__(self, parent=None):
super().__init__(parent)
# self.setMinimumSize(*self.size)
self.setWindowTitle(self.title)
# set by runtime after `trio` is engaged.
self.godwidget: GodWidget|None = None
self._status_bar: QStatusBar = None
self._status_label: QLabel = None
self._size: tuple[int, int]|None = None
2022-02-10 19:35:28 +00:00
# restore window geometry from previous session
settings = QSettings('pikers', 'piker')
geometry = settings.value('windowGeometry')
if geometry is not None:
self.restoreGeometry(geometry)
log.debug('Restored window geometry from previous session')
# zoom level for UI scaling (1.0 = 100%, 1.5 = 150%, etc)
# Change this value to set the default startup zoom level
2026-03-06 06:03:29 +00:00
self._zoom_level: float = 1.0 # Start at 100% (normal)
self._min_zoom: float = 0.5
2026-03-06 06:03:29 +00:00
self._max_zoom: float = 3.0 # Reduced from 10.0 to prevent extreme cropping
self._zoom_step: float = 0.2 # 20% per keypress
# event filter for global zoom shortcuts
self._zoom_filter: GlobalZoomEventFilter | None = None
def install_global_zoom_filter(self) -> None:
'''Install application-level event filter for global UI zoom shortcuts.'''
if self._zoom_filter is None:
self._zoom_filter = GlobalZoomEventFilter(self)
app = QApplication.instance()
app.installEventFilter(self._zoom_filter)
log.info('Installed global zoom shortcuts: Ctrl+Shift+Plus/Minus/0')
@property
def mode_label(self) -> QLabel:
# init mode label
if not self._status_label:
self._status_label = label = QLabel()
label.setStyleSheet(
2021-07-27 10:12:19 +00:00
f"""QLabel {{
color : {hcolor('gunmetal')};
}}
"""
# font-size : {font_size}px;
)
label.setTextFormat(
Qt.TextFormat.MarkdownText
)
label.setFont(_font_small.font)
label.setMargin(2)
label.setAlignment(
QtCore.Qt.AlignVCenter
|QtCore.Qt.AlignRight
)
self.statusBar().addPermanentWidget(label)
label.show()
return self._status_label
def closeEvent(
self,
event: QCloseEvent,
2021-08-03 13:45:33 +00:00
) -> None:
'''Cancel the root actor asap.
2021-07-24 20:06:05 +00:00
2021-08-03 13:45:33 +00:00
'''
# save window geometry for next session
settings = QSettings('pikers', 'piker')
settings.setValue('windowGeometry', self.saveGeometry())
log.debug('Saved window geometry for next session')
# raising KBI seems to get intercepted by by Qt so just use the system.
os.kill(os.getpid(), signal.SIGINT)
@property
def status_bar(self) -> QStatusBar:
# style and cached the status bar on first access
if not self._status_bar:
sb = self.statusBar()
sb.setStyleSheet((
f"color : {hcolor('gunmetal')};"
f"background : {hcolor('default_dark')};"
f"font-size : {_font_small.px_size}px;"
"padding : 0px;"
# "min-height : 19px;"
# "qproperty-alignment: AlignVCenter;"
))
self.setStatusBar(sb)
self._status_bar = MultiStatus(sb, [])
return self._status_bar
2021-07-24 20:06:05 +00:00
def set_mode_name(
self,
name: str,
) -> None:
self.mode_label.setText(f'mode:{name}')
def on_focus_change(
self,
2021-07-24 20:06:05 +00:00
last: QWidget,
current: QWidget,
2021-07-24 20:06:05 +00:00
) -> None:
Pass `loglevel` down through `.ui` graphics tasks Add `loglevel` propagation to UI graphics tasks and sampler stream opens to enable proper console logging in chart update loops. This ensures the graphics and FSP subsystems receive the same loglevel as their parent and/or sibling UI-actor tasks. Deats, - add `loglevel` param to `graphics_update_loop()` and `increment_history_view()` with default `'warning'`. - pass `loglevel` to `open_sample_stream()` calls in both fns. - use `partial()` to pass `loglevel` through to `nurse.start_soon()` calls in `display_symbol_data()` and `graphics_update_loop()`. Also logging, doc-strs, and code-style tweaks, - change `print()` -> `log.debug()` for hidden-chart and interaction-pause msgs in graphics loop. - change `log.info()` -> `log.debug()` for resize events in `GodWidget` and `MainWindow`. - add multiline style to resize log msg in `GodWidget`. - add docstring to `MainWindow.on_focus_change()`. - moar union type annot adjustments. - switch to explicit kwarg `period_s=` for `open_sample_stream()` in `increment_history_view()`. - multiline style for `names` list in `open_fsp_actor_cluster()`. - change `count=2` -> `count=len(names)` in `open_fsp_actor_cluster()`. - add TODO about using `.experimental` for cluster import (once that get's patched into upstream `tractor`). - multiline style for `or` in `FspAdmin.start_engine_task()`. - comment-out unused `brokernames` in `ui.cli.chart()`. - add commented breakpoint in `ui.cli.chart()`. - fix docstring style in `OrderMode.on_submit()`. (this commit msg was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code
2026-02-13 00:24:44 +00:00
'''
Focus handler.
For now updates the "current mode" name.
Pass `loglevel` down through `.ui` graphics tasks Add `loglevel` propagation to UI graphics tasks and sampler stream opens to enable proper console logging in chart update loops. This ensures the graphics and FSP subsystems receive the same loglevel as their parent and/or sibling UI-actor tasks. Deats, - add `loglevel` param to `graphics_update_loop()` and `increment_history_view()` with default `'warning'`. - pass `loglevel` to `open_sample_stream()` calls in both fns. - use `partial()` to pass `loglevel` through to `nurse.start_soon()` calls in `display_symbol_data()` and `graphics_update_loop()`. Also logging, doc-strs, and code-style tweaks, - change `print()` -> `log.debug()` for hidden-chart and interaction-pause msgs in graphics loop. - change `log.info()` -> `log.debug()` for resize events in `GodWidget` and `MainWindow`. - add multiline style to resize log msg in `GodWidget`. - add docstring to `MainWindow.on_focus_change()`. - moar union type annot adjustments. - switch to explicit kwarg `period_s=` for `open_sample_stream()` in `increment_history_view()`. - multiline style for `names` list in `open_fsp_actor_cluster()`. - change `count=2` -> `count=len(names)` in `open_fsp_actor_cluster()`. - add TODO about using `.experimental` for cluster import (once that get's patched into upstream `tractor`). - multiline style for `or` in `FspAdmin.start_engine_task()`. - comment-out unused `brokernames` in `ui.cli.chart()`. - add commented breakpoint in `ui.cli.chart()`. - fix docstring style in `OrderMode.on_submit()`. (this commit msg was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code
2026-02-13 00:24:44 +00:00
'''
log.debug(
f'widget focus changed from,\n'
f'{last} -> {current}'
)
2021-08-03 13:45:33 +00:00
if current is not None:
# cursor left window?
2021-08-03 13:45:33 +00:00
name = getattr(current, 'mode_name', '')
2021-07-24 20:06:05 +00:00
self.set_mode_name(name)
def current_screen(self) -> QScreen:
'''
Get a frickin screen (if we can, gawd).
'''
app = QApplication.instance()
for _ in range(3):
screen = app.screenAt(self.pos())
log.debug('trying to access QScreen...')
if screen is None:
time.sleep(0.5)
continue
break
else:
if screen is None:
# try for the first one we can find
screen = app.screens()[0]
assert screen, "Wow Qt is dumb as shit and has no screen..."
return screen
def configure_to_desktop(
self,
size: tuple[int, int]|None = None,
2022-02-10 19:20:15 +00:00
) -> None:
2022-02-10 19:20:15 +00:00
'''
Explicitly size the window dimensions (for stacked window
managers).
For tina systems (like windoze) try to do a sane window size on
startup.
'''
# https://stackoverflow.com/a/18975846
if not size and not self._size:
# app = QApplication.instance()
geo = self.current_screen().geometry()
h, w = geo.height(), geo.width()
# use approx 1/3 of the area of the screen by default
self._size = round(w * .666), round(h * .666)
2022-02-10 19:20:15 +00:00
self.resize(*size or self._size)
def resizeEvent(self, event: QtCore.QEvent) -> None:
if (
# event.spontaneous()
event.oldSize().height == event.size().height
):
event.ignore()
return
# XXX: uncomment for debugging..
# attrs = {}
# for key in dir(event):
# if key == '__dir__':
# continue
# attr = getattr(event, key)
# try:
# attrs[key] = attr()
# except TypeError:
# attrs[key] = attr
# from pprint import pformat
# print(
# f'{pformat(attrs)}\n'
# f'WINDOW RESIZE: {self.size()}\n\n'
# )
self.godwidget.on_win_resize(event)
event.accept()
def zoom_in(self) -> None:
'''
Increase overall UI-widgets zoom level by scaling it the
global font sizes.
'''
new_zoom: float = min(
self._zoom_level + self._zoom_step,
self._max_zoom,
)
if new_zoom != self._zoom_level:
self._zoom_level = new_zoom
font_size: int = self._apply_zoom()
log.info(
f'Zoomed in UI\n'
f'zoom_step: {self._zoom_step!r}\n'
f'zoom_level(%): {self._zoom_level:.1%}\n'
f'font_size: {font_size!r}'
)
def zoom_out(self) -> float:
'''
Decrease UI zoom level.
'''
new_zoom: float = max(self._zoom_level - self._zoom_step, self._min_zoom)
if new_zoom != self._zoom_level:
self._zoom_level = new_zoom
font_size: int = self._apply_zoom()
log.info(
f'Zoomed out UI\n'
f'zoom_step: {self._zoom_step!r}\n'
f'zoom_level(%): {self._zoom_level:.1%}\n'
f'font_size: {font_size!r}'
)
return new_zoom
def reset_zoom(self) -> None:
'''
Reset UI zoom to 100%.
'''
if self._zoom_level != 1.0:
self._zoom_level = 1.0
font_size: int = self._apply_zoom()
log.info(
f'Reset zoom level\n'
f'zoom_step: {self._zoom_step!r}\n'
f'zoom_level(%): {self._zoom_level:.1%}\n'
f'font_size: {font_size!r}'
)
return self._zoom_level
def _apply_zoom(self) -> int:
'''
Apply current zoom level to all UI elements.
'''
# reconfigure fonts with zoom multiplier
font_size: int = _style._config_fonts_to_screen(
zoom_level=self._zoom_level
)
# update status bar styling with new font size
if self._status_bar:
sb = self.statusBar()
sb.setStyleSheet((
f"color : {hcolor('gunmetal')};"
f"background : {hcolor('default_dark')};"
f"font-size : {_style._font_small.px_size}px;"
"padding : 0px;"
))
# force update of mode label if it exists
if self._status_label:
self._status_label.setFont(_style._font_small.font)
# update godwidget and its children
if self.godwidget:
# update search widget if it exists
if hasattr(self.godwidget, 'search') and self.godwidget.search:
self.godwidget.search.update_fonts()
# update order mode panes in all chart views
self._update_chart_order_panes()
# recursively update all other widgets with stylesheets
self._refresh_widget_fonts(self.godwidget)
self.godwidget.update()
return font_size
def _update_chart_order_panes(self) -> None:
'''
Update order entry panels in all charts.
'''
if not self.godwidget:
return
# iterate through all linked splits (hist and rt)
for splits_name in ['hist_linked', 'rt_linked']:
splits = getattr(self.godwidget, splits_name, None)
if not splits:
continue
# get main chart
chart = getattr(splits, 'chart', None)
if chart:
# update axes
self._update_chart_axes(chart)
# update order pane
if hasattr(chart, 'view'):
view = chart.view
if hasattr(view, 'order_mode') and view.order_mode:
order_mode = view.order_mode
if hasattr(order_mode, 'pane') and order_mode.pane:
order_mode.pane.update_fonts()
# also check subplots
subplots = getattr(splits, 'subplots', {})
for name, subplot_chart in subplots.items():
# update subplot axes
self._update_chart_axes(subplot_chart)
# update subplot order pane
if hasattr(subplot_chart, 'view'):
subplot_view = subplot_chart.view
if hasattr(subplot_view, 'order_mode') and subplot_view.order_mode:
subplot_order_mode = subplot_view.order_mode
if hasattr(subplot_order_mode, 'pane') and subplot_order_mode.pane:
subplot_order_mode.pane.update_fonts()
# resize all sidepanes to match main chart's sidepane width
# this ensures volume/subplot sidepanes match the main chart
if splits and hasattr(splits, 'resize_sidepanes'):
splits.resize_sidepanes()
def _update_chart_axes(self, chart) -> None:
'''Update axis fonts and sizing for a chart.'''
from . import _style
# update price axis (right side)
if hasattr(chart, 'pi') and chart.pi:
plot_item = chart.pi
# get all axes from plot item
for axis_name in ['left', 'right', 'bottom', 'top']:
axis = plot_item.getAxis(axis_name)
if axis and hasattr(axis, 'update_fonts'):
axis.update_fonts(_style._font)
# force plot item to recalculate its entire layout
plot_item.updateGeometry()
# force chart widget to update
if hasattr(chart, 'updateGeometry'):
chart.updateGeometry()
# trigger a full scene update
if hasattr(chart, 'update'):
chart.update()
def _refresh_widget_fonts(self, widget: QWidget) -> None:
'''
Recursively update font sizes in all child widgets.
This handles widgets that have font-size hardcoded in their stylesheets.
'''
from . import _style
# recursively process all children
for child in widget.findChildren(QWidget):
# skip widgets that have their own update_fonts method (handled separately)
if hasattr(child, 'update_fonts'):
continue
# update child's stylesheet if it has font-size
child_stylesheet = child.styleSheet()
if child_stylesheet and 'font-size' in child_stylesheet:
# for labels and simple widgets, regenerate stylesheet
# this is a heuristic - may need refinement
try:
child.setFont(_style._font.font)
except (AttributeError, RuntimeError):
pass
# update child's font
try:
child.setFont(_style._font.font)
except (AttributeError, RuntimeError):
pass
# singleton app per actor
_qt_win: QMainWindow = None
def main_window() -> MainWindow:
'Return the actor-global Qt window.'
global _qt_win
assert _qt_win
return _qt_win