Source code for libqtile.sh

# Copyright (c) 2008, Aldo Cortesi. All rights reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""
    A command shell for Qtile.
"""

import fcntl
import inspect
import pprint
import re
import readline
import sys
import struct
import termios
from typing import List, Optional, Tuple

from libqtile import command_graph
from libqtile.command_interface import format_selectors, CommandInterface, CommandError, CommandException


def terminal_width():
    width = None
    try:
        cr = struct.unpack('hh', fcntl.ioctl(0, termios.TIOCGWINSZ, '1234'))
        width = int(cr[1])
    except (IOError, ImportError):
        pass
    return width or 80


[docs]class QSh: """Qtile shell instance""" def __init__(self, client: CommandInterface, completekey="tab") -> None: self._client = client self._current_node = command_graph.CommandGraphRoot() # type: command_graph.CommandGraphNode self._completekey = completekey self._builtins = [i[3:] for i in dir(self) if i.startswith("do_")] self._termwidth = terminal_width() def complete(self, arg, state) -> Optional[str]: buf = readline.get_line_buffer() completers = self._complete(buf, arg) if completers and state < len(completers): return completers[state] return None def _complete(self, buf, arg) -> List[str]: if not re.search(r" |\(", buf) or buf.startswith("help "): options = self._builtins + self._commands lst = [i for i in options if i.startswith(arg)] return lst elif buf.startswith("cd ") or buf.startswith("ls "): last_slash = arg.rfind("/") + 1 path, last = arg[:last_slash], arg[last_slash:] node = self._find_path(path) if node is None: return [] options = [str(i) for i in self._ls(node)] lst = [] if path and not path.endswith("/"): path += "/" for i in options: if i.startswith(last): lst.append(path + i) if len(lst) == 1: # add a slash to continue completing the next part of the path return [lst[0] + "/"] return lst return [] @property def prompt(self) -> str: return "{} > ".format(format_selectors(self._current_node.selectors)) def columnize(self, lst, update_termwidth=True) -> str: if update_termwidth: self.termwidth = terminal_width() ret = [] if lst: lst = list(map(str, lst)) mx = max(map(len, lst)) cols = self.termwidth // (mx + 2) or 1 # We want `(n-1) * cols + 1 <= len(lst) <= n * cols` to return `n` # If we subtract 1, then do `// cols`, we get `n - 1`, so we can then add 1 rows = (len(lst) - 1) // cols + 1 for i in range(rows): # Because Python array slicing can go beyond the array bounds, # we don't need to be careful with the values here sl = lst[i * cols: (i + 1) * cols] sl = [x + " " * (mx - len(x)) for x in sl] ret.append(" ".join(sl)) return "\n".join(ret) @property def _commands(self) -> List[str]: try: commands_cmd = self._current_node.call("commands") return self._client.execute(commands_cmd, (), {}) except CommandError: return [] def _inspect(self, obj: command_graph.CommandGraphNode) -> Tuple[Optional[List[str]], Optional[List[str]]]: """Returns an (attrs, keys) tuple""" if isinstance(obj, command_graph.CommandGraphObject) and obj.selector is None: items_call = obj.parent.call("items") allow_root, items = self._client.execute(items_call, (obj.object_type,), {}) attrs = obj.children if allow_root else None return attrs, items else: return obj.children, None def _ls(self, obj: command_graph.CommandGraphNode) -> List[str]: attrs, itms = self._inspect(obj) all_items = [] # type: List[str] if attrs: all_items.extend(attrs) if itms: all_items.extend(itms) return all_items def _find_path(self, path: str) -> Optional[command_graph.CommandGraphNode]: """Find an object relative to the current node Finds and returns the command graph node that is defined relative to the current node. """ root = command_graph.CommandGraphRoot() if path.startswith("/") else self._current_node parts = [i for i in path.split("/") if i] return self._find_node(root, *parts) def _find_node(self, src: command_graph.CommandGraphNode, *paths: str) -> Optional[command_graph.CommandGraphNode]: """Find an object in the command graph Return the object in the command graph at the specified path relative to the given node. """ if len(paths) == 0: return src path, *next_path = paths next_node = None if path == "..": next_node = src.parent or src else: attrs, items = self._inspect(src) for transformation in [str, int]: try: transformed_path = transformation(path) except ValueError: continue if attrs is not None and transformed_path in attrs: nav_node = src.navigate(transformed_path, None) next_node = nav_node break elif items is not None and transformed_path in items: assert isinstance(src, command_graph.CommandGraphObject) nav_node = src.parent.navigate(src.object_type, transformed_path) next_node = nav_node break if next_node: return self._find_node(next_node, *next_path) else: return None
[docs] def do_cd(self, arg) -> str: """Change to another path. Examples ======== cd layout/0 cd ../layout """ next_node = self._find_path(arg) if next_node is not None: self._current_node = next_node return format_selectors(self._current_node.selectors) or '/' else: return "No such path."
[docs] def do_ls(self, arg: str) -> str: """List contained items on a node. Examples ======== > ls > ls ../layout """ if arg: node = self._find_path(arg) if not node: return "No such path." else: node = self._current_node ls = self._ls(node) formatted_ls = ["{}/".format(i) for i in ls] return self.columnize(formatted_ls)
[docs] def do_pwd(self, arg) -> str: """Returns the current working location This is the same information as presented in the qshell prompt, but is very useful when running iqshell. Examples ======== > pwd / > cd bar/top bar['top']> pwd bar['top'] """ return format_selectors(self._current_node.selectors) or '/'
[docs] def do_help(self, arg) -> str: """Give help on commands and builtins When invoked without arguments, provides an overview of all commands. When passed as an argument, also provides a detailed help on a specific command or builtin. Examples ======== > help > help command """ if not arg: lst = [ "help command -- Help for a specific command.", "", "Builtins", "========", self.columnize(self._builtins), ] cmds = self._commands if cmds: lst.extend([ "", "Commands for this object", "========================", self.columnize(cmds), ]) return "\n".join(lst) elif arg in self._commands: call = self._current_node.call("doc") return self._client.execute(call, (arg,), {}) elif arg in self._builtins: c = getattr(self, "do_" + arg) ret = inspect.getdoc(c) assert ret is not None return ret else: return "No such command: %s" % arg
[docs] def do_exit(self, args) -> None: """Exit qshell""" sys.exit(0)
do_quit = do_exit do_q = do_exit def process_line(self, line: str) -> str: builtin_match = re.fullmatch(r"(?P<cmd>\w+)(?:\s+(?P<arg>\S+))?", line) if builtin_match: cmd = builtin_match.group("cmd") args = builtin_match.group("arg") if cmd in self._builtins: builtin = getattr(self, "do_" + cmd) val = builtin(args) return val else: return "Invalid builtin: {}".format(cmd) command_match = re.fullmatch(r"(?P<cmd>\w+)\((?P<args>[\w\s,]*)\)", line) if command_match: cmd = command_match.group("cmd") args = command_match.group("args") if args: cmd_args = tuple(map(str.strip, args.split(","))) else: cmd_args = () if not self._client.has_command(self._current_node, cmd): return "Command does not exist: {}".format(cmd) cmd_call = self._current_node.call(cmd) try: return self._client.execute(cmd_call, cmd_args, {}) except CommandException as e: return "Command exception: {}\n".format(e) return "Invalid command: {}".format(line) def loop(self) -> None: readline.set_completer(self.complete) readline.parse_and_bind(self._completekey + ": complete") readline.set_completer_delims(" ()|") while True: try: line = input(self.prompt) except (EOFError, KeyboardInterrupt): print() return if not line: continue val = self.process_line(line) if isinstance(val, str): print(val) elif val: pprint.pprint(val)