diff --git a/piker/brokers/ib/_util.py b/piker/brokers/ib/_util.py index 2c71bc46..512e5358 100644 --- a/piker/brokers/ib/_util.py +++ b/piker/brokers/ib/_util.py @@ -34,6 +34,7 @@ from piker.brokers._util import get_logger if TYPE_CHECKING: from .api import Client from ib_insync import IB + import i3ipc log = get_logger('piker.brokers.ib') @@ -48,6 +49,37 @@ _reset_tech: Literal[ ] = 'vnc' +no_setup_msg:str = ( + 'No data reset hack test setup for {vnc_sockaddr}!\n' + 'See config setup tips @\n' + 'https://github.com/pikers/piker/tree/master/piker/brokers/ib' +) + + +def try_xdo_manual( + vnc_sockaddr: str, +): + ''' + Do the "manual" `xdo`-based screen switch + click + combo since apparently the `asyncvnc` client ain't workin.. + + Note this is only meant as a backup method for Xorg users, + ideally you can use a real vnc client and the `vnc_click_hack()` + impl! + + ''' + global _reset_tech + try: + i3ipc_xdotool_manual_click_hack() + _reset_tech = 'i3ipc_xdotool' + return True + except OSError: + log.exception( + no_setup_msg.format(vnc_sockaddr) + ) + return False + + async def data_reset_hack( # vnc_host: str, client: Client, @@ -90,15 +122,9 @@ async def data_reset_hack( vnc_port: int vnc_sockaddr: tuple[str] | None = client.conf.get('vnc_addrs') - no_setup_msg:str = ( - f'No data reset hack test setup for {vnc_sockaddr}!\n' - 'See config setup tips @\n' - 'https://github.com/pikers/piker/tree/master/piker/brokers/ib' - ) - if not vnc_sockaddr: log.warning( - no_setup_msg + no_setup_msg.format(vnc_sockaddr) + 'REQUIRES A `vnc_addrs: array` ENTRY' ) @@ -119,27 +145,38 @@ async def data_reset_hack( port=vnc_port, ) ) - except OSError: - if vnc_host != 'localhost': - log.warning(no_setup_msg) - return False - + except ( + OSError, # no VNC server avail.. + PermissionError, # asyncvnc pw fail.. + ): try: import i3ipc # noqa (since a deps dynamic check) except ModuleNotFoundError: - log.warning(no_setup_msg) + log.warning( + no_setup_msg.format(vnc_sockaddr) + ) return False - try: - i3ipc_xdotool_manual_click_hack() - _reset_tech = 'i3ipc_xdotool' - return True - except OSError: - log.exception(no_setup_msg) - return False + if vnc_host not in { + 'localhost', + '127.0.0.1', + }: + focussed, matches = i3ipc_fin_wins_titled() + if not matches: + log.warning( + no_setup_msg.format(vnc_sockaddr) + ) + return False + else: + try_xdo_manual(vnc_sockaddr) + + # localhost but no vnc-client or it borked.. + else: + try_xdo_manual(vnc_sockaddr) case 'i3ipc_xdotool': - i3ipc_xdotool_manual_click_hack() + try_xdo_manual(vnc_sockaddr) + # i3ipc_xdotool_manual_click_hack() case _ as tech: raise RuntimeError(f'{tech} is not supported for reset tech!?') @@ -178,9 +215,9 @@ async def vnc_click_hack( host, port=port, - # TODO: doesn't work see: - # https://github.com/barneygale/asyncvnc/issues/7 - # password='ibcansmbz', + # TODO: doesn't work? + # see, https://github.com/barneygale/asyncvnc/issues/7 + password='doggy', ) as client: @@ -194,70 +231,103 @@ async def vnc_click_hack( client.keyboard.press('Ctrl', 'Alt', key) # keys are stacked +def i3ipc_fin_wins_titled( + titles: list[str] = [ + 'Interactive Brokers', # tws running in i3 + 'IB Gateway', # gw running in i3 + # 'IB', # gw running in i3 (newer version?) + + # !TODO, remote vnc instance + # -[ ] something in title (or other Con-props) that indicates + # this is explicitly for ibrk sw? + # |_[ ] !can use modden spawn eventually! + 'TigerVNC', + # 'vncviewer', # the terminal.. + ], +) -> tuple[ + i3ipc.Con, # orig focussed win + list[tuple[str, i3ipc.Con]], # matching wins by title +]: + ''' + Attempt to find a local-DE window titled with an entry in + `titles`. + + If found deliver the current focussed window and all matching + `i3ipc.Con`s in a list. + + ''' + import i3ipc + ipc = i3ipc.Connection() + + # TODO: might be worth offering some kinda api for grabbing + # the window id from the pid? + # https://stackoverflow.com/a/2250879 + tree = ipc.get_tree() + focussed: i3ipc.Con = tree.find_focused() + + matches: list[i3ipc.Con] = [] + for name in titles: + results = tree.find_titled(name) + print(f'results for {name}: {results}') + if results: + con = results[0] + matches.append(( + name, + con, + )) + + return ( + focussed, + matches, + ) + + + def i3ipc_xdotool_manual_click_hack() -> None: ''' Do the data reset hack but expecting a local X-window using `xdotool`. ''' - import i3ipc - i3 = i3ipc.Connection() - - # TODO: might be worth offering some kinda api for grabbing - # the window id from the pid? - # https://stackoverflow.com/a/2250879 - t = i3.get_tree() - - orig_win_id = t.find_focused().window - - # for tws - win_names: list[str] = [ - 'Interactive Brokers', # tws running in i3 - 'IB Gateway', # gw running in i3 - # 'IB', # gw running in i3 (newer version?) - ] - + focussed, matches = i3ipc_fin_wins_titled() + orig_win_id = focussed.window try: - for name in win_names: - results = t.find_titled(name) - print(f'results for {name}: {results}') - if results: - con = results[0] - print(f'Resetting data feed for {name}') - win_id = str(con.window) - w, h = con.rect.width, con.rect.height + for name, con in matches: + print(f'Resetting data feed for {name}') + win_id = str(con.window) + w, h = con.rect.width, con.rect.height - # TODO: seems to be a few libs for python but not sure - # if they support all the sub commands we need, order of - # most recent commit history: - # https://github.com/rr-/pyxdotool - # https://github.com/ShaneHutter/pyxdotool - # https://github.com/cphyc/pyxdotool + # TODO: seems to be a few libs for python but not sure + # if they support all the sub commands we need, order of + # most recent commit history: + # https://github.com/rr-/pyxdotool + # https://github.com/ShaneHutter/pyxdotool + # https://github.com/cphyc/pyxdotool - # TODO: only run the reconnect (2nd) kc on a detected - # disconnect? - for key_combo, timeout in [ - # only required if we need a connection reset. - # ('ctrl+alt+r', 12), - # data feed reset. - ('ctrl+alt+f', 6) - ]: - subprocess.call([ - 'xdotool', - 'windowactivate', '--sync', win_id, + # TODO: only run the reconnect (2nd) kc on a detected + # disconnect? + for key_combo, timeout in [ + # only required if we need a connection reset. + # ('ctrl+alt+r', 12), + # data feed reset. + ('ctrl+alt+f', 6) + ]: + subprocess.call([ + 'xdotool', + 'windowactivate', '--sync', win_id, - # move mouse to bottom left of window (where - # there should be nothing to click). - 'mousemove_relative', '--sync', str(w-4), str(h-4), + # move mouse to bottom left of window (where + # there should be nothing to click). + 'mousemove_relative', '--sync', str(w-4), str(h-4), - # NOTE: we may need to stick a `--retry 3` in here.. - 'click', '--window', win_id, - '--repeat', '3', '1', + # NOTE: we may need to stick a `--retry 3` in here.. + 'click', '--window', win_id, + '--repeat', '3', '1', - # hackzorzes - 'key', key_combo, - ], - timeout=timeout, - ) + # hackzorzes + 'key', key_combo, + ], + timeout=timeout, + ) # re-activate and focus original window subprocess.call([