"""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_RUN_COMMAND = "pnr"
_PROJECT_COMMAND = "prj"
_QUIT_COMMAND = "quit"
_RUN_COMMAND = "run"
_VERSION_COMMAND = "ver"
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
@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._PROJECT_COMMAND: ProjectCommand(self._project_manager)})
self._commands.update({self._QUIT_COMMAND: QuitCommand(self)})
self._commands.update({self._RUN_COMMAND: RunCommand()})
self._commands.update({self._VERSION_COMMAND: VersionCommand(self._version_manager)})
# Initialize pseudo commands.
self._commands.update({self._PARSE_RUN_COMMAND: 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 <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"]
_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: 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", "executor"]
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: 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 [-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 <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 = ["executor", "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",)
_HELP_MESSAGE = """Usage: ver (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"]
_SUCCESS_FAVORITE = "Version {} is now a favorite."
_SUCCESS_DELETE = "Deleted version {} from project."
def __init__(self, version_manager: VersionManager):
self._version_manager = version_manager
self._subcommands = self._init_subcommands()
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
}
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:
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_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])
return CommandResult(self._SUCCESS_FAVORITE.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:
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])