fxreader.online/freelance-project-34-marketing-blog
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

[~] Refactor

Browse Source
This commit is contained in:
Siarhei Siniak 2024-11-10 21:49:33 +03:00
parent 40350d128a
commit 14b499637b
1 changed files with 34 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
dotfiles/.local/bin/commands
View File

@ -23,7 +23,7 @@ import tempfile
import time import time
import traceback import traceback
from typing import (Literal, Optional, TypedDict, Callable, Generator, TypeAlias,) from typing import (Literal, Optional, TypedDict, Callable, Generator, TypeAlias, Any,)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -258,8 +258,8 @@ def memory_stats() -> memory_stats_t.res_t:
raise NotImplementedError raise NotImplementedError
def chrome( def chrome(
argv argv: list[str]
): ) -> int:
assert isinstance(argv, list) and all([isinstance(o, str) for o in argv]) assert isinstance(argv, list) and all([isinstance(o, str) for o in argv])
parser = optparse.OptionParser() parser = optparse.OptionParser()
parser.add_option( parser.add_option(
@ -281,7 +281,7 @@ def chrome(
#assert os.path.exists(options.user_data_dir) #assert os.path.exists(options.user_data_dir)
if sys.platform == 'linux': if sys.platform == 'linux':
subprocess.check_call([ return subprocess.check_call([
'google-chrome-stable', 'google-chrome-stable',
'--enable-features=useOzonePlatform', '--enable-features=useOzonePlatform',
'--ozone-platform=wayland', '--ozone-platform=wayland',
@ -292,7 +292,7 @@ def chrome(
else: else:
raise NotImplementedError raise NotImplementedError
def eternal_oom(argv): def eternal_oom(argv: list[str]) -> None:
import signal import signal
import re import re
import time import time
@ -412,12 +412,15 @@ def eternal_oom(argv):
) )
return return
cpu_count = os.cpu_count()
assert isinstance(cpu_count, int)
if options.period is None: if options.period is None:
options.period = 1 options.period = 1
if options.memory_limit is None: if options.memory_limit is None:
options.memory_limit = 3 * 1024 * 1024 options.memory_limit = 3 * 1024 * 1024
if options.cpu_limit is None: if options.cpu_limit is None:
options.cpu_limit = 0.6 * os.cpu_count() options.cpu_limit = 0.6 * cpu_count
if options.cpu_wait is None: if options.cpu_wait is None:
options.cpu_wait = 10 options.cpu_wait = 10
if options.mean_size is None: if options.mean_size is None:
@ -431,24 +434,40 @@ def eternal_oom(argv):
and options.memory_limit > 512 * 1024 and options.memory_limit > 512 * 1024
assert isinstance(options.cpu_limit, float) \ assert isinstance(options.cpu_limit, float) \
and options.cpu_limit > 0.2 * os.cpu_count() and \ and options.cpu_limit > 0.2 * cpu_count and \
options.cpu_limit < os.cpu_count() * 0.95 options.cpu_limit < cpu_count * 0.95
assert options.period >= 1 assert options.period >= 1
assert options.cpu_wait >= 10 assert options.cpu_wait >= 10
assert options.mean_size >= 16 assert options.mean_size >= 16
def pandas_data_frame(lines, groups_regex, header_regex, extra_columns): def pandas_data_frame(
header = re.compile(header_regex).search(lines[0]).groups() lines: list[str],
rows = [ groups_regex: re.Pattern[str],
re.compile(groups_regex).search(row).groups() header_regex: re.Pattern[str],
for row in lines[1:] extra_columns: dict[
] str,
columns = { Callable[
[dict[str, str]],
Any
]
],
) -> dict[str, list[Any]]:
header_match = re.compile(header_regex).search(lines[0])
assert not header_match is None
header = header_match.groups()
rows = []
for line in lines[1:]:
row_match = re.compile(groups_regex).search(line)
assert not row_match is None
rows.append(row_match.groups())
columns: dict[str, list[Any]] = {
column: [] column: []
for column in header for column in header
} }
for row in rows: for row in rows:
for value, column in zip(row, header): for value, column in zip(row, header):
columns[column].append(value) columns[column].append(value)