Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 38 additions & 0 deletions benchkit/shell/args.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
# SPDX-License-Identifier: MIT


from abc import ABC, abstractmethod
from dataclasses import dataclass

from benchkit.utils.types import PathType


class Arg(ABC):
@abstractmethod
def accept(self, visitor):
pass


@dataclass
class StrArg(Arg):
string: str

def accept(self, visitor):
return visitor.visit_str_arg(self)


@dataclass
class FilePathArg(Arg):
path: PathType

def accept(self, visitor):
return visitor.visit_file_path_arg(self)


@dataclass
class ExecutableArg(Arg):
name: PathType

def accept(self, visitor):
return visitor.visit_executable_arg(self)
89 changes: 89 additions & 0 deletions benchkit/shell/commands.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,89 @@
# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
# SPDX-License-Identifier: MIT


from __future__ import annotations

import shlex
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List

from benchkit.communication import CommunicationLayer
from benchkit.shell.args import Arg, ExecutableArg, StrArg


class Command(ABC):
@abstractmethod
def accept(self, visitor):
pass

def run_with(
self,
comm_layer: CommunicationLayer,
) -> str:
return run_command(
command=self,
comm_layer=comm_layer,
)

def resolve(self, comm_layer: CommunicationLayer) -> Command:
from benchkit.shell.visitors.command_visitor import CommandResolver

return self.accept(visitor=CommandResolver(comm_layer=comm_layer))


@dataclass
class SingleCommand(Command):
program: Arg
args: List[Arg]

def accept(self, visitor):
return visitor.visit_single_command(self)

def to_list_str(self) -> List[str]:
from benchkit.shell.visitors.arg_visitor import ArgStringifier

arg_str = ArgStringifier()
argv0 = str(self.program.accept(arg_str))
argvn = [str(arg.accept(arg_str)) for arg in self.args]
command_lst = [argv0] + argvn
return command_lst

def to_str(self) -> str:
command_lst = self.to_list_str()
command_str = " ".join(command_lst)
return command_str


class WhichCommand(SingleCommand):
def __init__(self, name: str):
super().__init__(program=ExecutableArg("which"), args=[StrArg(name)])


def run_command(
command: Command,
comm_layer: CommunicationLayer,
) -> str:
# TODO return
from benchkit.shell.visitors.command_visitor import CommandExecutor

executor = CommandExecutor(comm_layer=comm_layer)
return executor.run(command=command)


def command_from_liststr(command: List[str]) -> SingleCommand:
if len(command) == 0:
raise ValueError(
"Empty command. There must be at least one argument, the program name or path."
)

return SingleCommand(
program=ExecutableArg(name=command[0]),
args=[StrArg(string=a) for a in command[1:]],
)


def command_from_str(command: str) -> SingleCommand:
shlexed_command = shlex.split(command)
return command_from_liststr(command=shlexed_command)
2 changes: 2 additions & 0 deletions benchkit/shell/visitors/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
# SPDX-License-Identifier: MIT
53 changes: 53 additions & 0 deletions benchkit/shell/visitors/arg_visitor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
# SPDX-License-Identifier: MIT


from abc import ABC, abstractmethod
from pathlib import Path

from benchkit.shell.args import ExecutableArg, FilePathArg, StrArg


class ArgVisitor(ABC):
@abstractmethod
def visit_str_arg(self, arg: StrArg):
pass

@abstractmethod
def visit_file_path_arg(self, arg: FilePathArg):
pass

@abstractmethod
def visit_executable_arg(self, arg: ExecutableArg):
pass


class ArgResolver(ArgVisitor):
def __init__(self, comm):
self.comm = comm

def visit_str_arg(self, arg: StrArg):
return arg

def visit_file_path_arg(self, arg: FilePathArg):
return FilePathArg(path=Path(arg.path).resolve())

def visit_executable_arg(self, arg: ExecutableArg) -> ExecutableArg:
from benchkit.shell.args import ExecutableArg
from benchkit.shell.commands import WhichCommand

which_cmd = WhichCommand(arg.name)
result = which_cmd.run_with(comm_layer=self.comm)
# TODO return ExecutableArg(name=Path(result.stdout.strip()).resolve())
return ExecutableArg(name=Path(result.strip()).resolve())


class ArgStringifier(ArgVisitor):
def visit_str_arg(self, arg: StrArg):
return arg.string

def visit_file_path_arg(self, arg: FilePathArg):
return str(arg.path)

def visit_executable_arg(self, arg: ExecutableArg):
return arg.name
49 changes: 49 additions & 0 deletions benchkit/shell/visitors/command_visitor.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
# SPDX-License-Identifier: MIT


from abc import ABC, abstractmethod

from benchkit.communication import CommunicationLayer
from benchkit.shell.commands import Command, SingleCommand


class CommandVisitor(ABC):
@abstractmethod
def visit_single_command(self, command: SingleCommand):
pass


class CommandResolver(CommandVisitor):
def __init__(self, comm_layer: CommunicationLayer):
self.comm_layer = comm_layer

def visit_single_command(self, command: SingleCommand):
from benchkit.shell.visitors.arg_visitor import ArgResolver

arg_resolver = ArgResolver(self.comm_layer)
return SingleCommand(
program=command.program.accept(arg_resolver),
args=[arg.accept(arg_resolver) for arg in command.args],
)


class CommandExecutor(CommandVisitor):
def __init__(self, comm_layer: CommunicationLayer):
self.comm_layer = comm_layer

def run(self, command: Command):
return command.accept(self)

def visit_single_command(
self,
command: SingleCommand,
):
shlexed_cmd = command.to_list_str()
output = self.comm_layer.shell( # TODO Replace backend
command=shlexed_cmd,
current_dir=None, # TODO
environment=None, # TODO
output_is_log=False, # TODO
).strip()
return output
28 changes: 28 additions & 0 deletions tests/test_cmd.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
# Copyright (C) 2025 Vrije Universiteit Brussel. All rights reserved.
# SPDX-License-Identifier: MIT

from benchkit.platforms import get_current_platform
from benchkit.shell.commands import command_from_str


def main():
platform = get_current_platform()

py_cmd = "python3 --version"
platform.comm.shell(command=py_cmd)

command = command_from_str(command=py_cmd)
print(command)
print(command.to_list_str())
print(command.to_str())
print(command.run_with(comm_layer=platform.comm))

resolved_command = command.resolve(comm_layer=platform.comm)
print(resolved_command)
print(resolved_command.to_list_str())
print(resolved_command.to_str())
print(resolved_command.run_with(comm_layer=platform.comm))


if __name__ == "__main__":
main()