From 8656b3b98589c3ac54a9f192d115ce891a3b53e7 Mon Sep 17 00:00:00 2001
From: Siarhei Siniak <siarheisiniak@gmail.com>
Date: Thu, 28 Nov 2024 12:24:38 +0300
Subject: [PATCH] [~] Refactor
---
dotfiles/.local/bin/commands | 208 +++++++++++++++++++++--------------
1 file changed, 124 insertions(+), 84 deletions(-)
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: