Source code for uppaal2jetracer.command_system

"""Command system

This file provides all interfaces required to access uppaal2jetracer commands.

This file can be imported as a module and contains the following classes:

    * CommandHandler:       Interface for handling uppaal2jetracer commands.
    * CommandResult:        Result of uppaal2jetracer commands.
    * CommandResultType:    Result type of uppaal2jetracer commands.
    * Command:              Interface for all uppaal2jetracer commands.
    * ParseCommand:         Command for parsing UPPAAL automata.
    * ParseRunCommand:      Command for parsing and running an UPPAAL system.
    * ProjectCommand:       Command for managing projects.
    * QuitCommand:          Command for quitting uppaal2jetracer.
    * RunCommand:           Command for running a parsed UPPAAL system.
    * VersionCommand:       Command for managing versions of parsed automata.
"""

from __future__ import annotations

import logging
import logging.config
import os.path
import pickle
from abc import ABC, abstractmethod
from enum import Enum
from typing import Dict, List, Callable

from bs4 import BeautifulSoup

from uppaal2jetracer.controller.executor import Executor, HardwareCallError
from uppaal2jetracer.controller.uppaal_controller import SystemController
from uppaal2jetracer.parser.parser import SystemParser, XMLParseError
from uppaal2jetracer.uppaalmodel.system import System
from uppaal2jetracer.versioncontrol.versioncontrol import VersionManager, ProjectManager, \
    ResponseObject, ProjectResponse, ResponseType, VersionResponse, GlobalResponse

logger = logging.getLogger("user_command")


[docs] class CommandHandler(ABC): """ Abstract base class for all command handlers. :ivar _project_manager: Manages all projects. :vartype _project_manager: ProjectManager :ivar _version_manager: Manages all versions in a project. :vartype _version_manager: VersionManager """ __slots__ = ("_running", "_commands", "_version_manager", "_project_manager") _PARSE_COMMAND = "parse" _PARSE_COMMAND_ALT = "p" _PARSE_RUN_COMMAND = "parseandrun" _PARSE_RUN_COMMAND_ALT = "pnr" _PROJECT_COMMAND = "project" _PROJECT_COMMAND_ALT = "prj" _QUIT_COMMAND = "quit" _QUIT_COMMAND_ALT = "q" _RUN_COMMAND = "run" _RUN_COMMAND_ALT = "r" _VERSION_COMMAND = "version" _VERSION_COMMAND_ALT = "v" def __init__(self, version_manager: VersionManager, project_manager: ProjectManager): logging.config.fileConfig(os.path.dirname(__file__) + os.sep + "logging.conf", disable_existing_loggers = True) self._running: bool = True self._version_manager: VersionManager = version_manager self._project_manager: ProjectManager = project_manager self._commands: Dict[str, Command] = {} self._init_commands()
[docs] @staticmethod def activate_debug(loggers: List[str]): """ Activates debug mode for given loggers. :param loggers: List of logger names. :type loggers: List[str] """ for log in loggers: logging.getLogger(log).setLevel(logging.DEBUG)
[docs] def deactivate_debug(self): """ Deactivates debug mode for all loggers. """ for command in list(self._commands.values()): for log in command.loggers: logging.getLogger(log).setLevel(logging.INFO)
[docs] def quit(self): """ Quits the program. """ self._running = False
[docs] @abstractmethod def handle_input(self): """ Handles the input of an actor using the command system. """
[docs] def get_confirmation(self, message: str) -> bool: """ Asks the user to confirm their input. :param message: The message to ask the user for. :type message: str :return: True if the user has confirmed, False otherwise. :rtype: bool """
@property def parse_command_name(self) -> str: """ Returns the name of the parse command. """ return self._PARSE_COMMAND @property def parse_run_command_name(self) -> str: """ Returns the name of the parse and run command. """ return self._PARSE_RUN_COMMAND @property def project_command_name(self) -> str: """ Returns the name of the project command. """ return self._PROJECT_COMMAND @property def quit_command_name(self) -> str: """ Returns the name of the quit command. """ return self._QUIT_COMMAND @property def run_command_name(self) -> str: """ Returns the name of the run command. """ return self._RUN_COMMAND @property def version_command_name(self) -> str: """ Returns the name of the version command. """ return self._VERSION_COMMAND def _init_commands(self): # Initialize basic commands. self._commands.update({self._PARSE_COMMAND: ParseCommand(self._version_manager)}) self._commands.update({self._PARSE_COMMAND_ALT: ParseCommand(self._version_manager)}) self._commands.update({self._PROJECT_COMMAND: ProjectCommand(self._project_manager)}) self._commands.update({self._PROJECT_COMMAND_ALT: ProjectCommand(self._project_manager)}) self._commands.update({self._QUIT_COMMAND: QuitCommand(self)}) self._commands.update({self._QUIT_COMMAND_ALT: QuitCommand(self)}) self._commands.update({self._RUN_COMMAND: RunCommand()}) self._commands.update({self._RUN_COMMAND_ALT: RunCommand()}) self._commands.update({self._VERSION_COMMAND: VersionCommand(self, self._version_manager)}) self._commands.update({self._VERSION_COMMAND_ALT: VersionCommand( self, self._version_manager )}) # Initialize pseudo commands. self._commands.update({self._PARSE_RUN_COMMAND: ParseRunCommand( ParseCommand(self._version_manager), RunCommand() )}) self._commands.update({self._PARSE_RUN_COMMAND_ALT: ParseRunCommand( ParseCommand(self._version_manager), RunCommand() )}) @abstractmethod def _execute_command(self, command: str): pass
[docs] class CommandResult: """ A class capsuling the result of a command execution. :ivar _message: The return message of the command. :vartype _message: str :ivar _payload: Response payload of the command. :vartype _payload: List[ResponseObject] :ivar _result_type: The type of the result. Either "success" or "failure". :vartype _result_type: CommandResultType """ __slots__ = ("_message", "_payload", "_result_type") def __init__(self, message: str, result_type: CommandResultType, payload: List[ResponseObject] = None): self._message = message self._payload = payload self._result_type = result_type @property def message(self) -> str: """ The return message of the command. """ return self._message @property def payload(self) -> List[ResponseObject]: """ The payload of the command. """ return self._payload @property def result_type(self) -> CommandResultType: """ The type of the result. Either "success" or "failure". """ return self._result_type
[docs] class CommandResultType(Enum): """ An enum class with result types of a command. """ SUCCESS = 0 FAILURE = 1
[docs] class Command(ABC): """ Abstract base class for all u2j-commands. """ _HELP_MESSAGE = "" _LOGGERS = [] _PKL_FILE_KEY = ".pkl" _XML_FILE_KEY = ".xml" _LINE_SEPARATOR = "\n" _ERROR_ARG_COUNT = "Invalid number of arguments provided." _ERROR_FILE_NOT_FOUND = "'{}' is not a valid file." _ERROR_HARDWARE_FAIL = "Execution of hardware command failed." _ERROR_INVALID_ARG = "'{}' is not a valid argument." _ERROR_EXECUTION_FAIL = "Execution of '{}' failed." _SUCCESS_TERMINATED = "Successfully terminated execution of '{}'." _SUCCESS_RUN = "Successfully ran '{}'."
[docs] @abstractmethod def execute(self, args: List[str]) -> CommandResult: """ Executes the command. :param args: Arguments passed to the command. :type args: List[str] :return: Result of the command. :rtype: CommandResult """
@property def help_message(self) -> str: """ Returns the help message for the command. :return: The help message. :rtype: str """ return self._HELP_MESSAGE @property def loggers(self) -> List[str]: """ Returns the debugger of the command. :return: The debugger. :rtype: str """ return self._LOGGERS
[docs] class ParseCommand(Command): """ A command to parse an UPPAAL system to an executable UPPAAL model. """ __slots__ = ("_version_manager",) _HELP_MESSAGE = """Usage: (parse | p) <path> [-d | --debug] [-h | --help] The parse command parses an UPPAAL system to an executable UPPAAL model. Options: -d, --debug Log debug messages. -h, --help Show this message.""" _LOGGERS = ["parser", "controller", "version_control", "uppaal_model", "yacc"] _BS_BUILDER = "lxml-xml" _ERROR_FILE_TYPE = "'{}' is not a valid xml file." _ERROR_UPPAAL_XML = "'{}' is not a valid UPPAAL xml." _SUCCESS_MESSAGE = "Parsed successfully and saved in '{}'." def __init__(self, version_manager: VersionManager): self._version_manager = version_manager logger.info("Initialized parse command.")
[docs] def execute(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) if not os.path.isfile(args[0]): return CommandResult(self._ERROR_FILE_NOT_FOUND.format(args[0]), CommandResultType.FAILURE) if not args[0].endswith(self._XML_FILE_KEY): return CommandResult(self._ERROR_FILE_TYPE.format(args[0]), CommandResultType.FAILURE) with open(args[0], "r", encoding = "utf-8") as file: try: system = SystemParser.parse(BeautifulSoup(file, self._BS_BUILDER)) except XMLParseError: return CommandResult(self._ERROR_UPPAAL_XML.format(args[0]), CommandResultType.FAILURE) pkl_file = args[0].replace(self._XML_FILE_KEY, self._PKL_FILE_KEY) with open(pkl_file, "wb") as file: file.write(pickle.dumps(system)) version_name: str = file.name.split(os.sep).pop().removesuffix(self._PKL_FILE_KEY) results: List[VersionResponse] = self._version_manager.add_version(version_name, system) for result in results: if result.type == ResponseType.ERROR: return CommandResult(result.message, CommandResultType.FAILURE, results) return CommandResult(self._SUCCESS_MESSAGE.format(pkl_file), CommandResultType.SUCCESS, results)
[docs] class ParseRunCommand(Command): """ A command to parse and execute an UPPAAL system. """ __slots__ = ("_parse_command", "_run_command",) _HELP_MESSAGE = """Usage: (parseandrun | pnr) <path> [-d | --debug] [-h | --help] The parse and run command parses an UPPAAL system and runs it. Options: -d, --debug Log debug messages. -h, --help Show this message.""" _LOGGERS = ["parser", "controller", "version_control", "uppaal_model", "yacc", "jetracer"] def __init__(self, parse_command: ParseCommand, run_command: RunCommand): self._parse_command = parse_command self._run_command = run_command
[docs] def execute(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) parse_result: CommandResult = self._parse_command.execute(args) if not parse_result.result_type == CommandResultType.SUCCESS: return CommandResult(parse_result.message, CommandResultType.FAILURE) args[0] = args[0].replace(self._XML_FILE_KEY, self._PKL_FILE_KEY) run_result: CommandResult = self._run_command.execute(args) return CommandResult(run_result.message, run_result.result_type, parse_result.payload)
[docs] class ProjectCommand(Command): """ A command for managing projects. """ __slots__ = ("_project_manager", "_subcommands",) _HELP_MESSAGE = """Usage: (project | prj) (list | current | config | max <int > 0> | open <name> | new <name> | delete <name>) [-d | --debug] [-h | --help] The project command manages projects which contain versions of parsed UPPAAL systems. Commands: list Lists all projects. current Shows the current project. config Shows the project configuration. max Set maximum number of versions in a project. open Open a project. new Create a new project. delete Delete a project. Options: -d, --debug Log debug messages. -h, --help Show this message.""" _LOGGERS = ["version_control"] _ERROR_PROJECT_NOT_FOUND = "Project '{}' not found." _SUCCESS_CURRENT = "Current project is '{}'." _SUCCESS_CONFIG = "Current version limit is {}." _SUCCESS_NEW = "New project '{}' created." _SUCCESS_OPEN = "Opened project '{}'." _SUCCESS_REMOVE = "Deleted project '{}'." _SUCCESS_SET_MAX = "Set maximum number of versions per project to {}." def __init__(self, project_manager: ProjectManager): self._project_manager = project_manager self._subcommands = self._init_subcommands() logger.info("Initialized project command.")
[docs] def execute(self, args: List[str]) -> CommandResult: if not 1 <= len(args) <= 2: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) if args[0] not in self._subcommands: return CommandResult(self._ERROR_INVALID_ARG.format(args[0]), CommandResultType.FAILURE) return self._subcommands.get(args[0])(args[1:])
def _init_subcommands(self) -> Dict[str, Callable[[List[str]], CommandResult]]: return { "list": self._execute_list, "current": self._execute_current, "config": self._execute_config, "max": self._execute_max, "open": self._execute_open, "new": self._execute_new, "delete": self._execute_delete } def _execute_list(self, args: List[str]) -> CommandResult: if not len(args) == 0: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) responses: List[ProjectResponse] = self._project_manager.get_all_projects() message = "" for response in reversed(responses): if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, responses) message += f"{response.payload.p_id}: {response.payload.name}\n" message = message.removesuffix(self._LINE_SEPARATOR) return CommandResult(message, CommandResultType.SUCCESS, responses) def _execute_current(self, args: List[str]) -> CommandResult: if not len(args) == 0: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) response: ProjectResponse = self._project_manager.get_current_project() if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) return CommandResult(self._SUCCESS_CURRENT.format(response.payload.name), CommandResultType.SUCCESS, [response]) def _execute_config(self, args: List[str]) -> CommandResult: if not len(args) == 0: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) response: GlobalResponse = self._project_manager.get_config() if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) return CommandResult(self._SUCCESS_CONFIG.format(response.payload.version_max), CommandResultType.SUCCESS, [response]) def _execute_max(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) try: if int(args[0]) <= 0: raise ValueError response: GlobalResponse = self._project_manager.change_version_limit(int(args[0])) if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) return CommandResult(self._SUCCESS_SET_MAX.format(args[0]), CommandResultType.SUCCESS, [response]) except ValueError: return CommandResult(self._ERROR_INVALID_ARG.format(args[0]), CommandResultType.FAILURE) def _execute_open(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) response: ProjectResponse = self._project_manager.select_project(args[0]) if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) return CommandResult(self._SUCCESS_OPEN.format(args[0]), CommandResultType.SUCCESS, [response]) def _execute_new(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) response: ProjectResponse = self._project_manager.add_project(args[0]) if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) return CommandResult(self._SUCCESS_NEW.format(args[0]), CommandResultType.SUCCESS, [response]) def _execute_delete(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) response: ProjectResponse = self._project_manager.delete_project(args[0]) if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) return CommandResult(self._SUCCESS_REMOVE.format(args[0]), CommandResultType.SUCCESS, [response])
[docs] class QuitCommand(Command): """ A command to quit uppaal2jetracer. """ __slots__ = ("_command_handler",) _HELP_MESSAGE = """Usage: (quit | q) [-h | --help] The quit command ends uppaal2jetracer. Options: -h, --help Show this message.""" _LOGGERS = [] def __init__(self, command_handler: CommandHandler): self._command_handler = command_handler logger.info("Initialized quit command.")
[docs] def execute(self, args: List[str]) -> CommandResult: if not len(args) == 0: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) self._command_handler.quit() logger.info("Quit uppaal2jetracer.") return CommandResult("", CommandResultType.SUCCESS)
[docs] class RunCommand(Command): """ A command to run an UPPAAL system. """ _HELP_MESSAGE = """Usage: (run | r) <path> [-d | --debug] [-h | --help] The run command starts the execution of an UPPAAL system. Options: -d, --debug Log debug messages. -h, --help Show this message.""" _LOGGERS = ["controller", "uppaal_model", "jetracer"] _ERROR_FILE_TYPE = "'{}' is not a valid pkl file." def __init__(self): logger.info("Initialized run command.")
[docs] def execute(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) if not os.path.isfile(args[0]): return CommandResult(self._ERROR_FILE_NOT_FOUND.format(args[0]), CommandResultType.FAILURE) if not args[0].endswith(self._PKL_FILE_KEY): return CommandResult(self._ERROR_FILE_TYPE.format(args[0]), CommandResultType.FAILURE) with open(args[0], "rb") as file: system: System = pickle.load(file) controller: SystemController = SystemController(system) try: controller.run_system() except KeyboardInterrupt: Executor.stop() return CommandResult(self._SUCCESS_TERMINATED.format(args[0]), CommandResultType.SUCCESS) except HardwareCallError: return CommandResult(self._ERROR_HARDWARE_FAIL, CommandResultType.FAILURE) return CommandResult(self._SUCCESS_RUN.format(args[0]), CommandResultType.SUCCESS)
[docs] class VersionCommand(Command): """ A command to manage versions of parsed UPPAAL systems. """ __slots__ = ("_version_manager", "_command_handler", "_controller") _HELP_MESSAGE = """Usage: (version | v) (list | run <id> | fav <id> | delete <id>) [-d | --debug] [-h | --help] The version command manages parsed versions of UPPAAL systems. Commands: list List all versions in current project. run Run an existing version. fav Prevent existing version from being automatically deleted. delete Delete existing version. Options: -d, --debug Log debug messages. -h, --help Show this message.""" _LOGGERS = ["version_control", "controller", "uppaal_model", "jetracer"] _SUCCESS_FAVORITE = "Version {} is now a favorite." _SUCCESS_DEFAVORITE = "Version {} is now no longer a favorite." _SUCCESS_DELETE = "Deleted version {} from project." _SUCCESS_STOP = "The controller has been stopped." _ERROR_NOT_RUNNING = "The controller isn't currently running." _ABORT_DELETE = "Aborted deletion of version." _DELETE_CONFIRMATION_REQUEST = "Are you sure you want to delete a favorite version?" def __init__(self, command_handler: CommandHandler, version_manager: VersionManager): self._command_handler = command_handler self._version_manager = version_manager self._subcommands = self._init_subcommands() self._controller = None logger.info("Initialized version command.")
[docs] def execute(self, args: List[str]) -> CommandResult: if not 1 <= len(args) <= 2: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) if args[0] not in self._subcommands: return CommandResult(self._ERROR_INVALID_ARG.format(args[0]), CommandResultType.FAILURE) return self._subcommands.get(args[0])(args[1:])
def _init_subcommands(self) -> Dict[str, Callable[[List[str]], CommandResult]]: return { "list": self._execute_list, "run": self._execute_run, "fav": self._execute_fav, "delete": self._execute_delete, "stop": self._execute_stop } def _execute_list(self, args: List[str]) -> CommandResult: if not len(args) == 0: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) responses: List[VersionResponse] = self._version_manager.get_all_versions() message = "" for response in responses: if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, responses) message += (f"{response.payload.v_id}: {response.payload.name}: {response.payload.date}" f"\n") message = message.removesuffix(self._LINE_SEPARATOR) return CommandResult(message, CommandResultType.SUCCESS, responses) def _execute_run(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) try: response: VersionResponse = self._version_manager.get_version(int(args[0])) except ValueError: return CommandResult(self._ERROR_INVALID_ARG.format(args[0]), CommandResultType.FAILURE) if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) controller: SystemController = SystemController(response.payload.file) try: if self._controller is not None: self._controller.is_running = False self._controller = controller controller.run_system() except KeyboardInterrupt: Executor.stop() return CommandResult(self._SUCCESS_TERMINATED.format(args[0]), CommandResultType.SUCCESS) except HardwareCallError: return CommandResult(self._ERROR_HARDWARE_FAIL, CommandResultType.FAILURE) return CommandResult(self._SUCCESS_RUN.format(args[0]), CommandResultType.SUCCESS) def _execute_stop(self, args: List[str]) -> CommandResult: if not len(args) == 0: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) if self._controller is None or not self._controller.is_running: return CommandResult(self._ERROR_NOT_RUNNING, CommandResultType.FAILURE) self._controller.is_running = False self._controller = None return CommandResult(self._SUCCESS_STOP, CommandResultType.SUCCESS) def _execute_fav(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) try: response: VersionResponse = self._version_manager.favorite_version(int(args[0])) except ValueError: return CommandResult(self._ERROR_INVALID_ARG.format(args[0]), CommandResultType.FAILURE) if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) if self._version_manager.get_version(int(args[0])).payload.favorite: return CommandResult(self._SUCCESS_FAVORITE.format(args[0]), CommandResultType.SUCCESS, [response]) return CommandResult(self._SUCCESS_DEFAVORITE.format(args[0]), CommandResultType.SUCCESS, [response]) def _execute_delete(self, args: List[str]) -> CommandResult: if not len(args) == 1: return CommandResult(self._ERROR_ARG_COUNT, CommandResultType.FAILURE) try: if (self._version_manager.get_version(int(args[0])).payload.favorite and not self._command_handler.get_confirmation( self._DELETE_CONFIRMATION_REQUEST)): return CommandResult(self._ABORT_DELETE, CommandResultType.SUCCESS, None) response: VersionResponse = self._version_manager.delete_version(int(args[0])) except ValueError: return CommandResult(self._ERROR_INVALID_ARG.format(args[0]), CommandResultType.FAILURE) if response.type == ResponseType.ERROR: return CommandResult(response.message, CommandResultType.FAILURE, [response]) return CommandResult(self._SUCCESS_DELETE.format(args[0]), CommandResultType.SUCCESS, [response])