diff --git a/dotfiles/.local/bin/commands b/dotfiles/.local/bin/commands index 50e5360..7fce9e1 100755 --- a/dotfiles/.local/bin/commands +++ b/dotfiles/.local/bin/commands @@ -24,7 +24,10 @@ import tempfile import time import traceback -from typing import (Literal, Optional, TypedDict, Callable, Generator, TypeAlias, Any,) +from typing import ( + Literal, Optional, TypedDict, Callable, Generator, TypeAlias, Any, + cast, +) logger = logging.getLogger(__name__) @@ -155,8 +158,8 @@ def intercept_output( buffer.clear() buffer.append(t3[t3_pos + 1:]) while len(buffer_lines) > 0: - yield dict( - aggegated=False, + yield intercept_output_t.line_res_t( + aggregated=False, line=buffer_lines.popleft(), ) @@ -293,6 +296,9 @@ def chrome( else: raise NotImplementedError +def raise_not_implemented() -> None: + raise NotImplementedError + def eternal_oom(argv: list[str]) -> None: import signal import re @@ -485,14 +491,24 @@ def eternal_oom(argv: list[str]) -> None: return columns - def pandas_merge(left, right, on): - index = {} - input_data_frames = [ + def pandas_merge( + left: dict[str, list[Any]], + right: dict[str, list[Any]], + on: str, + ) -> dict[str, list[Any]]: + index : dict[str, dict[Any, list[int]]] = {} + + input_data_frames : list[ + tuple[ + str, + dict[str, list[Any]] + ] + ] = [ ('left', left), ('right', right), ] for index_name, data_frame in input_data_frames: - current_index = {} + current_index: dict[Any, list[int]] = {} for row_index, value in enumerate(data_frame[on]): if not value in current_index: current_index[value] = [] @@ -500,7 +516,11 @@ def eternal_oom(argv: list[str]) -> None: index[index_name] = current_index - merged_data_frame = dict( + class MergedDataFrame(TypedDict): + header: list[str] + columns: dict[str, list[Any]] + + merged_data_frame: MergedDataFrame = dict( header=[ column + '_x' for column in left @@ -514,12 +534,16 @@ def eternal_oom(argv: list[str]) -> None: for column in merged_data_frame['header']: merged_data_frame['columns'][column] = [] - common_values = { + common_values: set[Any] = { left_value for left_value in index['left'] if left_value in index['right'] } - common_rows = sorted( + class RowMatch(TypedDict): + left_row_index: int + right_row_index: int + + common_rows: list[RowMatch] = sorted( [ dict( left_row_index=index['left'][value][0], @@ -533,7 +557,17 @@ def eternal_oom(argv: list[str]) -> None: row = sum([ [ values[ - common_row['%s_row_index' % index_name] + common_row[ + cast( + Literal['left_row_index' | 'right_row_index'], + 'left_row_index' + if index_name == 'left' + else + 'right_row_index' + if index_name == 'right' + else raise_not_implemented() + ) + ] ] for column, values in data_frame.items() ] @@ -3794,7 +3828,12 @@ class Command(enum.StrEnum): vpn = 'vpn' backup = 'backup' -def commands_cli() -> None: +def commands_cli( + argv: Optional[list[str]] = None +) -> None: + if argv is None: + argv = sys.argv[1:] + logging.getLogger().setLevel(logging.INFO) logger.setLevel(logging.INFO) handler = logging.StreamHandler(sys.stderr) @@ -3802,80 +3841,81 @@ def commands_cli() -> None: msg : Optional[str] = None - parser = argparse.ArgumentParser('online_fxreader.commands') - parser.add_argument( - 'command', - choices=[ - o.value - for o in Command - ], - ) - - options, args = parser.parse_known_args() - - options.command = Command(options.command) - try: - if len(args) > 0 and args[0].startswith('media'): - msg = media_keys(args).get('msg') - elif options.command is Command.status: - status(args) - elif options.command is Command.http_server: - http_server(args) - elif options.command is Command.pass_ssh_osx: - pass_ssh_osx(args) - elif options.command is Command.wl_screenshot: - subprocess.check_call(r''' - grim -g "$(slurp)" - | wl-copy - ''', shell=True) - elif options.command is Command.chrome: - chrome(args) - elif options.command is Command.eternal_oom: - eternal_oom(args) - elif options.command is Command.resilient_vlc: - resilient_vlc(args) - elif options.command is Command.eternal_firefox: - eternal_firefox( - profile=sys.argv[2], - group_name=sys.argv[3], - window_position=json.loads(sys.argv[4]), - debug=json.loads(sys.argv[5]), - tabs=sys.argv[6:], - ) - elif options.command is Command.install: - install(args) - elif options.command is Command.resilient_ethernet: - resilient_ethernet( - ip_addr=sys.argv[2], - ethernet_device=sys.argv[3], - ) - elif options.command is Command.player: - player_v1( - folder_url=sys.argv[2], - item_id=int(sys.argv[3]), - ) - elif options.command is Command.share_wifi: - share_wifi(args) - elif options.command is Command.socat_ssh: - socat_ssh(args) - elif options.command is Command.gnome_shortcuts: - gnome_shortcuts(args) - elif options.command is Command.sway_sock: - print(sway_sock()) - elif options.command is Command.suspend_timer: - suspend_timer(args) - elif options.command is Command.desktop_services: - desktop_services(args) - elif options.command is Command.pm_service: - pm_service(args) - elif options.command is Command.backup: - backup(args) - elif options.command is Command.scrap_yt_music: - scrap_yt_music(args) - elif options.command is Command.vpn: - vpn(args) + if len(argv) > 0 and argv[0].startswith('media'): + msg = media_keys(argv).get('msg') else: - raise NotImplementedError + parser = argparse.ArgumentParser('online_fxreader.commands') + parser.add_argument( + '_command', + choices=[ + o.value + for o in Command + ], + ) + + options, args = parser.parse_known_args() + options.command = Command(options._command) + + + if options.command is Command.status: + status(args) + elif options.command is Command.http_server: + http_server(args) + elif options.command is Command.pass_ssh_osx: + pass_ssh_osx(args) + elif options.command is Command.wl_screenshot: + subprocess.check_call(r''' + grim -g "$(slurp)" - | wl-copy + ''', shell=True) + elif options.command is Command.chrome: + chrome(args) + elif options.command is Command.eternal_oom: + eternal_oom(args) + elif options.command is Command.resilient_vlc: + resilient_vlc(args) + elif options.command is Command.eternal_firefox: + eternal_firefox( + profile=sys.argv[2], + group_name=sys.argv[3], + window_position=json.loads(sys.argv[4]), + debug=json.loads(sys.argv[5]), + tabs=sys.argv[6:], + ) + elif options.command is Command.install: + install(args) + elif options.command is Command.resilient_ethernet: + resilient_ethernet( + ip_addr=sys.argv[2], + ethernet_device=sys.argv[3], + ) + elif options.command is Command.player: + player_v1( + folder_url=sys.argv[2], + item_id=int(sys.argv[3]), + ) + elif options.command is Command.share_wifi: + share_wifi(args) + elif options.command is Command.socat_ssh: + socat_ssh(args) + elif options.command is Command.gnome_shortcuts: + gnome_shortcuts(args) + elif options.command is Command.sway_sock: + print(sway_sock()) + elif options.command is Command.suspend_timer: + suspend_timer(args) + elif options.command is Command.desktop_services: + desktop_services(args) + elif options.command is Command.pm_service: + pm_service(args) + elif options.command is Command.backup: + backup(args) + elif options.command is Command.scrap_yt_music: + scrap_yt_music(args) + elif options.command is Command.vpn: + vpn(args) + else: + raise NotImplementedError except SystemExit: pass except Exception: