pikers
/
piker
4
1
Fork 0
Code Issues 14 Pull Requests 13 Packages Projects Releases Wiki Activity

fix_tractor_logging: porting to latest tractor.log API(s) #74

Merged
goodboy merged 19 commits from fix_tractor_logging into hist_backfill_fixes 2026-02-23 18:35:44 +00:00
Conversation 1 Commits 19 Files Changed 51
51 changed files with 727 additions and 293 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
piker/accounting/__init__.py
View File

@ -19,8 +19,10 @@
for tendiez. for tendiez.
''' '''
from ..log import get_logger from piker.log import (
get_console_log,
get_logger,
)
from .calc import ( from .calc import (
iter_by_dt, iter_by_dt,
) )
@ -51,7 +53,17 @@ from ._allocate import (
log = get_logger(__name__) log = get_logger(__name__)
# ?TODO, enable console on import
# [ ] necessary? or `open_brokerd_dialog()` doing it is sufficient?
#
# bc might as well enable whenev imported by
# other sub-sys code (namely `.clearing`).
get_console_log(
level='warning',
name=__name__,
)
# TODO, the `as <samename>` style?
__all__ = [ __all__ = [
'Account', 'Account',
'Allocator', 'Allocator',

8
piker/accounting/_pos.py
View File

@ -60,12 +60,16 @@ from ..clearing._messages import (
BrokerdPosition, BrokerdPosition,
) )
from piker.types import Struct from piker.types import Struct
from piker.log import get_logger from piker.log import (
get_logger,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from piker.data._symcache import SymbologyCache from piker.data._symcache import SymbologyCache
log = get_logger(__name__) log = get_logger(
name=__name__,
)
class Position(Struct): class Position(Struct):

12
piker/accounting/cli.py
View File

@ -21,7 +21,6 @@ CLI front end for trades ledger and position tracking management.
from __future__ import annotations from __future__ import annotations
from pprint import pformat from pprint import pformat
from rich.console import Console from rich.console import Console
from rich.markdown import Markdown from rich.markdown import Markdown
import polars as pl import polars as pl
@ -29,7 +28,10 @@ import tractor
import trio import trio
import typer import typer
from ..log import get_logger from piker.log import (
get_console_log,
get_logger,
)
from ..service import ( from ..service import (
open_piker_runtime, open_piker_runtime,
) )
@ -45,6 +47,7 @@ from .calc import (
open_ledger_dfs, open_ledger_dfs,
) )
log = get_logger(name=__name__)
ledger = typer.Typer() ledger = typer.Typer()
@ -79,7 +82,10 @@ def sync(
"-l", "-l",
), ),
): ):
log = get_logger(loglevel) log = get_console_log(
level=loglevel,
name=__name__,
)
console = Console() console = Console()
pair: tuple[str, str] pair: tuple[str, str]

10
piker/brokers/__init__.py
View File

