aboutsummaryrefslogtreecommitdiff
path: root/src/avp/command.py
diff options
context:
space:
mode:
authorAeliton G. Silva2026-01-12 22:39:55 -0300
committerAeliton G. Silva2026-01-13 04:22:25 -0300
commitf975144f25d34f97329b2d4e52891061573cea08 (patch)
tree226fe223b31af6f217b1dd413629ab2cf26964d4 /src/avp/command.py
parentb8703752ffc7768b0275897b3c2a869ff41504e5 (diff)
Use pyproject.toml + uv_build
This replaces setup.py by a modern pyproject.toml using uv_build backend. Dependencies are being also managed by uv, so to install dependencies and run the project one can execute: ``` uv sync uv run pytest # optional python -m avp ``` To build the both source and binary (wheel) distribution package run: ``` uv build ``` Uv can be installed with `pip install uv`. The directory structure has been changed to reflect best practices. - src/* -> src/avp/ - src/tests -> ../tests
Diffstat (limited to 'src/avp/command.py')
-rw-r--r--src/avp/command.py316
1 files changed, 316 insertions, 0 deletions
diff --git a/src/avp/command.py b/src/avp/command.py
new file mode 100644
index 0000000..783ac26
--- /dev/null
+++ b/src/avp/command.py
@@ -0,0 +1,316 @@
+"""
+When using commandline mode, this module's object handles interpreting
+the arguments and giving them to Core, which tracks the main program state.
+Then it immediately exports a video.
+"""
+
+from PyQt6 import QtCore
+import argparse
+import os
+import sys
+import time
+import glob
+import signal
+import shutil
+import logging
+
+from . import core
+
+
+log = logging.getLogger("AVP.Commandline")
+
+
+class Command(QtCore.QObject):
+ """
+ This replaces the GUI MainWindow when in commandline mode.
+ """
+
+ createVideo = QtCore.pyqtSignal()
+
+ def __init__(self):
+ super().__init__()
+ self.core = core.Core()
+ core.Core.mode = "commandline"
+ self.dataDir = self.core.dataDir
+ self.canceled = False
+ self.settings = core.Core.settings
+
+ # ctrl-c stops the export thread
+ signal.signal(signal.SIGINT, self.stopVideo)
+
+ def parseArgs(self):
+ parser = argparse.ArgumentParser(
+ prog="avp" if os.path.basename(sys.argv[0]) == "__main__.py" else None,
+ description="Create a visualization for an audio file",
+ epilog="EXAMPLE COMMAND: avp myvideotemplate "
+ "-i ~/Music/song.mp3 -o ~/video.mp4 "
+ "-c 0 image path=~/Pictures/thisWeeksPicture.jpg "
+ '-c 1 video "preset=My Logo" -c 2 vis layout=classic',
+ )
+
+ # input/output automatic-export commands
+ parser.add_argument("-i", "--input", metavar="SOUND", help="input audio file")
+ parser.add_argument(
+ "-o", "--output", metavar="OUTPUT", help="output video file"
+ )
+ parser.add_argument(
+ "--export-project",
+ action="store_true",
+ help="use input and output files from project file if -i or -o is missing",
+ )
+
+ # mutually exclusive debug options
+ debugCommands = parser.add_mutually_exclusive_group()
+ debugCommands.add_argument(
+ "--test",
+ action="store_true",
+ help="run tests and create a report full of debugging info",
+ )
+ debugCommands.add_argument(
+ "--debug",
+ action="store_true",
+ help="create bigger logfiles while program is running",
+ )
+
+ # project/GUI options
+ parser.add_argument(
+ "projpath",
+ metavar="path-to-project",
+ help="open a project file (.avp)",
+ nargs="?",
+ )
+ parser.add_argument(
+ "-c",
+ "--comp",
+ metavar=("LAYER", "ARG"),
+ help="first arg must be component NAME to insert at LAYER."
+ '"help" for information about possible args for a component.',
+ nargs="*",
+ action="append",
+ )
+ parser.add_argument(
+ "--no-preview",
+ action="store_true",
+ help="disable live preview during export",
+ )
+
+ args = parser.parse_args()
+
+ if args.debug:
+ core.FILE_LOGLVL = logging.DEBUG
+ core.STDOUT_LOGLVL = logging.DEBUG
+ core.Core.makeLogger()
+
+ if args.test:
+ self.runTests()
+ quit(0)
+
+ if args.projpath:
+ projPath = args.projpath
+ if not os.path.dirname(projPath):
+ projPath = os.path.join(self.settings.value("projectDir"), projPath)
+ if not projPath.endswith(".avp"):
+ projPath += ".avp"
+ self.core.openProject(self, projPath)
+ self.core.selectedComponents = list(reversed(self.core.selectedComponents))
+ self.core.componentListChanged()
+
+ if args.comp:
+ for comp in args.comp:
+ pos = comp[0]
+ name = comp[1]
+ compargs = comp[2:]
+ try:
+ pos = int(pos)
+ except ValueError:
+ print(pos, "is not a layer number.")
+ quit(1)
+ realName = self.parseCompName(name)
+ if not realName:
+ print(name, "is not a valid component name.")
+ quit(1)
+ modI = self.core.moduleIndexFor(realName)
+ i = self.core.insertComponent(pos, modI, self)
+ if i is None:
+ print(name, "could not be initialized.")
+ quit(1)
+ for arg in compargs:
+ self.core.selectedComponents[i].command(arg)
+
+ if args.export_project and args.projpath:
+ errcode, data = self.core.parseAvFile(projPath)
+ input_ = None
+ output = None
+ for key, value in data["WindowFields"]:
+ if "outputFile" in key:
+ output = value
+ if output and not os.path.dirname(value):
+ output = os.path.join(os.path.expanduser("~"), output)
+ if "audioFile" in key:
+ input_ = value
+
+ # use input/output from project file, overwritten by -i and -o
+ if (not input_ and not args.input) or (not output and not args.output):
+ parser.print_help()
+ quit(1)
+
+ self.createAudioVisualization(
+ input_ if not args.input else args.input,
+ output if not args.output else args.output,
+ )
+ return "commandline"
+
+ elif args.input and args.output:
+ self.createAudioVisualization(args.input, args.output)
+ return "commandline"
+
+ elif args.no_preview:
+ core.Core.previewEnabled = False
+
+ elif (
+ args.projpath is None
+ and "help" not in sys.argv
+ and "--debug" not in sys.argv
+ and "--test" not in sys.argv
+ ):
+ parser.print_help()
+ quit(1)
+
+ return "GUI"
+
+ def createAudioVisualization(self, input, output):
+ if not self.core.selectedComponents:
+ print("No components selected. Adding a default visualizer.")
+ time.sleep(1)
+ self.core.insertComponent(0, 0, self)
+ self.core.selectedComponents = list(reversed(self.core.selectedComponents))
+ self.core.componentListChanged()
+ self.worker = self.core.newVideoWorker(self, input, output)
+ # quit(0) after video is created
+ self.worker.videoCreated.connect(self.videoCreated)
+ self.lastProgressUpdate = time.time()
+ self.worker.progressBarSetText.connect(self.progressBarSetText)
+ self.createVideo.emit()
+
+ def stopVideo(self, *args):
+ self.worker.error = True
+ self.worker.cancelExport()
+ self.worker.cancel()
+
+ @QtCore.pyqtSlot(str)
+ def progressBarSetText(self, value):
+ if "Export " in value:
+ # Don't duplicate completion/failure messages
+ return
+ if (
+ not value.startswith("Exporting")
+ and time.time() - self.lastProgressUpdate >= 0.05
+ ):
+ # Show most messages very often
+ print(value)
+ elif time.time() - self.lastProgressUpdate >= 2.0:
+ # Give user time to read ffmpeg's output during the export
+ print("##### %s" % value)
+ else:
+ return
+ self.lastProgressUpdate = time.time()
+
+ @QtCore.pyqtSlot()
+ def videoCreated(self):
+ self.quit(0)
+
+ def quit(self, code):
+ quit(code)
+
+ def showMessage(self, **kwargs):
+ print(kwargs["msg"])
+ if "detail" in kwargs:
+ print(kwargs["detail"])
+
+ @QtCore.pyqtSlot(str, str)
+ def videoThreadError(self, msg, detail):
+ print(msg)
+ print(detail)
+ quit(1)
+
+ def drawPreview(self, *args):
+ pass
+
+ def parseCompName(self, name):
+ """Deduces a proper component name out of a commandline arg"""
+
+ if name.title() in self.core.compNames:
+ return name.title()
+ for compName in self.core.compNames:
+ if name.capitalize() in compName:
+ return compName
+
+ compFileNames = [
+ os.path.splitext(os.path.basename(mod.__file__))[0]
+ for mod in self.core.modules
+ ]
+ for i, compFileName in enumerate(compFileNames):
+ if name.lower() in compFileName:
+ return self.core.compNames[i]
+ return
+
+ return None
+
+ def runTests(self):
+ from . import tests
+
+ test_report = os.path.join(core.Core.logDir, "test_report.log")
+ tests.run(test_report)
+
+ # Choose a numbered location to put the output file
+ logNumber = 0
+
+ def getFilename():
+ """Get a numbered filename for the final test report"""
+ nonlocal logNumber
+ name = os.path.join(os.path.expanduser("~"), "avp_test_report")
+ while True:
+ possibleName = f"{name}{logNumber:0>2}.txt"
+ if os.path.exists(possibleName) and logNumber < 100:
+ logNumber += 1
+ continue
+ break
+ return possibleName
+
+ # Copy latest debug log to chosen test report location
+ filename = getFilename()
+ if logNumber == 100:
+ print("Test Report could not be created.")
+ return
+ try:
+ shutil.copy(os.path.join(core.Core.logDir, "avp_debug.log"), filename)
+ with open(filename, "a") as f:
+ f.write(f"{'='*60} debug log ends {'='*60}\n")
+ except FileNotFoundError:
+ with open(filename, "w") as f:
+ f.write(f"{'='*60} no debug log {'='*60}\n")
+
+ def concatenateLogs(logPattern):
+ nonlocal filename
+ renderLogs = glob.glob(os.path.join(core.Core.logDir, logPattern))
+ with open(filename, "a") as fw:
+ for renderLog in renderLogs:
+ with open(renderLog, "r") as fr:
+ fw.write(f"{'='*60} {os.path.basename(renderLog)} {'='*60}\n")
+ logContents = fr.readlines()
+ fw.write("".join(logContents[:5]))
+ fw.write("...trimmed...\n")
+ fw.write("".join(logContents[-10:]))
+ fw.write(f"{'='*60} {os.path.basename(renderLog)} {'='*60}\n")
+
+ concatenateLogs("render_*.log")
+ concatenateLogs("preview_*.log")
+
+ # Append actual test report to debug log
+ with open(test_report, "r") as f:
+ output = f.readlines()
+ test_output = "".join(output)
+ print(test_output)
+ with open(filename, "a") as f:
+ f.write(test_output)
+ print(f"Test Report created at {filename}")