Javanaut/ffx
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refine tests, CLI

Browse Source
This commit is contained in:
Javanaut
2026-04-09 13:34:38 +02:00
parent 60ae58500a
commit 01b5fdb289
11 changed files with 391 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
src/ffx/cli.py
View File

@@ -58,7 +58,7 @@ def ffx(ctx, database_file, verbose, dry_run):
ctx.obj = {}
if ctx.invoked_subcommand in ('setup_dependencies', 'upgrade'):
if ctx.invoked_subcommand in ('configure_workstation', 'upgrade'):
ctx.obj['dry_run'] = dry_run
ctx.obj['verbosity'] = verbose
return
@@ -104,8 +104,8 @@ def getRepoRootPath():
return os.path.dirname(os.path.dirname(os.path.dirname(currentFilePath)))
def getPrepareScriptPath():
return os.path.join(getRepoRootPath(), 'tools', 'prepare.sh')
def getConfigureWorkstationScriptPath():
return os.path.join(getRepoRootPath(), 'tools', 'configure_workstation.sh')
def getBundleVenvDirectory():
@@ -120,22 +120,23 @@ def getBundleRepoPath():
return getRepoRootPath()
@ffx.command(name='setup_dependencies')
@ffx.command(name='configure_workstation')
@click.pass_context
@click.option('--check', is_flag=True, default=False, help='Only verify dependency readiness')
@click.argument('prepare_args', nargs=-1, type=click.UNPROCESSED)
def setup_dependencies(ctx, check, prepare_args):
prepareScriptPath = getPrepareScriptPath()
@click.option('--check', is_flag=True, default=False, help='Only verify workstation-configuration readiness')
@click.argument('configure_args', nargs=-1, type=click.UNPROCESSED)
def configure_workstation(ctx, check, configure_args):
"""Prepare workstation dependencies and local config after bundle install."""
configureScriptPath = getConfigureWorkstationScriptPath()
if not os.path.isfile(prepareScriptPath):
raise click.ClickException(f"Preparation script not found at {prepareScriptPath}")
if not os.path.isfile(configureScriptPath):
raise click.ClickException(f"Workstation configuration script not found at {configureScriptPath}")
commandSequence = ['bash', prepareScriptPath]
commandSequence = ['bash', configureScriptPath]
if check:
commandSequence.append('--check')
commandSequence += list(prepare_args)
commandSequence += list(configure_args)
if ctx.obj.get('dry_run', False):
click.echo(' '.join(commandSequence))

23
src/ffx/ffx_controller.py
View File

