pebble-firmware/python_libs/pebble-commander/pebble/commander/interactive.py

# Copyright 2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import absolute_import

import shlex
import traceback

from log_hashing.logdehash import LogDehash
import prompt_toolkit

from .commander import PebbleCommander


class InteractivePebbleCommander(object):
    """ Interactive Pebble Commander.
        Most/all UI implementations should either use this directly or sub-class it.
    """
    def __init__(self, loghash_path=None, tty=None, capfile=None):
        self.cmdr = PebbleCommander(tty=tty, interactive=True, capfile=capfile)
        if loghash_path is None:
            loghash_path = "build/src/fw/loghash_dict.json"
        self.dehasher = LogDehash(loghash_path)
        self.cmdr.attach_log_listener(self.log_listener)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def __del__(self):
        self.close()

    def close(self):
        try:
            self.cmdr.close()
        except:
            pass

    def attach_prompt_toolkit(self):
        """ Attaches prompt_toolkit things
        """
        self.history = prompt_toolkit.history.InMemoryHistory()
        self.cli = prompt_toolkit.CommandLineInterface(
            application=prompt_toolkit.shortcuts.create_prompt_application(u"> ",
                                                                           history=self.history),
            eventloop=prompt_toolkit.shortcuts.create_eventloop())
        self.patch_context = self.cli.patch_stdout_context(raw=True)
        self.patch_context.__enter__()

    def log_listener(self, msg):
        """ This is called on every incoming log message.
            `msg` is the raw log message class, without any dehashing.

            Subclasses should override this probably.
        """
        line_dict = self.dehasher.dehash(msg)
        line = self.dehasher.commander_format_line(line_dict)
        print line

    def dispatch_command(self, string):
        """ Dispatches a command string.

            Subclasses should not override this.
        """
        args = shlex.split(string)
        # Starting with '!' passes the rest of the line directly to prompt.
        # Otherwise we try to run a command; if that fails, the line goes to prompt.
        if string.startswith("!"):
            string = string[1:]  # Chop off the '!' marker
        else:
            cmd = self.cmdr.get_command(args[0])
            if cmd:  # If we provide the command, run it.
                return cmd(*args[1:])

        return self.cmdr.send_prompt_command(string)

    def input_handle(self, string):
        """ Handles an input line.
            Generally the flow is to handle any UI-specific commands, then pass on to
            dispatch_command.

            Subclasses should override this probably.
        """
        # Handle "quit" strings
        if string in ["exit", "q", "quit"]:
            return False

        try:
            resp = self.dispatch_command(string)
            if resp is not None:
                print "\x1b[1m" + '\n'.join(resp) + "\x1b[m"
        except:
            print "An error occurred!"
            traceback.print_exc()

        return True

    def get_command(self):
        """ Get a command input line.
            If there is no line, return an empty string or None.
            This may block.

            Subclasses should override this probably.
        """
        if self.cli is None:
            self.attach_prompt_toolkit()
        doc = self.cli.run(reset_current_buffer=True)
        if doc:
            return doc.text
        else:
            return None

    def command_loop(self):
        """ The main command loop.

            Subclasses could override this, but it's probably not useful to do.
        """
        while True:
            try:
                cmd = self.get_command()
                if cmd and not self.input_handle(cmd):
                    break
            except (KeyboardInterrupt, EOFError):
                break