@ -25,15 +25,16 @@ from types import ModuleType
from tractor.trionics import maybe_open_context from tractor.trionics import maybe_open_context
from piker.log import (
get_logger,
)
from ._util import ( from ._util import (
log,
BrokerError, BrokerError,
SymbolNotFound, SymbolNotFound,
NoData, NoData,
DataUnavailable, DataUnavailable,
DataThrottle, DataThrottle,
resproc, resproc,
get_logger,
) )
__all__: list[str] = [ __all__: list[str] = [
@ -43,7 +44,6 @@ __all__: list[str] = [
'DataUnavailable', 'DataUnavailable',
'DataThrottle', 'DataThrottle',
'resproc', 'resproc',
'get_logger',
] ]
__brokers__: list[str] = [ __brokers__: list[str] = [
@ -65,6 +65,10 @@ __brokers__: list[str] = [
# bitso # bitso
] ]
log = get_logger(
name=__name__,
)
def get_brokermod(brokername: str) -> ModuleType: def get_brokermod(brokername: str) -> ModuleType:
''' '''

33
piker/brokers/_daemon.py
View File

@ -33,12 +33,18 @@ import exceptiongroup as eg
import tractor import tractor
import trio import trio
from piker.log import (
get_logger,
get_console_log,
)
from . import _util from . import _util
from . import get_brokermod from . import get_brokermod
if TYPE_CHECKING: if TYPE_CHECKING:
from ..data import _FeedsBus from ..data import _FeedsBus
log = get_logger(name=__name__)
# `brokerd` enabled modules # `brokerd` enabled modules
# TODO: move this def to the `.data` subpkg.. # TODO: move this def to the `.data` subpkg..
# NOTE: keeping this list as small as possible is part of our caps-sec # NOTE: keeping this list as small as possible is part of our caps-sec
@ -59,7 +65,7 @@ _data_mods: str = [
async def _setup_persistent_brokerd( async def _setup_persistent_brokerd(
ctx: tractor.Context, ctx: tractor.Context,
brokername: str, brokername: str,
loglevel: str | None = None, loglevel: str|None = None,
) -> None: ) -> None:
''' '''
@ -72,13 +78,14 @@ async def _setup_persistent_brokerd(
# since all hosted daemon tasks will reference this same # since all hosted daemon tasks will reference this same
# log instance's (actor local) state and thus don't require # log instance's (actor local) state and thus don't require
# any further (level) configuration on their own B) # any further (level) configuration on their own B)
log = _util.get_console_log( actor: tractor.Actor = tractor.current_actor()
loglevel or tractor.current_actor().loglevel, tll: str = actor.loglevel
log = get_console_log(
level=loglevel or tll,
name=f'{_util.subsys}.{brokername}', name=f'{_util.subsys}.{brokername}',
with_tractor_log=bool(tll),
) )
assert log.name == _util.subsys
# set global for this actor to this new process-wide instance B)
_util.log = log
# further, set the log level on any broker broker specific # further, set the log level on any broker broker specific
# logger instance. # logger instance.
@ -97,7 +104,7 @@ async def _setup_persistent_brokerd(
# NOTE: see ep invocation details inside `.data.feed`. # NOTE: see ep invocation details inside `.data.feed`.
try: try:
async with ( async with (
tractor.trionics.collapse_eg(), # tractor.trionics.collapse_eg(),
trio.open_nursery() as service_nursery trio.open_nursery() as service_nursery
): ):
bus: _FeedsBus = feed.get_feed_bus( bus: _FeedsBus = feed.get_feed_bus(
@ -193,7 +200,6 @@ def broker_init(
async def spawn_brokerd( async def spawn_brokerd(
brokername: str, brokername: str,
loglevel: str | None = None, loglevel: str | None = None,
@ -201,8 +207,10 @@ async def spawn_brokerd(
) -> bool: ) -> bool:
from piker.service._util import log # use service mngr log log.info(
log.info(f'Spawning {brokername} broker daemon') f'Spawning broker-daemon,\n'
f'backend: {brokername!r}'
)
( (
brokermode, brokermode,
@ -249,7 +257,7 @@ async def spawn_brokerd(
async def maybe_spawn_brokerd( async def maybe_spawn_brokerd(
brokername: str, brokername: str,
loglevel: str | None = None, loglevel: str|None = None,
**pikerd_kwargs, **pikerd_kwargs,
@ -265,8 +273,7 @@ async def maybe_spawn_brokerd(
from piker.service import maybe_spawn_daemon from piker.service import maybe_spawn_daemon
async with maybe_spawn_daemon( async with maybe_spawn_daemon(
service_name=f'brokerd.{brokername}',
f'brokerd.{brokername}',
service_task_target=spawn_brokerd, service_task_target=spawn_brokerd,
spawn_args={ spawn_args={
'brokername': brokername, 'brokername': brokername,

26
piker/brokers/_util.py
View File

@ -19,15 +19,13 @@ Handy cross-broker utils.
""" """
from __future__ import annotations from __future__ import annotations
from functools import partial # from functools import partial
import json import json
import httpx import httpx
import logging import logging
from ..log import ( from piker.log import (
get_logger,
get_console_log,
colorize_json, colorize_json,
) )
subsys: str = 'piker.brokers' subsys: str = 'piker.brokers'
@ -35,12 +33,22 @@ subsys: str = 'piker.brokers'
# NOTE: level should be reset by any actor that is spawned # NOTE: level should be reset by any actor that is spawned
# as well as given a (more) explicit name/key such # as well as given a (more) explicit name/key such
# as `piker.brokers.binance` matching the subpkg. # as `piker.brokers.binance` matching the subpkg.
log = get_logger(subsys) # log = get_logger(subsys)
get_console_log = partial( # ?TODO?? we could use this approach, but we need to be able
get_console_log, # to pass multiple `name=` values so for example we can include the
name=subsys, # emissions in `.accounting._pos` and others!
) # [ ] maybe we could do the `log = get_logger()` above,
# then cycle through the list of subsys mods we depend on
# and then get all their loggers and pass them to
# `get_console_log(logger=)`??
# [ ] OR just write THIS `get_console_log()` as a hook which does
# that based on who calls it?.. i dunno
#
# get_console_log = partial(
# get_console_log,
# name=subsys,
# )
class BrokerError(Exception): class BrokerError(Exception):

14
piker/brokers/binance/broker.py
View File

@ -37,8 +37,9 @@ import trio
from piker.accounting import ( from piker.accounting import (
Asset, Asset,
) )
from piker.brokers._util import ( from piker.log import (
get_logger, get_logger,
get_console_log,
) )
from piker.data._web_bs import ( from piker.data._web_bs import (
open_autorecon_ws, open_autorecon_ws,
@ -69,7 +70,9 @@ from .venues import (
) )
from .api import Client from .api import Client
log = get_logger('piker.brokers.binance') log = get_logger(
name=__name__,
)
# Fee schedule template, mostly for paper engine fees modelling. # Fee schedule template, mostly for paper engine fees modelling.
@ -245,9 +248,16 @@ async def handle_order_requests(
@tractor.context @tractor.context
async def open_trade_dialog( async def open_trade_dialog(
ctx: tractor.Context, ctx: tractor.Context,
loglevel: str = 'warning',
) -> AsyncIterator[dict[str, Any]]: ) -> AsyncIterator[dict[str, Any]]:
# enable piker.clearing console log for *this* `brokerd` subactor
get_console_log(
level=loglevel,
name=__name__,
)
# TODO: how do we set this from the EMS such that # TODO: how do we set this from the EMS such that
# positions are loaded from the correct venue on the user # positions are loaded from the correct venue on the user
# stream at startup? (that is in an attempt to support both # stream at startup? (that is in an attempt to support both

4
piker/brokers/binance/feed.py
View File

@ -64,9 +64,9 @@ from piker.data._web_bs import (
open_autorecon_ws, open_autorecon_ws,
NoBsWs, NoBsWs,
) )
from piker.log import get_logger
from piker.brokers._util import ( from piker.brokers._util import (
DataUnavailable, DataUnavailable,
get_logger,
) )
from .api import ( from .api import (
@ -78,7 +78,7 @@ from .venues import (
get_api_eps, get_api_eps,
) )
log = get_logger('piker.brokers.binance') log = get_logger(name=__name__)
class L1(Struct): class L1(Struct):

26
piker/brokers/cli.py
View File

@ -27,14 +27,12 @@ import click
import trio import trio
import tractor import tractor
from ..cli import cli from piker.cli import cli
from .. import watchlists as wl from piker import watchlists as wl
from ..log import ( from piker.log import (
colorize_json, colorize_json,
)
from ._util import (
log,
get_console_log, get_console_log,
get_logger,
) )
from ..service import ( from ..service import (
maybe_spawn_brokerd, maybe_spawn_brokerd,
@ -45,12 +43,15 @@ from ..brokers import (
get_brokermod, get_brokermod,
data, data,
) )
DEFAULT_BROKER = 'binance'
log = get_logger(
name=__name__,
)
DEFAULT_BROKER = 'binance'
_config_dir = click.get_app_dir('piker') _config_dir = click.get_app_dir('piker')
_watchlists_data_path = os.path.join(_config_dir, 'watchlists.json') _watchlists_data_path = os.path.join(_config_dir, 'watchlists.json')
OK = '\033[92m' OK = '\033[92m'
WARNING = '\033[93m' WARNING = '\033[93m'
FAIL = '\033[91m' FAIL = '\033[91m'
@ -345,7 +346,10 @@ def contracts(ctx, loglevel, broker, symbol, ids):
''' '''
brokermod = get_brokermod(broker) brokermod = get_brokermod(broker)
get_console_log(loglevel) get_console_log(
level=loglevel,
name=__name__,
)
contracts = trio.run(partial(core.contracts, brokermod, symbol)) contracts = trio.run(partial(core.contracts, brokermod, symbol))
if not ids: if not ids:
@ -477,11 +481,12 @@ def search(
# the `piker --pdb` XD .. # the `piker --pdb` XD ..
# -[ ] pull from the parent click ctx's values..dumdum # -[ ] pull from the parent click ctx's values..dumdum
# assert pdb # assert pdb
loglevel: str = config['loglevel']
# define tractor entrypoint # define tractor entrypoint
async def main(func): async def main(func):
async with maybe_open_pikerd( async with maybe_open_pikerd(
loglevel=config['loglevel'], loglevel=loglevel,
debug_mode=pdb, debug_mode=pdb,
): ):
return await func() return await func()
@ -494,6 +499,7 @@ def search(
core.symbol_search, core.symbol_search,
brokermods, brokermods,
pattern, pattern,
loglevel=loglevel,
), ),
) )

6
piker/brokers/core.py
View File

@ -28,12 +28,14 @@ from typing import (
import trio import trio
from ._util import log from piker.log import get_logger
from . import get_brokermod from . import get_brokermod
from ..service import maybe_spawn_brokerd from ..service import maybe_spawn_brokerd
from . import open_cached_client from . import open_cached_client
from ..accounting import MktPair from ..accounting import MktPair
log = get_logger(name=__name__)
async def api(brokername: str, methname: str, **kwargs) -> dict: async def api(brokername: str, methname: str, **kwargs) -> dict:
''' '''
@ -147,6 +149,7 @@ async def search_w_brokerd(
async def symbol_search( async def symbol_search(
brokermods: list[ModuleType], brokermods: list[ModuleType],
pattern: str, pattern: str,
loglevel: str = 'warning',
**kwargs, **kwargs,
) -> dict[str, dict[str, dict[str, Any]]]: ) -> dict[str, dict[str, dict[str, Any]]]:
@ -176,6 +179,7 @@ async def symbol_search(
'_infect_asyncio', '_infect_asyncio',
False, False,
), ),
loglevel=loglevel
) as portal: ) as portal:
results.append(( results.append((

12
piker/brokers/data.py
View File

@ -41,12 +41,15 @@ import tractor
from tractor.experimental import msgpub from tractor.experimental import msgpub
from async_generator import asynccontextmanager from async_generator import asynccontextmanager
from ._util import ( from piker.log import(
log, get_logger,
get_console_log, get_console_log,
) )
from . import get_brokermod from . import get_brokermod
log = get_logger(
name='piker.brokers.binance',
)
async def wait_for_network( async def wait_for_network(
net_func: Callable, net_func: Callable,
@ -243,7 +246,10 @@ async def start_quote_stream(
''' '''
# XXX: why do we need this again? # XXX: why do we need this again?
get_console_log(tractor.current_actor().loglevel) get_console_log(
level=tractor.current_actor().loglevel,
name=__name__,
)
# pull global vars from local actor # pull global vars from local actor
symbols = list(symbols) symbols = list(symbols)

4
piker/brokers/ib/_util.py
View File

@ -34,13 +34,13 @@ import subprocess
import tractor import tractor
from piker.brokers._util import get_logger from piker.log import get_logger
if TYPE_CHECKING: if TYPE_CHECKING:
from .api import Client from .api import Client
import i3ipc import i3ipc
log = get_logger('piker.brokers.ib') log = get_logger(name=__name__)
_reset_tech: Literal[ _reset_tech: Literal[
'vnc', 'vnc',

15
piker/brokers/ib/broker.py
View File

@ -50,6 +50,10 @@ from ib_insync.objects import (
) )
from piker import config from piker import config
from piker.log import (
get_logger,
get_console_log,
)
from piker.types import Struct from piker.types import Struct
from piker.accounting import ( from piker.accounting import (
Position, Position,
@ -77,7 +81,6 @@ from piker.clearing._messages import (
BrokerdFill, BrokerdFill,
BrokerdError, BrokerdError,
) )
from ._util import log
from .api import ( from .api import (
_accounts2clients, _accounts2clients,
get_config, get_config,
@ -95,6 +98,10 @@ from .ledger import (
update_ledger_from_api_trades, update_ledger_from_api_trades,
) )
log = get_logger(
name=__name__,
)
def pack_position( def pack_position(
pos: IbPosition, pos: IbPosition,
@ -536,9 +543,15 @@ class IbAcnt(Struct):
@tractor.context @tractor.context
async def open_trade_dialog( async def open_trade_dialog(
ctx: tractor.Context, ctx: tractor.Context,
loglevel: str = 'warning',
) -> AsyncIterator[dict[str, Any]]: ) -> AsyncIterator[dict[str, Any]]:
get_console_log(
level=loglevel,
name=__name__,
)
# task local msg dialog tracking # task local msg dialog tracking
flows = OrderDialogs() flows = OrderDialogs()
accounts_def = config.load_accounts(['ib']) accounts_def = config.load_accounts(['ib'])

5
piker/brokers/ib/feed.py
View File

@ -56,11 +56,11 @@ from piker.brokers._util import (
NoData, NoData,
DataUnavailable, DataUnavailable,
) )
from piker.log import get_logger
from .api import ( from .api import (
# _adhoc_futes_set, # _adhoc_futes_set,
Client, Client,
con2fqme, con2fqme,
log,
load_aio_clients, load_aio_clients,
MethodProxy, MethodProxy,
open_client_proxies, open_client_proxies,
@ -78,6 +78,9 @@ from .symbols import get_mkt_info
if TYPE_CHECKING: if TYPE_CHECKING:
from trio._core._run import Task from trio._core._run import Task
log = get_logger(
name=__name__,
)
# XXX NOTE: See available types table docs: # XXX NOTE: See available types table docs:
# https://interactivebrokers.github.io/tws-api/tick_types.html # https://interactivebrokers.github.io/tws-api/tick_types.html

5
piker/brokers/ib/ledger.py
View File

@ -44,6 +44,7 @@ from ib_insync import (
CommissionReport, CommissionReport,
) )
from piker.log import get_logger
from piker.types import Struct from piker.types import Struct
from piker.data import ( from piker.data import (
SymbologyCache, SymbologyCache,
@ -57,7 +58,6 @@ from piker.accounting import (
iter_by_dt, iter_by_dt,
) )
from ._flex_reports import parse_flex_dt from ._flex_reports import parse_flex_dt
from ._util import log
if TYPE_CHECKING: if TYPE_CHECKING:
from .api import ( from .api import (
@ -65,6 +65,9 @@ if TYPE_CHECKING:
MethodProxy, MethodProxy,
) )
log = get_logger(
name=__name__,
)
tx_sort: Callable = partial( tx_sort: Callable = partial(
iter_by_dt, iter_by_dt,

9
piker/brokers/ib/symbols.py
View File

@ -42,10 +42,7 @@ from piker.accounting import (
from piker._cacheables import ( from piker._cacheables import (
async_lifo_cache, async_lifo_cache,
) )
from piker.log import get_logger
from ._util import (
log,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from .api import ( from .api import (
@ -53,6 +50,10 @@ if TYPE_CHECKING:
Client, Client,
) )
log = get_logger(
name=__name__,
)
_futes_venues = ( _futes_venues = (
'GLOBEX', 'GLOBEX',
'NYMEX', 'NYMEX',

13
piker/brokers/kraken/broker.py
View File

@ -62,9 +62,12 @@ from piker.clearing._messages import (
from piker.brokers import ( from piker.brokers import (
open_cached_client, open_cached_client,
) )
from piker.log import (
get_console_log,
get_logger,
)
from piker.data import open_symcache from piker.data import open_symcache
from .api import ( from .api import (
log,
Client, Client,
BrokerError, BrokerError,
) )
@ -78,6 +81,8 @@ from .ledger import (
verify_balances, verify_balances,
) )
log = get_logger(name=__name__)
MsgUnion = Union[ MsgUnion = Union[
BrokerdCancel, BrokerdCancel,
BrokerdError, BrokerdError,
@ -431,9 +436,15 @@ def trades2pps(
@tractor.context @tractor.context
async def open_trade_dialog( async def open_trade_dialog(
ctx: tractor.Context, ctx: tractor.Context,
loglevel: str = 'warning',
) -> AsyncIterator[dict[str, Any]]: ) -> AsyncIterator[dict[str, Any]]:
get_console_log(
level=loglevel,
name=__name__,
)
async with ( async with (
# TODO: maybe bind these together and deliver # TODO: maybe bind these together and deliver
# a tuple from `.open_cached_client()`? # a tuple from `.open_cached_client()`?

19
piker/brokers/questrade.py
View File

@ -50,13 +50,19 @@ from . import open_cached_client
from piker._cacheables import async_lifo_cache from piker._cacheables import async_lifo_cache
from .. import config from .. import config
from ._util import resproc, BrokerError, SymbolNotFound from ._util import resproc, BrokerError, SymbolNotFound
from ..log import ( from piker.log import (
colorize_json, colorize_json,
)
from ._util import (
log,
get_console_log, get_console_log,
) )
from piker.log import (
get_logger,
)
log = get_logger(
name=__name__,
)
_use_practice_account = False _use_practice_account = False
_refresh_token_ep = 'https://{}login.questrade.com/oauth2/' _refresh_token_ep = 'https://{}login.questrade.com/oauth2/'
@ -1205,7 +1211,10 @@ async def stream_quotes(
# feed_type: str = 'stock', # feed_type: str = 'stock',
) -> AsyncGenerator[str, Dict[str, Any]]: ) -> AsyncGenerator[str, Dict[str, Any]]:
# XXX: required to propagate ``tractor`` loglevel to piker logging # XXX: required to propagate ``tractor`` loglevel to piker logging
get_console_log(loglevel) get_console_log(
level=loglevel,
name=__name__,
)
async with open_cached_client('questrade') as client: async with open_cached_client('questrade') as client:
if feed_type == 'stock': if feed_type == 'stock':

11
piker/brokers/robinhood.py
View File

@ -30,9 +30,16 @@ import asks
from ._util import ( from ._util import (
resproc, resproc,
BrokerError, BrokerError,
log,
) )
from ..calc import percent_change from piker.calc import percent_change
from piker.log import (
get_logger,
)
log = get_logger(
name=__name__,
)
_service_ep = 'https://api.robinhood.com' _service_ep = 'https://api.robinhood.com'

2
piker/clearing/_client.py
View File

@ -215,7 +215,7 @@ async def relay_orders_from_sync_code(
async def open_ems( async def open_ems(
fqme: str, fqme: str,
mode: str = 'live', mode: str = 'live',
loglevel: str = 'error', loglevel: str = 'warning',
) -> tuple[ ) -> tuple[
OrderClient, # client OrderClient, # client

47
piker/clearing/_ems.py
View File

@ -47,6 +47,7 @@ from tractor import trionics
from ._util import ( from ._util import (
log, # sub-sys logger log, # sub-sys logger
get_console_log, get_console_log,
subsys,
) )
from ..accounting._mktinfo import ( from ..accounting._mktinfo import (
unpack_fqme, unpack_fqme,
@ -136,7 +137,7 @@ class DarkBook(Struct):
tuple[ tuple[
Callable[[float], bool], # predicate Callable[[float], bool], # predicate
tuple[str, ...], # tickfilter tuple[str, ...], # tickfilter
dict | Order, # cmd / msg type dict|Order, # cmd / msg type
# live submission constraint parameters # live submission constraint parameters
float, # percent_away max price diff float, # percent_away max price diff
@ -278,7 +279,7 @@ async def clear_dark_triggers(
# remove exec-condition from set # remove exec-condition from set
log.info(f'Removing trigger for {oid}') log.info(f'Removing trigger for {oid}')
trigger: tuple | None = execs.pop(oid, None) trigger: tuple|None = execs.pop(oid, None)
if not trigger: if not trigger:
log.warning( log.warning(
f'trigger for {oid} was already removed!?' f'trigger for {oid} was already removed!?'
@ -336,8 +337,8 @@ async def open_brokerd_dialog(
brokermod: ModuleType, brokermod: ModuleType,
portal: tractor.Portal, portal: tractor.Portal,
exec_mode: str, exec_mode: str,
fqme: str | None = None, fqme: str|None = None,
loglevel: str | None = None, loglevel: str|None = None,
) -> tuple[ ) -> tuple[
tractor.MsgStream, tractor.MsgStream,
@ -351,9 +352,21 @@ async def open_brokerd_dialog(
broker backend, configuration, or client code usage. broker backend, configuration, or client code usage.
''' '''
get_console_log(
level=loglevel,
name='clearing',
)
# enable `.accounting` console since normally used by
# each `brokerd`.
get_console_log(
level=loglevel,
name='piker.accounting',
)
broker: str = brokermod.name broker: str = brokermod.name
def mk_paper_ep(): def mk_paper_ep(
loglevel: str,
):
from . import _paper_engine as paper_mod from . import _paper_engine as paper_mod
nonlocal brokermod, exec_mode nonlocal brokermod, exec_mode
@ -405,17 +418,21 @@ async def open_brokerd_dialog(
if ( if (
trades_endpoint is not None trades_endpoint is not None
or exec_mode != 'paper' or
exec_mode != 'paper'
): ):
# open live brokerd trades endpoint # open live brokerd trades endpoint
open_trades_endpoint = portal.open_context( open_trades_endpoint = portal.open_context(
trades_endpoint, trades_endpoint,
loglevel=loglevel,
) )
@acm @acm
async def maybe_open_paper_ep(): async def maybe_open_paper_ep():
if exec_mode == 'paper': if exec_mode == 'paper':
async with mk_paper_ep() as msg: async with mk_paper_ep(
loglevel=loglevel,
) as msg:
yield msg yield msg
return return
@ -426,7 +443,9 @@ async def open_brokerd_dialog(
# runtime indication that the backend can't support live # runtime indication that the backend can't support live
# order ctrl yet, so boot the paperboi B0 # order ctrl yet, so boot the paperboi B0
if first == 'paper': if first == 'paper':
async with mk_paper_ep() as msg: async with mk_paper_ep(
loglevel=loglevel,
) as msg:
yield msg yield msg
return return
else: else:
@ -761,12 +780,16 @@ _router: Router = None
@tractor.context @tractor.context
async def _setup_persistent_emsd( async def _setup_persistent_emsd(
ctx: tractor.Context, ctx: tractor.Context,
loglevel: str | None = None, loglevel: str|None = None,
) -> None: ) -> None:
if loglevel: if loglevel:
get_console_log(loglevel) _log = get_console_log(
level=loglevel,
name=subsys,
)
assert _log.name == 'piker.clearing'
global _router global _router
@ -822,7 +845,7 @@ async def translate_and_relay_brokerd_events(
f'Rx brokerd trade msg:\n' f'Rx brokerd trade msg:\n'
f'{fmsg}' f'{fmsg}'
) )
status_msg: Status | None = None status_msg: Status|None = None
match brokerd_msg: match brokerd_msg:
# BrokerdPosition # BrokerdPosition
@ -1283,7 +1306,7 @@ async def process_client_order_cmds(
and status.resp == 'dark_open' and status.resp == 'dark_open'
): ):
# remove from dark book clearing # remove from dark book clearing
entry: tuple | None = dark_book.triggers[fqme].pop(oid, None) entry: tuple|None = dark_book.triggers[fqme].pop(oid, None)
if entry: if entry:
( (
pred, pred,

16
piker/clearing/_paper_engine.py
View File

@ -59,9 +59,9 @@ from piker.data import (
open_symcache, open_symcache,
) )
from piker.types import Struct from piker.types import Struct
from ._util import ( from piker.log import (
log, # sub-sys logger
get_console_log, get_console_log,
get_logger,
) )
from ._messages import ( from ._messages import (
BrokerdCancel, BrokerdCancel,
@ -73,6 +73,8 @@ from ._messages import (
BrokerdError, BrokerdError,
) )
log = get_logger(name=__name__)
class PaperBoi(Struct): class PaperBoi(Struct):
''' '''
@ -550,16 +552,18 @@ _sells: defaultdict[
@tractor.context @tractor.context
async def open_trade_dialog( async def open_trade_dialog(
ctx: tractor.Context, ctx: tractor.Context,
broker: str, broker: str,
fqme: str | None = None, # if empty, we only boot broker mode fqme: str|None = None, # if empty, we only boot broker mode
loglevel: str = 'warning', loglevel: str = 'warning',
) -> None: ) -> None:
# enable piker.clearing console log for *this* subactor # enable piker.clearing console log for *this* `brokerd` subactor
get_console_log(loglevel) get_console_log(
level=loglevel,
name=__name__,
)
symcache: SymbologyCache symcache: SymbologyCache
async with open_symcache(get_brokermod(broker)) as symcache: async with open_symcache(get_brokermod(broker)) as symcache:

6
piker/clearing/_util.py
View File

@ -28,12 +28,14 @@ from ..log import (
from piker.types import Struct from piker.types import Struct
subsys: str = 'piker.clearing' subsys: str = 'piker.clearing'
log = get_logger(subsys) log = get_logger(
name='piker.clearing',
)
# TODO, oof doesn't this ignore the `loglevel` then??? # TODO, oof doesn't this ignore the `loglevel` then???
get_console_log = partial( get_console_log = partial(
get_console_log, get_console_log,
name=subsys, name='clearing',
) )

94
piker/cli/__init__.py
View File

@ -61,7 +61,8 @@ def load_trans_eps(
if ( if (
network network
and not maddrs and
not maddrs
): ):
# load network section and (attempt to) connect all endpoints # load network section and (attempt to) connect all endpoints
# which are reachable B) # which are reachable B)
@ -112,31 +113,27 @@ def load_trans_eps(
default=None, default=None,
help='Multiaddrs to bind or contact', help='Multiaddrs to bind or contact',
) )
# @click.option(
# '--tsdb',
# is_flag=True,
# help='Enable local ``marketstore`` instance'
# )
# @click.option(
# '--es',
# is_flag=True,
# help='Enable local ``elasticsearch`` instance'
# )
def pikerd( def pikerd(
maddr: list[str] | None, maddr: list[str] | None,
loglevel: str, loglevel: str,
tl: bool, tl: bool,
pdb: bool, pdb: bool,
# tsdb: bool,
# es: bool,
): ):
''' '''
Spawn the piker broker-daemon. Start the "root service actor", `pikerd`, run it until
cancellation.
This "root daemon" operates as the top most service-mngr and
subsys-as-subactor supervisor, think of it as the "init proc" of
any of any `piker` application or daemon-process tree.
''' '''
# from tractor.devx import maybe_open_crash_handler # from tractor.devx import maybe_open_crash_handler
# with maybe_open_crash_handler(pdb=False): # with maybe_open_crash_handler(pdb=False):
log = get_console_log(loglevel, name='cli') log = get_console_log(
level=loglevel,
with_tractor_log=tl,
)
if pdb: if pdb:
log.warning(( log.warning((
@ -237,6 +234,14 @@ def cli(
regaddr: str, regaddr: str,
) -> None: ) -> None:
'''
The "root" `piker`-cmd CLI endpoint.
NOTE, this def generally relies on and requires a sub-cmd to be
provided by the user, OW only a `--help` msg (listing said
subcmds) will be dumped to console.
'''
if configdir is not None: if configdir is not None:
assert os.path.isdir(configdir), f"`{configdir}` is not a valid path" assert os.path.isdir(configdir), f"`{configdir}` is not a valid path"
config._override_config_dir(configdir) config._override_config_dir(configdir)
@ -295,17 +300,50 @@ def cli(
@click.option('--tl', is_flag=True, help='Enable tractor logging') @click.option('--tl', is_flag=True, help='Enable tractor logging')
@click.argument('ports', nargs=-1, required=False) @click.argument('ports', nargs=-1, required=False)
@click.pass_obj @click.pass_obj
def services(config, tl, ports): def services(
config,
tl: bool,
ports: list[int],
):
'''
List all `piker` "service deamons" to the console in
a `json`-table which maps each actor's UID in the form,
from ..service import ( `{service_name}.{subservice_name}.{UUID}`
to its (primary) IPC server address.
(^TODO, should be its multiaddr form once we support it)
Note that by convention actors which operate as "headless"
processes (those without GUIs/graphics, and which generally
parent some noteworthy subsystem) are normally suffixed by
a "d" such as,
- pikerd: the root runtime supervisor
- brokerd: a broker-backend order ctl daemon
- emsd: the internal dark-clearing and order routing daemon
- datad: a data-provider-backend data feed daemon
- samplerd: the real-time data sampling and clock-syncing daemon
"Headed units" are normally just given an obvious app-like name
with subactors indexed by `.` such as,
- chart: the primary modal charting iface, a Qt app
- chart.fsp_0: a financial-sig-proc cascade instance which
delivers graphics to a parent `chart` app.
- polars_boi: some (presumably) `polars` using console app.
'''
from piker.service import (
open_piker_runtime, open_piker_runtime,
_default_registry_port, _default_registry_port,
_default_registry_host, _default_registry_host,
) )
host = _default_registry_host # !TODO, mk this to work with UDS!
host: str = _default_registry_host
if not ports: if not ports:
ports = [_default_registry_port] ports: list[int] = [_default_registry_port]
addr = tractor._addr.wrap_address( addr = tractor._addr.wrap_address(
addr=(host, ports[0]) addr=(host, ports[0])
@ -316,7 +354,11 @@ def services(config, tl, ports):
async with ( async with (
open_piker_runtime( open_piker_runtime(
name='service_query', name='service_query',
loglevel=config['loglevel'] if tl else None, loglevel=(
config['loglevel']
if tl
else None
),
), ),
tractor.get_registry( tractor.get_registry(
addr=addr, addr=addr,
@ -336,7 +378,15 @@ def services(config, tl, ports):
def _load_clis() -> None: def _load_clis() -> None:
# from ..service import elastic # noqa '''
Dynamically load and register all subsys CLI endpoints (at call
time).
NOTE, obviously this is normally expected to be called at
`import` time and implicitly relies on our use of various
`click`/`typer` decorator APIs.
'''
from ..brokers import cli # noqa from ..brokers import cli # noqa
from ..ui import cli # noqa from ..ui import cli # noqa
from ..watchlists import cli # noqa from ..watchlists import cli # noqa
@ -346,5 +396,5 @@ def _load_clis() -> None:
from ..accounting import cli # noqa from ..accounting import cli # noqa
# load downstream cli modules # load all subsytem cli eps
_load_clis() _load_clis()

23
piker/data/_sampling.py
View File

@ -336,10 +336,18 @@ async def register_with_sampler(
open_index_stream: bool = True, # open a 2way stream for sample step msgs? open_index_stream: bool = True, # open a 2way stream for sample step msgs?
sub_for_broadcasts: bool = True, # sampler side to send step updates? sub_for_broadcasts: bool = True, # sampler side to send step updates?
loglevel: str|None = None,
) -> set[int]: ) -> set[int]:
get_console_log(tractor.current_actor().loglevel) get_console_log(
level=(
loglevel
or
tractor.current_actor().loglevel
),
name=__name__,
)
incr_was_started: bool = False incr_was_started: bool = False
try: try:
@ -476,6 +484,7 @@ async def spawn_samplerd(
register_with_sampler, register_with_sampler,
period_s=1, period_s=1,
sub_for_broadcasts=False, sub_for_broadcasts=False,
loglevel=loglevel,
) )
return True return True
@ -484,7 +493,6 @@ async def spawn_samplerd(
@acm @acm
async def maybe_open_samplerd( async def maybe_open_samplerd(
loglevel: str|None = None, loglevel: str|None = None,
**pikerd_kwargs, **pikerd_kwargs,
@ -513,10 +521,10 @@ async def open_sample_stream(
shms_by_period: dict[float, dict]|None = None, shms_by_period: dict[float, dict]|None = None,
open_index_stream: bool = True, open_index_stream: bool = True,
sub_for_broadcasts: bool = True, sub_for_broadcasts: bool = True,
loglevel: str|None = None,
cache_key: str|None = None, # cache_key: str|None = None,
allow_new_sampler: bool = True, # allow_new_sampler: bool = True,
ensure_is_active: bool = False, ensure_is_active: bool = False,
) -> AsyncIterator[dict[str, float]]: ) -> AsyncIterator[dict[str, float]]:
@ -551,7 +559,9 @@ async def open_sample_stream(
# XXX: this should be singleton on a host, # XXX: this should be singleton on a host,
# a lone broker-daemon per provider should be # a lone broker-daemon per provider should be
# created for all practical purposes # created for all practical purposes
maybe_open_samplerd() as portal, maybe_open_samplerd(
loglevel=loglevel,
) as portal,
portal.open_context( portal.open_context(
register_with_sampler, register_with_sampler,
@ -560,6 +570,7 @@ async def open_sample_stream(
'shms_by_period': shms_by_period, 'shms_by_period': shms_by_period,
'open_index_stream': open_index_stream, 'open_index_stream': open_index_stream,
'sub_for_broadcasts': sub_for_broadcasts, 'sub_for_broadcasts': sub_for_broadcasts,
'loglevel': loglevel,
}, },
) as (ctx, shm_periods) ) as (ctx, shm_periods)
): ):

4
piker/data/_util.py
View File

@ -26,7 +26,9 @@ from ..log import (
) )
subsys: str = 'piker.data' subsys: str = 'piker.data'
log = get_logger(subsys) log = get_logger(
name=subsys,
)
get_console_log = partial( get_console_log = partial(
get_console_log, get_console_log,

35
piker/data/feed.py
View File

@ -62,7 +62,6 @@ from ._util import (
log, log,
get_console_log, get_console_log,
) )
from .flows import Flume
from .validate import ( from .validate import (
FeedInit, FeedInit,
validate_backend, validate_backend,
@ -77,6 +76,7 @@ from ._sampling import (
) )
if TYPE_CHECKING: if TYPE_CHECKING:
from .flows import Flume
from tractor._addr import Address from tractor._addr import Address
from tractor.msg.types import Aid from tractor.msg.types import Aid
@ -239,7 +239,6 @@ async def allocate_persistent_feed(
brokername: str, brokername: str,
symstr: str, symstr: str,
loglevel: str, loglevel: str,
start_stream: bool = True, start_stream: bool = True,
init_timeout: float = 616, init_timeout: float = 616,
@ -278,7 +277,7 @@ async def allocate_persistent_feed(
# ``stream_quotes()``, a required broker backend endpoint. # ``stream_quotes()``, a required broker backend endpoint.
init_msgs: ( init_msgs: (
list[FeedInit] # new list[FeedInit] # new
| dict[str, dict[str, str]] # legacy / deprecated |dict[str, dict[str, str]] # legacy / deprecated
) )
# TODO: probably make a struct msg type for this as well # TODO: probably make a struct msg type for this as well
@ -348,11 +347,14 @@ async def allocate_persistent_feed(
izero_rt, izero_rt,
rt_shm, rt_shm,
) = await bus.nursery.start( ) = await bus.nursery.start(
partial(
manage_history, manage_history,
mod, mod=mod,
mkt, mkt=mkt,
some_data_ready, some_data_ready=some_data_ready,
feed_is_live, feed_is_live=feed_is_live,
loglevel=loglevel,
)
) )
# yield back control to starting nursery once we receive either # yield back control to starting nursery once we receive either
@ -362,6 +364,8 @@ async def allocate_persistent_feed(
) )
await some_data_ready.wait() await some_data_ready.wait()
# XXX, avoid cycle; it imports this mod.
from .flows import Flume
flume = Flume( flume = Flume(
# TODO: we have to use this for now since currently the # TODO: we have to use this for now since currently the
@ -458,7 +462,6 @@ async def allocate_persistent_feed(
@tractor.context @tractor.context
async def open_feed_bus( async def open_feed_bus(
ctx: tractor.Context, ctx: tractor.Context,
brokername: str, brokername: str,
symbols: list[str], # normally expected to the broker-specific fqme symbols: list[str], # normally expected to the broker-specific fqme
@ -479,13 +482,16 @@ async def open_feed_bus(
''' '''
if loglevel is None: if loglevel is None:
loglevel = tractor.current_actor().loglevel loglevel: str = tractor.current_actor().loglevel
# XXX: required to propagate ``tractor`` loglevel to piker # XXX: required to propagate ``tractor`` loglevel to piker
# logging # logging
get_console_log( get_console_log(
loglevel level=(loglevel
or tractor.current_actor().loglevel or
tractor.current_actor().loglevel
),
name=__name__,
) )
# local state sanity checks # local state sanity checks
@ -500,7 +506,6 @@ async def open_feed_bus(
sub_registered = trio.Event() sub_registered = trio.Event()
flumes: dict[str, Flume] = {} flumes: dict[str, Flume] = {}
for symbol in symbols: for symbol in symbols:
# if no cached feed for this symbol has been created for this # if no cached feed for this symbol has been created for this
@ -684,6 +689,7 @@ class Feed(Struct):
''' '''
mods: dict[str, ModuleType] = {} mods: dict[str, ModuleType] = {}
portals: dict[ModuleType, tractor.Portal] = {} portals: dict[ModuleType, tractor.Portal] = {}
flumes: dict[ flumes: dict[
str, # FQME str, # FQME
Flume, Flume,
@ -797,7 +803,7 @@ async def install_brokerd_search(
@acm @acm
async def maybe_open_feed( async def maybe_open_feed(
fqmes: list[str], fqmes: list[str],
loglevel: str | None = None, loglevel: str|None = None,
**kwargs, **kwargs,
@ -881,7 +887,6 @@ async def open_feed(
# one actor per brokerd for now # one actor per brokerd for now
brokerd_ctxs = [] brokerd_ctxs = []
for brokermod, bfqmes in providers.items(): for brokermod, bfqmes in providers.items():
# if no `brokerd` for this backend exists yet we spawn # if no `brokerd` for this backend exists yet we spawn
@ -951,6 +956,8 @@ async def open_feed(
assert len(feed.mods) == len(feed.portals) assert len(feed.mods) == len(feed.portals)
# XXX, avoid cycle; it imports this mod.
from .flows import Flume
async with ( async with (
trionics.gather_contexts(bus_ctxs) as ctxs, trionics.gather_contexts(bus_ctxs) as ctxs,
): ):

41
piker/fsp/_engine.py
View File

@ -24,6 +24,7 @@ from functools import partial
from typing import ( from typing import (
AsyncIterator, AsyncIterator,
Callable, Callable,
TYPE_CHECKING,
) )
import numpy as np import numpy as np
@ -33,12 +34,12 @@ import tractor
from tractor.msg import NamespacePath from tractor.msg import NamespacePath
from piker.types import Struct from piker.types import Struct
from ..log import get_logger, get_console_log from ..log import (
from .. import data get_logger,
from ..data.feed import ( get_console_log,
Flume,
Feed,
) )
from .. import data
from ..data.flows import Flume
from ..data._sharedmem import ShmArray from ..data._sharedmem import ShmArray
from ..data._sampling import ( from ..data._sampling import (
_default_delay_s, _default_delay_s,
@ -52,6 +53,9 @@ from ._api import (
) )
from ..toolz import Profiler from ..toolz import Profiler
if TYPE_CHECKING:
from ..data.feed import Feed
log = get_logger(__name__) log = get_logger(__name__)
@ -169,8 +173,10 @@ class Cascade(Struct):
if not synced: if not synced:
fsp: Fsp = self.fsp fsp: Fsp = self.fsp
log.warning( log.warning(
'***DESYNCED FSP***\n' f'***DESYNCED fsp***\n'
f'{fsp.ns_path}@{src_shm.token}\n' f'------------------\n'
f'ns-path: {fsp.ns_path!r}\n'
f'shm-token: {src_shm.token}\n'
f'step_diff: {step_diff}\n' f'step_diff: {step_diff}\n'
f'len_diff: {len_diff}\n' f'len_diff: {len_diff}\n'
) )
@ -398,7 +404,6 @@ async def connect_streams(
@tractor.context @tractor.context
async def cascade( async def cascade(
ctx: tractor.Context, ctx: tractor.Context,
# data feed key # data feed key
@ -412,7 +417,7 @@ async def cascade(
shm_registry: dict[str, _Token], shm_registry: dict[str, _Token],
zero_on_step: bool = False, zero_on_step: bool = False,
loglevel: str | None = None, loglevel: str|None = None,
) -> None: ) -> None:
''' '''
@ -426,7 +431,17 @@ async def cascade(
) )
if loglevel: if loglevel:
get_console_log(loglevel) log = get_console_log(
loglevel,
name=__name__,
)
# XXX TODO!
# figure out why this writes a dict to,
# `tractor._state._runtime_vars['_root_mailbox']`
# XD .. wtf
# TODO, solve this as reported in,
# https://www.pikers.dev/pikers/piker/issues/70
# await tractor.pause()
src: Flume = Flume.from_msg(src_flume_addr) src: Flume = Flume.from_msg(src_flume_addr)
dst: Flume = Flume.from_msg( dst: Flume = Flume.from_msg(
@ -469,7 +484,8 @@ async def cascade(
# open a data feed stream with requested broker # open a data feed stream with requested broker
feed: Feed feed: Feed
async with data.feed.maybe_open_feed( async with data.feed.maybe_open_feed(
[fqme], fqmes=[fqme],
loglevel=loglevel,
# TODO throttle tick outputs from *this* daemon since # TODO throttle tick outputs from *this* daemon since
# it'll emit tons of ticks due to the throttle only # it'll emit tons of ticks due to the throttle only
@ -567,7 +583,8 @@ async def cascade(
# on every step msg received from the global `samplerd` # on every step msg received from the global `samplerd`
# service. # service.
async with open_sample_stream( async with open_sample_stream(
float(delay_s) period_s=float(delay_s),
loglevel=loglevel,
) as istream: ) as istream:
profiler(f'{func_name}: sample stream up') profiler(f'{func_name}: sample stream up')

71
piker/log.py
View File

@ -37,35 +37,84 @@ _proj_name: str = 'piker'
def get_logger( def get_logger(
name: str = None, name: str|None = None,
**tractor_log_kwargs,
) -> logging.Logger: ) -> logging.Logger:
''' '''
Return the package log or a sub-log for `name` if provided. Return the package log or a sub-logger if a `name=` is provided,
which defaults to the calling module's pkg-namespace path.
See `tractor.log.get_logger()` for details.
''' '''
pkg_name: str = _proj_name
if (
name
and
pkg_name in name
):
name: str = name.lstrip(f'{_proj_name}.')
return tractor.log.get_logger( return tractor.log.get_logger(
name=name, name=name,
_root_name=_proj_name, pkg_name=pkg_name,
**tractor_log_kwargs,
) )
def get_console_log( def get_console_log(
goodboy commented 2026-02-23 03:33:12 +00:00
Review

Note the changes here to handle re-routing to tractor.log.get_console_log() appropriately given use throughout piker.

Note the changes here to handle re-routing to `tractor.log.get_console_log()` appropriately given use throughout `piker`.
level: str | None = None, level: str|None = None,
name: str | None = None, name: str|None = None,
pkg_name: str|None = None,
with_tractor_log: bool = False,
# ?TODO, support a "log-spec" style `str|dict[str, str]` which
# dictates both the sublogger-key and a level?
# -> see similar idea in `modden`'s usage.
**tractor_log_kwargs,
) -> logging.Logger: ) -> logging.Logger:
''' '''
Get the package logger and enable a handler which writes to stderr. Get the package logger and enable a handler which writes to
stderr.
Yeah yeah, i know we can use ``DictConfig``. You do it... Yeah yeah, i know we can use `DictConfig`.
You do it.. Bp
''' '''
pkg_name: str = _proj_name
if (
name
and
pkg_name in name
):
name: str = name.lstrip(f'{_proj_name}.')
tll: str|None = None
if (
with_tractor_log is not False
):
tll = level
elif maybe_actor := tractor.current_actor(
err_on_no_runtime=False,
):
tll = maybe_actor.loglevel
if tll:
t_log = tractor.log.get_console_log(
level=tll,
name='tractor', # <- XXX, force root tractor log!
**tractor_log_kwargs,
)
# TODO/ allow only enabling certain tractor sub-logs?
assert t_log.name == 'tractor'
return tractor.log.get_console_log( return tractor.log.get_console_log(
level, level=level,
name=name, name=name,
_root_name=_proj_name, pkg_name=pkg_name,
) # our root logger **tractor_log_kwargs,
)
def colorize_json( def colorize_json(

22
piker/service/_actor_runtime.py
View File

@ -21,7 +21,6 @@
from __future__ import annotations from __future__ import annotations
import os import os
from typing import ( from typing import (
Optional,
Any, Any,
ClassVar, ClassVar,
) )
@ -32,9 +31,12 @@ from contextlib import (
import tractor import tractor
import trio import trio
from ._util import ( from piker.log import (
get_console_log, get_console_log,
) )
from ._util import (
subsys,
)
from ._mngr import ( from ._mngr import (
Services, Services,
) )
@ -59,7 +61,7 @@ async def open_piker_runtime(
registry_addrs: list[tuple[str, int]] = [], registry_addrs: list[tuple[str, int]] = [],
enable_modules: list[str] = [], enable_modules: list[str] = [],
loglevel: Optional[str] = None, loglevel: str|None = None,
# XXX NOTE XXX: you should pretty much never want debug mode # XXX NOTE XXX: you should pretty much never want debug mode
# for data daemons when running in production. # for data daemons when running in production.
@ -69,7 +71,7 @@ async def open_piker_runtime(
# and spawn the service tree distributed per that. # and spawn the service tree distributed per that.
start_method: str = 'trio', start_method: str = 'trio',
tractor_runtime_overrides: dict | None = None, tractor_runtime_overrides: dict|None = None,
**tractor_kwargs, **tractor_kwargs,
) -> tuple[ ) -> tuple[
@ -97,7 +99,8 @@ async def open_piker_runtime(
# setting it as the root actor on localhost. # setting it as the root actor on localhost.
registry_addrs = ( registry_addrs = (
registry_addrs registry_addrs
or [_default_reg_addr] or
[_default_reg_addr]
) )
if ems := tractor_kwargs.pop('enable_modules', None): if ems := tractor_kwargs.pop('enable_modules', None):
@ -163,8 +166,7 @@ _root_modules: list[str] = [
@acm @acm
async def open_pikerd( async def open_pikerd(
registry_addrs: list[tuple[str, int]], registry_addrs: list[tuple[str, int]],
loglevel: str|None = None,
loglevel: str | None = None,
# XXX: you should pretty much never want debug mode # XXX: you should pretty much never want debug mode
# for data daemons when running in production. # for data daemons when running in production.
@ -192,7 +194,6 @@ async def open_pikerd(
async with ( async with (
open_piker_runtime( open_piker_runtime(
name=_root_dname, name=_root_dname,
loglevel=loglevel, loglevel=loglevel,
debug_mode=debug_mode, debug_mode=debug_mode,
@ -273,7 +274,10 @@ async def maybe_open_pikerd(
''' '''
if loglevel: if loglevel:
get_console_log(loglevel) get_console_log(
name=subsys,
level=loglevel
)
# subtle, we must have the runtime up here or portal lookup will fail # subtle, we must have the runtime up here or portal lookup will fail
query_name = kwargs.pop( query_name = kwargs.pop(

15
piker/service/_ahab.py
View File

@ -49,13 +49,15 @@ from requests.exceptions import (
ReadTimeout, ReadTimeout,
) )
from ._mngr import Services from piker.log import (
from ._util import (
log, # sub-sys logger
get_console_log, get_console_log,
get_logger,
) )
from ._mngr import Services
from .. import config from .. import config
log = get_logger(name=__name__)
class DockerNotStarted(Exception): class DockerNotStarted(Exception):
'Prolly you dint start da daemon bruh' 'Prolly you dint start da daemon bruh'
@ -336,13 +338,16 @@ class Container:
async def open_ahabd( async def open_ahabd(
ctx: tractor.Context, ctx: tractor.Context,
endpoint: str, # ns-pointer str-msg-type endpoint: str, # ns-pointer str-msg-type
loglevel: str | None = None, loglevel: str = 'cancel',
**ep_kwargs, **ep_kwargs,
) -> None: ) -> None:
log = get_console_log(loglevel or 'cancel') log = get_console_log(
level=loglevel,
name='piker.service',
)
async with open_docker() as client: async with open_docker() as client:

22
piker/service/_daemon.py
View File

@ -30,8 +30,9 @@ from contextlib import (
import tractor import tractor
from trio.lowlevel import current_task from trio.lowlevel import current_task
from ._util import ( from piker.log import (
log, # sub-sys logger get_console_log,
get_logger,
) )
from ._mngr import ( from ._mngr import (
Services, Services,
@ -39,16 +40,17 @@ from ._mngr import (
from ._actor_runtime import maybe_open_pikerd from ._actor_runtime import maybe_open_pikerd
from ._registry import find_service from ._registry import find_service
log = get_logger(name=__name__)
@acm @acm
async def maybe_spawn_daemon( async def maybe_spawn_daemon(
service_name: str, service_name: str,
service_task_target: Callable, service_task_target: Callable,
spawn_args: dict[str, Any], spawn_args: dict[str, Any],
loglevel: str | None = None, loglevel: str|None = None,
singleton: bool = False, singleton: bool = False,
**pikerd_kwargs, **pikerd_kwargs,
@ -66,6 +68,12 @@ async def maybe_spawn_daemon(
clients. clients.
''' '''
log = get_console_log(
level=loglevel,
name=__name__,
)
assert log.name == 'piker.service'
# serialize access to this section to avoid # serialize access to this section to avoid
# 2 or more tasks racing to create a daemon # 2 or more tasks racing to create a daemon
lock = Services.locks[service_name] lock = Services.locks[service_name]
@ -152,8 +160,7 @@ async def maybe_spawn_daemon(
async def spawn_emsd( async def spawn_emsd(
loglevel: str|None = None,
loglevel: str | None = None,
**extra_tractor_kwargs **extra_tractor_kwargs
) -> bool: ) -> bool:
@ -190,9 +197,8 @@ async def spawn_emsd(
@acm @acm
async def maybe_open_emsd( async def maybe_open_emsd(
brokername: str, brokername: str,
loglevel: str | None = None, loglevel: str|None = None,
**pikerd_kwargs, **pikerd_kwargs,

6
piker/service/_mngr.py
View File

@ -34,9 +34,9 @@ from tractor import (
Portal, Portal,
) )
from ._util import ( from piker.log import get_logger
log, # sub-sys logger
) log = get_logger(name=__name__)
# TODO: we need remote wrapping and a general soln: # TODO: we need remote wrapping and a general soln:

69
piker/service/_registry.py
View File

@ -27,15 +27,29 @@ from typing import (
) )
import tractor import tractor
from tractor import Portal from tractor import (
msg,
from ._util import ( Actor,
log, # sub-sys logger Portal,
) )
from piker.log import get_logger
log = get_logger(name=__name__)
# TODO? default path-space for UDS registry?
# [ ] needs to be Xplatform tho!
# _default_registry_path: Path = (
# Path(os.environ['XDG_RUNTIME_DIR'])
# /'piker'
# )
_default_registry_host: str = '127.0.0.1' _default_registry_host: str = '127.0.0.1'
_default_registry_port: int = 6116 _default_registry_port: int = 6116
_default_reg_addr: tuple[str, int] = ( _default_reg_addr: tuple[
str,
int, # |str TODO, once we support UDS, see above.
] = (
_default_registry_host, _default_registry_host,
_default_registry_port, _default_registry_port,
) )
@ -75,16 +89,22 @@ async def open_registry(
''' '''
global _tractor_kwargs global _tractor_kwargs
actor = tractor.current_actor() actor: Actor = tractor.current_actor()
uid = actor.uid aid: msg.Aid = actor.aid
preset_reg_addrs: list[tuple[str, int]] = Registry.addrs uid: tuple[str, str] = aid.uid
preset_reg_addrs: list[
tuple[str, int]
] = Registry.addrs
if ( if (
preset_reg_addrs preset_reg_addrs
and addrs and
addrs
): ):
if preset_reg_addrs != addrs: if preset_reg_addrs != addrs:
# if any(addr in preset_reg_addrs for addr in addrs): # if any(addr in preset_reg_addrs for addr in addrs):
diff: set[tuple[str, int]] = set(preset_reg_addrs) - set(addrs) diff: set[
tuple[str, int]
] = set(preset_reg_addrs) - set(addrs)
if diff: if diff:
log.warning( log.warning(
f'`{uid}` requested only subset of registrars: {addrs}\n' f'`{uid}` requested only subset of registrars: {addrs}\n'
@ -98,7 +118,6 @@ async def open_registry(
) )
was_set: bool = False was_set: bool = False
if ( if (
not tractor.is_root_process() not tractor.is_root_process()
and and
@ -115,16 +134,23 @@ async def open_registry(
f"`{uid}` registry should already exist but doesn't?" f"`{uid}` registry should already exist but doesn't?"
) )
if ( if not Registry.addrs:
not Registry.addrs
):
was_set = True was_set = True
Registry.addrs = addrs or [_default_reg_addr] Registry.addrs = (
addrs
or
[_default_reg_addr]
)
# NOTE: only spot this seems currently used is inside # NOTE: only spot this seems currently used is inside
# `.ui._exec` which is the (eventual qtloops) bootstrapping # `.ui._exec` which is the (eventual qtloops) bootstrapping
# with guest mode. # with guest mode.
_tractor_kwargs['registry_addrs'] = Registry.addrs reg_addrs: list[tuple[str, str|int]] = Registry.addrs
# !TODO, a struct-API to stringently allow this only in special
# cases?
# -> better would be to have some way to (atomically) rewrite
# and entire `RuntimeVars`?? ideas welcome obvi..
_tractor_kwargs['registry_addrs'] = reg_addrs
try: try:
yield Registry.addrs yield Registry.addrs
@ -149,7 +175,7 @@ async def find_service(
| None | None
): ):
# try: # try:
reg_addrs: list[tuple[str, int]] reg_addrs: list[tuple[str, int|str]]
async with open_registry( async with open_registry(
addrs=( addrs=(
registry_addrs registry_addrs
@ -172,15 +198,13 @@ async def find_service(
only_first=first_only, # if set only returns single ref only_first=first_only, # if set only returns single ref
) as maybe_portals: ) as maybe_portals:
if not maybe_portals: if not maybe_portals:
# log.info( log.info(
print(
f'Could NOT find service {service_name!r} -> {maybe_portals!r}' f'Could NOT find service {service_name!r} -> {maybe_portals!r}'
) )
yield None yield None
return return
# log.info( log.info(
print(
f'Found service {service_name!r} -> {maybe_portals}' f'Found service {service_name!r} -> {maybe_portals}'
) )
yield maybe_portals yield maybe_portals
@ -195,8 +219,7 @@ async def find_service(
async def check_for_service( async def check_for_service(
service_name: str, service_name: str,
) -> None|tuple[str, int]:
) -> None | tuple[str, int]:
''' '''
Service daemon "liveness" predicate. Service daemon "liveness" predicate.

18
piker/service/_util.py
View File

@ -14,20 +14,12 @@
# You should have received a copy of the GNU Affero General Public License # 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/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
""" """
Sub-sys module commons. Sub-sys module commons (if any ?? Bp).
""" """
from functools import partial
from ..log import (
get_logger,
get_console_log,
)
subsys: str = 'piker.service' subsys: str = 'piker.service'
log = get_logger(subsys) # ?TODO, if we were going to keep a `get_console_log()` in here to be
# invoked at `import`-time, how do we dynamically hand in the
get_console_log = partial( # `level=` value? seems too early in the runtime to be injected
get_console_log, # right?
name=subsys,
)

15
piker/service/elastic.py
View File

@ -16,6 +16,7 @@
from __future__ import annotations from __future__ import annotations
from contextlib import asynccontextmanager as acm from contextlib import asynccontextmanager as acm
from pprint import pformat
from typing import ( from typing import (
Any, Any,
TYPE_CHECKING, TYPE_CHECKING,
@ -26,12 +27,17 @@ import asks
if TYPE_CHECKING: if TYPE_CHECKING:
import docker import docker
from ._ahab import DockerContainer from ._ahab import DockerContainer
from . import (
Services,
)
from ._util import log # sub-sys logger from piker.log import (
from ._util import (
get_console_log, get_console_log,
get_logger,
) )
log = get_logger(name=__name__)
# container level config # container level config
_config = { _config = {
@ -67,7 +73,10 @@ def start_elasticsearch(
elastic elastic
''' '''
get_console_log('info', name=__name__) get_console_log(
level='info',
name=__name__,
)
dcntr: DockerContainer = client.containers.run( dcntr: DockerContainer = client.containers.run(
'piker:elastic', 'piker:elastic',

7
piker/service/marketstore.py
View File

@ -52,17 +52,18 @@ import pendulum
# TODO: import this for specific error set expected by mkts client # TODO: import this for specific error set expected by mkts client
# import purerpc # import purerpc
from ..data.feed import maybe_open_feed from piker.data.feed import maybe_open_feed
from . import Services from . import Services
from ._util import ( from piker.log import (
log, # sub-sys logger
get_console_log, get_console_log,
get_logger,
) )
if TYPE_CHECKING: if TYPE_CHECKING:
import docker import docker
from ._ahab import DockerContainer from ._ahab import DockerContainer
log = get_logger(name=__name__)
# ahabd-supervisor and container level config # ahabd-supervisor and container level config

4
piker/tsp/_anal.py
View File

@ -54,10 +54,10 @@ from ..log import (
# for "time series processing" # for "time series processing"
subsys: str = 'piker.tsp' subsys: str = 'piker.tsp'
log = get_logger(subsys) log = get_logger(name=__name__)
get_console_log = partial( get_console_log = partial(
get_console_log, get_console_log,
name=subsys, name=subsys, # activate for subsys-pkg "downward"
) )
# NOTE: union type-defs to handle generic `numpy` and `polars` types # NOTE: union type-defs to handle generic `numpy` and `polars` types

14
piker/tsp/_history.py
View File

@ -63,8 +63,10 @@ from ..data._sharedmem import (
maybe_open_shm_array, maybe_open_shm_array,
ShmArray, ShmArray,
) )
from ..data._source import def_iohlcv_fields from piker.data._source import (
from ..data._sampling import ( def_iohlcv_fields,
)
from piker.data._sampling import (
open_sample_stream, open_sample_stream,
) )
@ -96,7 +98,9 @@ if TYPE_CHECKING:
# from .feed import _FeedsBus # from .feed import _FeedsBus
log = get_logger(__name__) log = get_logger(
name=__name__,
)
# `ShmArray` buffer sizing configuration: # `ShmArray` buffer sizing configuration:
@ -550,7 +554,7 @@ async def start_backfill(
) )
# ?TODO, check against venue closure hours # ?TODO, check against venue closure hours
# if/when provided by backend? # if/when provided by backend?
await tractor.pause() # await tractor.pause()
expected_dur: Interval = ( expected_dur: Interval = (
last_start_dt.subtract( last_start_dt.subtract(
@ -1320,6 +1324,7 @@ async def manage_history(
mkt: MktPair, mkt: MktPair,
some_data_ready: trio.Event, some_data_ready: trio.Event,
feed_is_live: trio.Event, feed_is_live: trio.Event,
loglevel: str = 'warning',
timeframe: float = 60, # in seconds timeframe: float = 60, # in seconds
wait_for_live_timeout: float = 0.5, wait_for_live_timeout: float = 0.5,
@ -1497,6 +1502,7 @@ async def manage_history(
# data feed layer that needs to consume it). # data feed layer that needs to consume it).
open_index_stream=True, open_index_stream=True,
sub_for_broadcasts=False, sub_for_broadcasts=False,
loglevel=loglevel,
) as sample_stream: ) as sample_stream:
# register 1s and 1m buffers with the global # register 1s and 1m buffers with the global

12
piker/ui/_app.py
View File

@ -33,7 +33,10 @@ from . import _search
from ..accounting import unpack_fqme from ..accounting import unpack_fqme
from ..data._symcache import open_symcache from ..data._symcache import open_symcache
from ..data.feed import install_brokerd_search from ..data.feed import install_brokerd_search
from ..log import get_logger from ..log import (
get_logger,
get_console_log,
)
from ..service import maybe_spawn_brokerd from ..service import maybe_spawn_brokerd
from ._exec import run_qtractor from ._exec import run_qtractor
@ -87,6 +90,13 @@ async def _async_main(
Provision the "main" widget with initial symbol data and root nursery. Provision the "main" widget with initial symbol data and root nursery.
""" """
# enable chart's console logging
if loglevel:
get_console_log(
level=loglevel,
name=__name__,
)
# set as singleton # set as singleton
_chart._godw = main_widget _chart._godw = main_widget

11
piker/ui/_cursor.py
View File

@ -413,9 +413,18 @@ class Cursor(pg.GraphicsObject):
self, self,
item: pg.GraphicsObject, item: pg.GraphicsObject,
) -> None: ) -> None:
assert getattr(item, 'delete'), f"{item} must define a ``.delete()``" assert getattr(
item,
'delete',
), f"{item} must define a ``.delete()``"
self._hovered.add(item) self._hovered.add(item)
def is_hovered(
self,
item: pg.GraphicsObject,
) -> bool:
return item in self._hovered
def add_plot( def add_plot(
self, self,
plot: ChartPlotWidget, # noqa plot: ChartPlotWidget, # noqa

2
piker/ui/_dataviz.py
View File

@ -45,7 +45,7 @@ from piker.ui.qt import QLineF
from ..data._sharedmem import ( from ..data._sharedmem import (
ShmArray, ShmArray,
) )
from ..data.feed import Flume from ..data.flows import Flume
from ..data._formatters import ( from ..data._formatters import (
IncrementalFormatter, IncrementalFormatter,
OHLCBarsFmtr, # Plain OHLC renderer OHLCBarsFmtr, # Plain OHLC renderer

38
piker/ui/_display.py
View File

@ -21,6 +21,7 @@ this module ties together quote and computational (fsp) streams with
graphics update methods via our custom ``pyqtgraph`` charting api. graphics update methods via our custom ``pyqtgraph`` charting api.
''' '''
from functools import partial
import itertools import itertools
from math import floor from math import floor
import time import time
@ -208,6 +209,7 @@ class DisplayState(Struct):
async def increment_history_view( async def increment_history_view(
# min_istream: tractor.MsgStream, # min_istream: tractor.MsgStream,
ds: DisplayState, ds: DisplayState,
loglevel: str = 'warning',
): ):
hist_chart: ChartPlotWidget = ds.hist_chart hist_chart: ChartPlotWidget = ds.hist_chart
hist_viz: Viz = ds.hist_viz hist_viz: Viz = ds.hist_viz
@ -229,7 +231,10 @@ async def increment_history_view(
hist_viz.reset_graphics() hist_viz.reset_graphics()
# hist_viz.update_graphics(force_redraw=True) # hist_viz.update_graphics(force_redraw=True)
async with open_sample_stream(1.) as min_istream: async with open_sample_stream(
period_s=1.,
loglevel=loglevel,
) as min_istream:
async for msg in min_istream: async for msg in min_istream:
profiler = Profiler( profiler = Profiler(
@ -310,7 +315,6 @@ async def increment_history_view(
async def graphics_update_loop( async def graphics_update_loop(
dss: dict[str, DisplayState], dss: dict[str, DisplayState],
nurse: trio.Nursery, nurse: trio.Nursery,
godwidget: GodWidget, godwidget: GodWidget,
@ -319,6 +323,7 @@ async def graphics_update_loop(
pis: dict[str, list[pgo.PlotItem, pgo.PlotItem]] = {}, pis: dict[str, list[pgo.PlotItem, pgo.PlotItem]] = {},
vlm_charts: dict[str, ChartPlotWidget] = {}, vlm_charts: dict[str, ChartPlotWidget] = {},
loglevel: str = 'warning',
) -> None: ) -> None:
''' '''
@ -462,9 +467,12 @@ async def graphics_update_loop(
# }) # })
nurse.start_soon( nurse.start_soon(
partial(
increment_history_view, increment_history_view,
# min_istream, # min_istream,
ds, ds=ds,
loglevel=loglevel,
),
) )
await trio.sleep(0) await trio.sleep(0)
@ -511,14 +519,19 @@ async def graphics_update_loop(
fast_chart.linked.isHidden() fast_chart.linked.isHidden()
or not rt_pi.isVisible() or not rt_pi.isVisible()
): ):
print(f'{fqme} skipping update for HIDDEN CHART') log.debug(
f'{fqme} skipping update for HIDDEN CHART'
)
fast_chart.pause_all_feeds() fast_chart.pause_all_feeds()
continue continue
ic = fast_chart.view._in_interact ic = fast_chart.view._in_interact
if ic: if ic:
fast_chart.pause_all_feeds() fast_chart.pause_all_feeds()
print(f'{fqme} PAUSING DURING INTERACTION') log.debug(
f'Pausing chart updaates during interaction\n'
f'fqme: {fqme!r}'
)
await ic.wait() await ic.wait()
fast_chart.resume_all_feeds() fast_chart.resume_all_feeds()
@ -1591,15 +1604,18 @@ async def display_symbol_data(
# start update loop task # start update loop task
dss: dict[str, DisplayState] = {} dss: dict[str, DisplayState] = {}
ln.start_soon( ln.start_soon(
partial(
graphics_update_loop, graphics_update_loop,
dss, dss=dss,
ln, nurse=ln,
godwidget, godwidget=godwidget,
feed, feed=feed,
# min_istream, # min_istream,
pis, pis=pis,
vlm_charts, vlm_charts=vlm_charts,
loglevel=loglevel,
)
) )
# boot order-mode # boot order-mode

11
piker/ui/_fsp.py
View File

@ -183,13 +183,17 @@ async def open_fsp_sidepane(
@acm @acm
async def open_fsp_actor_cluster( async def open_fsp_actor_cluster(
names: list[str] = ['fsp_0', 'fsp_1'], names: list[str] = [
'fsp_0',
'fsp_1',
],
) -> AsyncGenerator[ ) -> AsyncGenerator[
int, int,
dict[str, tractor.Portal] dict[str, tractor.Portal]
]: ]:
# TODO! change to .experimental!
from tractor._clustering import open_actor_cluster from tractor._clustering import open_actor_cluster
# profiler = Profiler( # profiler = Profiler(
@ -197,7 +201,7 @@ async def open_fsp_actor_cluster(
# disabled=False # disabled=False
# ) # )
async with open_actor_cluster( async with open_actor_cluster(
count=2, count=len(names),
names=names, names=names,
modules=['piker.fsp._engine'], modules=['piker.fsp._engine'],
@ -497,7 +501,8 @@ class FspAdmin:
portal: tractor.Portal = ( portal: tractor.Portal = (
self.cluster.get(worker_name) self.cluster.get(worker_name)
or self.rr_next_portal() or
self.rr_next_portal()
) )
# TODO: this should probably be turned into a # TODO: this should probably be turned into a

67
piker/ui/_interaction.py
View File

@ -43,6 +43,7 @@ from pyqtgraph import (
functions as fn, functions as fn,
) )
import numpy as np import numpy as np
import tractor
import trio import trio
from piker.ui.qt import ( from piker.ui.qt import (
@ -72,7 +73,10 @@ if TYPE_CHECKING:
GodWidget, GodWidget,
) )
from ._dataviz import Viz from ._dataviz import Viz
from .order_mode import OrderMode from .order_mode import (
OrderMode,
Dialog,
)
from ._display import DisplayState from ._display import DisplayState
@ -130,7 +134,12 @@ async def handle_viewmode_kb_inputs(
async for kbmsg in recv_chan: async for kbmsg in recv_chan:
event, etype, key, mods, text = kbmsg.to_tuple() event, etype, key, mods, text = kbmsg.to_tuple()
log.debug(f'key: {key}, mods: {mods}, text: {text}') log.debug(
f'View-mode kb-msg received,\n'
f'mods: {mods!r}\n'
f'key: {key!r}\n'
f'text: {text!r}\n'
)
now = time.time() now = time.time()
period = now - last period = now - last
@ -158,8 +167,12 @@ async def handle_viewmode_kb_inputs(
# have no previous keys or we do and the min_tap period is # have no previous keys or we do and the min_tap period is
# met # met
if ( if (
not fast_key_seq or not fast_key_seq
period <= min_tap and fast_key_seq or (
period <= min_tap
and
fast_key_seq
)
): ):
fast_key_seq.append(text) fast_key_seq.append(text)
log.debug(f'fast keys seqs {fast_key_seq}') log.debug(f'fast keys seqs {fast_key_seq}')
@ -174,7 +187,8 @@ async def handle_viewmode_kb_inputs(
# UI REPL-shell, with ctrl-p (for "pause") # UI REPL-shell, with ctrl-p (for "pause")
if ( if (
ctrl ctrl
and key in { and
key in {
Qt.Key_P, Qt.Key_P,
} }
): ):
@ -184,7 +198,6 @@ async def handle_viewmode_kb_inputs(
vlm_chart = chart.linked.subplots['volume'] # noqa vlm_chart = chart.linked.subplots['volume'] # noqa
vlm_viz = vlm_chart.main_viz # noqa vlm_viz = vlm_chart.main_viz # noqa
dvlm_pi = vlm_chart._vizs['dolla_vlm'].plot # noqa dvlm_pi = vlm_chart._vizs['dolla_vlm'].plot # noqa
import tractor
await tractor.pause() await tractor.pause()
view.interact_graphics_cycle() view.interact_graphics_cycle()
@ -192,7 +205,8 @@ async def handle_viewmode_kb_inputs(
# shown data `Viz`s for the current chart app. # shown data `Viz`s for the current chart app.
if ( if (
ctrl ctrl
and key in { and
key in {
Qt.Key_R, Qt.Key_R,
} }
): ):
@ -231,7 +245,8 @@ async def handle_viewmode_kb_inputs(
key == Qt.Key_Escape key == Qt.Key_Escape
or ( or (
ctrl ctrl
and key == Qt.Key_C and
key == Qt.Key_C
) )
): ):
# ctrl-c as cancel # ctrl-c as cancel
@ -242,17 +257,35 @@ async def handle_viewmode_kb_inputs(
# cancel order or clear graphics # cancel order or clear graphics
if ( if (
key == Qt.Key_C key == Qt.Key_C
or key == Qt.Key_Delete or
key == Qt.Key_Delete
): ):
# log.info('Handling <c> hotkey!')
try:
dialogs: list[Dialog] = order_mode.cancel_orders_under_cursor()
except BaseException:
log.exception('Failed to cancel orders !?\n')
await tractor.pause()
order_mode.cancel_orders_under_cursor() if not dialogs:
log.warning(
goodboy commented 2026-02-23 18:33:06 +00:00
Review

This was originally bc i couldn’t figure out why on sway i couldn’t mouse-over-c-cancel orders XD

Turns out it was bc i had keyboard-disabled-while-typing set.

Ideally we can somehow detect this kinda thing more fancy like in the longer term. I just haven’t dug into it yet..

This was originally bc i couldn't figure out why on `sway` i couldn't mouse-over-`c`-cancel orders XD Turns out it was bc i had keyboard-disabled-while-typing set. Ideally we can somehow detect this kinda thing more fancy like in the longer term. I just haven't dug into it yet..
'No orders were cancelled?\n'
'Is there an order-line under the cursor?\n'
'If you think there IS your DE might be "hiding the mouse" before '
'we rx the keyboard input via Qt..\n'
'=> Check your DE and/or TWM settings to be sure! <=\n'
)
# ^TODO?, some way to detect if there's lines and
# the DE is cuckin with things?
# await tractor.pause()
# View modes # View modes
if ( if (
ctrl ctrl
and ( and (
key == Qt.Key_Equal key == Qt.Key_Equal
or key == Qt.Key_I or
key == Qt.Key_I
) )
): ):
view.wheelEvent( view.wheelEvent(
@ -264,7 +297,8 @@ async def handle_viewmode_kb_inputs(
ctrl ctrl
and ( and (
key == Qt.Key_Minus key == Qt.Key_Minus
or key == Qt.Key_O or
key == Qt.Key_O
) )
): ):
view.wheelEvent( view.wheelEvent(
@ -275,7 +309,8 @@ async def handle_viewmode_kb_inputs(
elif ( elif (
not ctrl not ctrl
and key == Qt.Key_R and
key == Qt.Key_R
): ):
# NOTE: seems that if we don't yield a Qt render # NOTE: seems that if we don't yield a Qt render
# cycle then the m4 downsampled curves will show here # cycle then the m4 downsampled curves will show here
@ -477,7 +512,8 @@ async def handle_viewmode_mouse(
# view.raiseContextMenu(event) # view.raiseContextMenu(event)
if ( if (
view.order_mode.active and view.order_mode.active
and
button == QtCore.Qt.LeftButton button == QtCore.Qt.LeftButton
): ):
# when in order mode, submit execution # when in order mode, submit execution
@ -781,7 +817,8 @@ class ChartView(ViewBox):
# Scale or translate based on mouse button # Scale or translate based on mouse button
if btn & ( if btn & (
QtCore.Qt.LeftButton | QtCore.Qt.MidButton QtCore.Qt.LeftButton
| QtCore.Qt.MidButton
): ):
# zoom y-axis ONLY when click-n-drag on it # zoom y-axis ONLY when click-n-drag on it
# if axis == 1: # if axis == 1:

20
piker/ui/_lines.py
View File

@ -52,10 +52,13 @@ from ._anchors import (
from ..calc import humanize from ..calc import humanize
from ._label import Label from ._label import Label
from ._style import hcolor, _font from ._style import hcolor, _font
from ..log import get_logger
if TYPE_CHECKING: if TYPE_CHECKING:
from ._cursor import Cursor from ._cursor import Cursor
log = get_logger(__name__)
# TODO: probably worth investigating if we can # TODO: probably worth investigating if we can
# make .boundingRect() faster: # make .boundingRect() faster:
@ -347,7 +350,7 @@ class LevelLine(pg.InfiniteLine):
) -> None: ) -> None:
# TODO: enter labels edit mode # TODO: enter labels edit mode
print(f'double click {ev}') log.debug(f'double click {ev}')
def paint( def paint(
self, self,
@ -461,10 +464,19 @@ class LevelLine(pg.InfiniteLine):
# hovered # hovered
if ( if (
not ev.isExit() not ev.isExit()
and ev.acceptDrags(QtCore.Qt.LeftButton) and
ev.acceptDrags(QtCore.Qt.LeftButton)
): ):
# if already hovered we don't need to run again # if already hovered we don't need to run again
if self.mouseHovering is True: if (
self.mouseHovering is True
and
cur.is_hovered(self)
):
log.debug(
f'Already hovering ??\n'
f'cur._hovered: {cur._hovered!r}\n'
)
return return
if self.only_show_markers_on_hover: if self.only_show_markers_on_hover:
@ -481,6 +493,7 @@ class LevelLine(pg.InfiniteLine):
cur._y_label_update = False cur._y_label_update = False
# add us to cursor state # add us to cursor state
log.debug(f'Adding line {self!r}\n')
cur.add_hovered(self) cur.add_hovered(self)
if self._hide_xhair_on_hover: if self._hide_xhair_on_hover:
@ -508,6 +521,7 @@ class LevelLine(pg.InfiniteLine):
self.currentPen = self.pen self.currentPen = self.pen
log.debug(f'Removing line {self!r}\n')
cur._hovered.remove(self) cur._hovered.remove(self)
if self.only_show_markers_on_hover: if self.only_show_markers_on_hover:

10
piker/ui/_widget.py
View File

@ -300,7 +300,10 @@ class GodWidget(QWidget):
getattr(widget, 'on_resize') getattr(widget, 'on_resize')
self._widgets[widget.mode_name] = widget self._widgets[widget.mode_name] = widget
def on_win_resize(self, event: QtCore.QEvent) -> None: def on_win_resize(
self,
event: QtCore.QEvent,
) -> None:
''' '''
Top level god widget handler from window (the real yaweh) resize Top level god widget handler from window (the real yaweh) resize
events such that any registered widgets which wish to be events such that any registered widgets which wish to be
@ -315,7 +318,10 @@ class GodWidget(QWidget):
self._resizing = True self._resizing = True
log.info('God widget resize') log.debug(
f'God widget resize\n'
f'{event}\n'
)
for name, widget in self._widgets.items(): for name, widget in self._widgets.items():
widget.on_resize() widget.on_resize()

10
piker/ui/_window.py
View File

@ -255,8 +255,16 @@ class MainWindow(QMainWindow):
current: QWidget, current: QWidget,
) -> None: ) -> None:
'''
Focus handler.
log.info(f'widget focus changed from {last} -> {current}') For now updates the "current mode" name.
'''
log.debug(
f'widget focus changed from,\n'
f'{last} -> {current}'
)
if current is not None: if current is not None:
# cursor left window? # cursor left window?

3
piker/ui/cli.py
View File

@ -177,7 +177,7 @@ def chart(
return return
# global opts # global opts
brokernames = config['brokers'] # brokernames: list[str] = config['brokers']
brokermods = config['brokermods'] brokermods = config['brokermods']
assert brokermods assert brokermods
tractorloglevel = config['tractorloglevel'] tractorloglevel = config['tractorloglevel']
@ -216,6 +216,7 @@ def chart(
layers['tcp']['port'], layers['tcp']['port'],
)) ))
# breakpoint()
from tractor.devx import maybe_open_crash_handler from tractor.devx import maybe_open_crash_handler
pdb: bool = config['pdb'] pdb: bool = config['pdb']
with maybe_open_crash_handler(pdb=pdb): with maybe_open_crash_handler(pdb=pdb):

41
piker/ui/order_mode.py
View File

@ -77,7 +77,6 @@ from ._style import _font
from ._forms import open_form_input_handling from ._forms import open_form_input_handling
from ._notify import notify_from_ems_status_msg from ._notify import notify_from_ems_status_msg
if TYPE_CHECKING: if TYPE_CHECKING:
from ._chart import ( from ._chart import (
ChartPlotWidget, ChartPlotWidget,
@ -436,7 +435,7 @@ class OrderMode:
lines=lines, lines=lines,
last_status_close=self.multistatus.open_status( last_status_close=self.multistatus.open_status(
f'submitting {order.exec_mode}-{order.action}', f'submitting {order.exec_mode}-{order.action}',
final_msg=f'submitted {order.exec_mode}-{order.action}', # final_msg=f'submitted {order.exec_mode}-{order.action}',
clear_on_next=True, clear_on_next=True,
) )
) )
@ -514,13 +513,14 @@ class OrderMode:
def on_submit( def on_submit(
self, self,
uuid: str, uuid: str,
order: Order | None = None, order: Order|None = None,
) -> Dialog | None: ) -> Dialog|None:
''' '''
Order submitted status event handler. Order submitted status event handler.
Commit the order line and registered order uuid, store ack time stamp. Commit the order line and registered order uuid, store ack
time stamp.
''' '''
lines = self.lines.commit_line(uuid) lines = self.lines.commit_line(uuid)
@ -528,7 +528,7 @@ class OrderMode:
# a submission is the start of a new order dialog # a submission is the start of a new order dialog
dialog = self.dialogs[uuid] dialog = self.dialogs[uuid]
dialog.lines = lines dialog.lines = lines
cls: Callable | None = dialog.last_status_close cls: Callable|None = dialog.last_status_close
if cls: if cls:
cls() cls()
@ -658,7 +658,7 @@ class OrderMode:
return True return True
def cancel_orders_under_cursor(self) -> list[str]: def cancel_orders_under_cursor(self) -> list[Dialog]:
return self.cancel_orders( return self.cancel_orders(
self.oids_from_lines( self.oids_from_lines(
self.lines.lines_under_cursor() self.lines.lines_under_cursor()
@ -687,24 +687,28 @@ class OrderMode:
self, self,
oids: list[str], oids: list[str],
) -> None: ) -> list[Dialog]:
''' '''
Cancel all orders from a list of order ids: `oids`. Cancel all orders from a list of order ids: `oids`.
''' '''
key = self.multistatus.open_status( # key = self.multistatus.open_status(
goodboy commented 2026-02-23 18:34:15 +00:00
Review

Oh yeah right, and i disabled all these “status” things bc they never seem to clear from the bottom of the main widget/window..

will try to see how we can do em better in the future.

Oh yeah right, and i disabled all these "status" things bc they never seem to clear from the bottom of the main widget/window.. will try to see how we can do em better in the future.
f'cancelling {len(oids)} orders', # f'cancelling {len(oids)} orders',
final_msg=f'cancelled orders:\n{oids}', # final_msg=f'cancelled orders:\n{oids}',
group_key=True # group_key=True
) # )
dialogs: list[Dialog] = []
for oid in oids: for oid in oids:
if dialog := self.dialogs.get(oid): if dialog := self.dialogs.get(oid):
self.client.cancel_nowait(uuid=oid) self.client.cancel_nowait(uuid=oid)
cancel_status_close = self.multistatus.open_status( # cancel_status_close = self.multistatus.open_status(
f'cancelling order {oid}', # f'cancelling order {oid}',
group_key=key, # group_key=key,
) # )
dialog.last_status_close = cancel_status_close # dialog.last_status_close = cancel_status_close
dialogs.append(dialog)
return dialogs
def cancel_all_orders(self) -> None: def cancel_all_orders(self) -> None:
''' '''
@ -776,7 +780,6 @@ class OrderMode:
@asynccontextmanager @asynccontextmanager
async def open_order_mode( async def open_order_mode(
feed: Feed, feed: Feed,
godw: GodWidget, godw: GodWidget,
fqme: str, fqme: str,

6
uv.lock
View File

@ -1159,11 +1159,11 @@ uis = [
[[package]] [[package]]
name = "platformdirs" name = "platformdirs"
version = "4.5.1" version = "4.6.0"
source = { registry = "https://pypi.org/simple" } source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/cf/86/0248f086a84f01b37aaec0fa567b397df1a119f73c16f6c7a9aac73ea309/platformdirs-4.5.1.tar.gz", hash = "sha256:61d5cdcc6065745cdd94f0f878977f8de9437be93de97c1c12f853c9c0cdcbda", size = 21715, upload-time = "2025-12-05T13:52:58.638Z" } sdist = { url = "https://files.pythonhosted.org/packages/20/e5/474d0a8508029286b905622e6929470fb84337cfa08f9d09fbb624515249/platformdirs-4.6.0.tar.gz", hash = "sha256:4a13c2db1071e5846c3b3e04e5b095c0de36b2a24be9a3bc0145ca66fce4e328", size = 23433, upload-time = "2026-02-12T14:36:21.288Z" }
wheels = [ wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/28/3bfe2fa5a7b9c46fe7e13c97bda14c895fb10fa2ebf1d0abb90e0cea7ee1/platformdirs-4.5.1-py3-none-any.whl", hash = "sha256:d03afa3963c806a9bed9d5125c8f4cb2fdaf74a55ab60e5d59b3fde758104d31", size = 18731, upload-time = "2025-12-05T13:52:56.823Z" }, { url = "https://files.pythonhosted.org/packages/da/10/1b0dcf51427326f70e50d98df21b18c228117a743a1fc515a42f8dc7d342/platformdirs-4.6.0-py3-none-any.whl", hash = "sha256:dd7f808d828e1764a22ebff09e60f175ee3c41876606a6132a688d809c7c9c73", size = 19549, upload-time = "2026-02-12T14:36:19.743Z" },
] ]
[[package]] [[package]]