path: root/src/avp/command.py

"""
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, __version__


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',
        )

        parser.add_argument(
            "--version", "-V", action="version", version=f"%(prog)s {__version__}"
        )

        # 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(
            "--log",
            action="store_true",
            help="copy and shorten recent log files into ~/avp_log.txt",
        )
        debugCommands.add_argument(
            "--verbose", "-v",
            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.verbose:
            core.STDOUT_LOGLVL = logging.DEBUG
            core.Core.makeLogger(deleteOldLogs=False, fileLogLvl=logging.DEBUG)

        if args.log:
            self.createLogFile()
            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 "--verbose" not in sys.argv
            and "-v" not in sys.argv
            and "--log" 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 None

    def createLogFile(self):
        # Choose a numbered location to put the output file
        logNumber = 0

        def getFilename():
            """Get a numbered filename for the final log file"""
            nonlocal logNumber
            name = os.path.join(os.path.expanduser("~"), "avp_log")
            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 log file location
        filename = getFilename()
        if logNumber == 100:
            print("Log file could not be created (too many exist).")
            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")
        print(f"Log file created at {filename}")