diff --git a/python/meson.build b/python/meson.build index 805b2e4..a5b3dad 100644 --- a/python/meson.build +++ b/python/meson.build @@ -5,7 +5,7 @@ project( ).stdout().strip('\n'), # 'online.fxreader.uv', # ['c', 'cpp'], - version: '0.1.5.42', + version: '0.1.5.43', # default_options: [ # 'cpp_std=c++23', # # 'prefer_static=true', diff --git a/python/online/fxreader/pr34/commands.py b/python/online/fxreader/pr34/commands.py index 3399ae0..3cfdb49 100644 --- a/python/online/fxreader/pr34/commands.py +++ b/python/online/fxreader/pr34/commands.py @@ -3552,119 +3552,6 @@ def share_wifi(argv): time.sleep(1) -def status(argv): - import inspect - import textwrap - - assert isinstance(argv, list) and all([isinstance(o, str) for o in argv]) - - class c1(optparse.IndentedHelpFormatter): - def format_option(self, *args, **kwargs): - f1 = lambda text, width: '\n'.join([textwrap.fill('\t' + o, width, replace_whitespace=False) for o in text.splitlines()]).splitlines() - t1 = inspect.getsource(optparse.IndentedHelpFormatter.format_option) - t2 = ( - '\n'.join([o[4:] for o in t1.splitlines()[:]]) - .replace( - 'textwrap.wrap', - 'f1', - ) - .replace('format_option', 'f2') - ) - exec(t2, dict(f1=f1), locals()) - return locals()['f2'](self, *args, **kwargs) - - parser = optparse.OptionParser( - formatter=c1( - width=None, - ), - ) - parser.add_option( - '--sh', - dest='sh', - default=[], - action='append', - type=str, - ) - parser.add_option( - '--timeout', - dest='timeout', - default=None, - type=float, - ) - parser.add_option( - '--config', - dest='config', - default=None, - type=str, - help=''.join( - [ - '.json file with array of strings, each is a shell command ', - 'that outputs a separate status text value, ', - 'like\n', - r""" -ping -w 1 -i 0.02 -c 3 | tail -n 2| head -n 1 | grep -Po $'time\\s+.*$' -sensors -j | jq -r '.\"coretemp-isa-0000\".\"Package id 0\".temp1_input|tostring + \" C\"' -printf '%d RPM' $(cat /sys/devices/platform/applesmc.768/fan1_input) -printf '% 3.0f%%' $(upower -d | grep -Po 'percentage:\\s+\\d+(\\.\\d+)?%' | grep -Po '\\d+(\\.\\d+)?' | head -n 1) - """.strip(), - ] - ), - ) - options, args = parser.parse_args(argv) - - if options.timeout is None: - options.timeout = 0.5 - - timeout2 = max(options.timeout, 0.0) - - assert timeout2 >= 0.0 and timeout2 <= 4 - - config = dict() - try: - if not options.config is None: - with io.open(options.config, 'r') as f: - config.update(json.load(f)) - except Exception: - logging.error(traceback.format_exc()) - pass - - options.sh.extend(config.get('sh', [])) - - t1 = [] - - for sh_index, o in enumerate( - [ - *options.sh, - *[ - r""" - A=$(free -h | grep -P Mem: | grep -Po '[\w\.\d]+'); - echo -n $A | awk '{print $2, $7}'; - """, - r""" - date +'%Y-%m-%d %l:%M:%S %p'; - """, - ], - ] - ): - try: - t1.append( - subprocess.check_output( - o, - shell=True, - timeout=timeout2, - ) - .decode('utf-8') - .strip() - ) - except Exception: - t1.append('fail %d' % sh_index) - - t3 = ' | '.join(t1).replace('\n\r', '') - - sys.stdout.write(t3) - sys.stdout.flush() - - def custom_translate( current_string, check, @@ -4107,6 +3994,8 @@ def commands_cli(argv: Optional[list[str]] = None) -> int: options.command = Command(options._command) if options.command is Command.status: + from .commands_typed.status import run as status + status(args) elif options.command is Command.http_server: http_server(args) diff --git a/python/online/fxreader/pr34/commands_typed/status.py b/python/online/fxreader/pr34/commands_typed/status.py new file mode 100644 index 0000000..3ae2246 --- /dev/null +++ b/python/online/fxreader/pr34/commands_typed/status.py @@ -0,0 +1,155 @@ +import sys +import io +import json +import subprocess +import logging +import inspect +import textwrap +import optparse +import traceback + +from typing import ( + Any, + Optional, +) + +logger = logging.getLogger(__name__) + + +def run(argv: list[str]): + assert isinstance(argv, list) and all([isinstance(o, str) for o in argv]) + + class c1(optparse.IndentedHelpFormatter): + def format_option(self, *args: Any, **kwargs: Any) -> Any: + def f1(text: str, width: Optional[int]) -> list[str]: + width = None + return '\n'.join([textwrap.fill('\t' + o, width, replace_whitespace=False) for o in text.splitlines()]).splitlines() + + t1 = inspect.getsource(optparse.IndentedHelpFormatter.format_option) + t2 = ( + '\n'.join([o[4:] for o in t1.splitlines()[:]]) + .replace( + 'textwrap.wrap', + 'f1', + ) + .replace('format_option', 'f2') + ) + ns: dict[str, Any] = dict() + exec(t2, dict(f1=f1), ns) + return ns['f2'](self, *args, **kwargs) + + parser = optparse.OptionParser( + formatter=c1( + width=None, + ), + ) + + def add_option( + p: optparse.OptionParser, + option_name: str, + dest: str, + default: Optional[Any] = None, + action: Optional[str] = None, + **kwargs: Any, + ) -> None: + getattr(p, 'add_option')( + option_name, + dest=dest, + default=default, + action=action, + **kwargs, + ) + + add_option( + parser, + '--sh', + dest='sh', + default=[], + action='append', + type=str, + ) + add_option( + parser, + '--timeout', + dest='timeout', + default=None, + type=float, + ) + add_option( + parser, + '--config', + dest='config', + default=None, + type=str, + help=''.join( + [ + '.json file with array of strings, each is a shell command ', + 'that outputs a separate status text value, ', + 'like\n', + r""" +ping -w 1 -i 0.02 -c 3 | tail -n 2| head -n 1 | grep -Po $'time\\s+.*$' +sensors -j | jq -r '.\"coretemp-isa-0000\".\"Package id 0\".temp1_input|tostring + \" C\"' +printf '%d RPM' $(cat /sys/devices/platform/applesmc.768/fan1_input) +printf '% 3.0f%%' $(upower -d | grep -Po 'percentage:\\s+\\d+(\\.\\d+)?%' | grep -Po '\\d+(\\.\\d+)?' | head -n 1) + """.strip(), + ] + ), + ) + options, args = parser.parse_args(argv) + + if options.timeout is None: + options.timeout = 0.5 + + timeout2 = max(options.timeout, 0.0) + + assert timeout2 >= 0.0 and timeout2 <= 4 + + config: dict[str, Any] = dict() + + try: + if not options.config is None: + with io.open(options.config, 'r') as f: + config.update(json.load(f)) + except Exception: + logging.error(traceback.format_exc()) + pass + + options.sh.extend(config.get('sh', [])) + + t1: list[str] = [] + + for sh_index, o in enumerate( + [ + *options.sh, + *[ + r""" + A=$(free -h | grep -P Mem: | grep -Po '[\w\.\d]+'); + echo -n $A | awk '{print $2, $7}'; + """, + r""" + date +'%Y-%m-%d %l:%M:%S %p'; + """, + ], + ] + ): + try: + t1.append( + subprocess.check_output( + o, + shell=True, + timeout=timeout2, + ) + .decode('utf-8') + .strip() + ) + except Exception: + t1.append('fail %d' % sh_index) + + t3 = ' | '.join(t1).replace('\n\r', '') + + sys.stdout.write(t3) + sys.stdout.flush() + + +if __name__ == '__main__': + run(sys.argv[1:]) diff --git a/python/pyproject.toml b/python/pyproject.toml index 3a18b73..badcd38 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -76,7 +76,7 @@ oom_firefox = 'online.fxreader.pr34.oom_firefox:main' [tool.ruff] -line-length = 160 +line-length = 80 target-version = 'py310' # builtins = ['_', 'I', 'P'] include = [