@@ -54,6 +54,13 @@ class FfxController():
self.__logger: Logger = context['logger']
def executeCommandSequence(self, commandSequence):
out, err, rc = executeProcess(commandSequence, context=self.__context)
if rc:
raise click.ClickException(f"Command resulted in error: rc={rc} error={err}")
return out, err, rc
def generateAV1Tokens(self, quality, preset, subIndex : int = 0):
return [f"-c:v:{int(subIndex)}", 'libsvtav1',
@@ -288,9 +295,7 @@ class FfxController():
self.__logger.debug("FfxController.runJob(): Running command sequence")
if not self.__context['dry_run']:
out, err, rc = executeProcess(commandSequence, context=self.__context)
if rc:
raise click.ClickException(f"Command resulted in error: rc={rc} error={err}")
self.executeCommandSequence(commandSequence)
return
if videoEncoder == VideoEncoder.AV1:
@@ -320,7 +325,7 @@ class FfxController():
self.__logger.debug(f"FfxController.runJob(): Running command sequence")
if not self.__context['dry_run']:
executeProcess(commandSequence, context = self.__context)
self.executeCommandSequence(commandSequence)
if videoEncoder == VideoEncoder.H264:
@@ -350,7 +355,7 @@ class FfxController():
self.__logger.debug(f"FfxController.runJob(): Running command sequence")
if not self.__context['dry_run']:
executeProcess(commandSequence, context = self.__context)
self.executeCommandSequence(commandSequence)
@@ -382,7 +387,7 @@ class FfxController():
self.__logger.debug(f"FfxController.runJob(): Running command sequence 1")
if not self.__context['dry_run']:
executeProcess(commandSequence1, context = self.__context)
self.executeCommandSequence(commandSequence1)
commandSequence2 = (commandTokens
+ self.__targetMediaDescriptor.getImportFileTokens()
@@ -409,9 +414,7 @@ class FfxController():
self.__logger.debug(f"FfxController.runJob(): Running command sequence 2")
if not self.__context['dry_run']:
out, err, rc = executeProcess(commandSequence2, context = self.__context)
if rc:
raise click.ClickException(f"Command resulted in error: rc={rc} error={err}")
self.executeCommandSequence(commandSequence2)
@@ -436,4 +439,4 @@ class FfxController():
str(length),
path]
out, err, rc = executeProcess(commandTokens, context = self.__context)
self.executeCommandSequence(commandTokens)

91
src/ffx/process.py
View File

@@ -1,19 +1,23 @@
import shlex
import subprocess
from typing import List
from typing import Iterable, List
from .logging_utils import get_ffx_logger
def executeProcess(commandSequence: List[str], directory: str = None, context: dict = None):
COMMAND_TIMED_OUT_RETURN_CODE = 124
COMMAND_NOT_FOUND_RETURN_CODE = 127
def formatCommandSequence(commandSequence: Iterable[str]) -> str:
return shlex.join([str(token) for token in commandSequence])
def getWrappedCommandSequence(commandSequence: List[str], context: dict = None) -> List[str]:
"""
niceness -20 bis +19
cpu_percent: 1 bis 99
"""
if context is None:
logger = get_ffx_logger()
else:
logger = context['logger']
niceSequence = []
niceness = int((context or {}).get('resource_limits', {}).get('niceness', 99))
@@ -24,11 +28,72 @@ def executeProcess(commandSequence: List[str], directory: str = None, context: d
if cpu_percent >= 1:
niceSequence += ['cpulimit', '-l', str(cpu_percent), '--']
niceCommand = niceSequence + commandSequence
return niceSequence + [str(token) for token in commandSequence]
logger.debug(f"executeProcess() command sequence: {' '.join(niceCommand)}")
process = subprocess.Popen(niceCommand, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8', cwd = directory)
output, error = process.communicate()
return output, error, process.returncode
def getProcessTimeoutSeconds(context: dict = None, timeoutSeconds: float = None):
if timeoutSeconds is None:
timeoutSeconds = (context or {}).get('resource_limits', {}).get('timeout_seconds')
if timeoutSeconds is None:
return None
timeoutSeconds = float(timeoutSeconds)
return timeoutSeconds if timeoutSeconds > 0 else None
def executeProcess(
commandSequence: List[str],
directory: str = None,
context: dict = None,
timeoutSeconds: float = None,
):
logger = context['logger'] if context is not None and 'logger' in context else get_ffx_logger()
wrappedCommandSequence = getWrappedCommandSequence(commandSequence, context=context)
timeoutSeconds = getProcessTimeoutSeconds(context=context, timeoutSeconds=timeoutSeconds)
logger.debug(
"executeProcess() cwd=%s timeout=%s command=%s",
directory or '.',
timeoutSeconds if timeoutSeconds is not None else 'none',
formatCommandSequence(wrappedCommandSequence),
)
try:
completed = subprocess.run(
wrappedCommandSequence,
capture_output=True,
text=True,
cwd=directory,
timeout=timeoutSeconds,
check=False,
)
except FileNotFoundError as ex:
error = (
"Command not found while running "
+ f"{formatCommandSequence(wrappedCommandSequence)}: {ex.filename or ex}"
)
logger.error(error)
return '', error, COMMAND_NOT_FOUND_RETURN_CODE
except subprocess.TimeoutExpired as ex:
stdout = ex.stdout or ''
stderr = ex.stderr or ''
error = (
f"Command timed out after {timeoutSeconds} seconds while running "
+ formatCommandSequence(wrappedCommandSequence)
)
if stderr:
error = f"{error}\n{stderr}"
logger.error(error)
return stdout, error, COMMAND_TIMED_OUT_RETURN_CODE
if completed.returncode != 0:
logger.warning(
"executeProcess() rc=%s command=%s",
completed.returncode,
formatCommandSequence(wrappedCommandSequence),
)
return completed.stdout, completed.stderr, completed.returncode