Best Python code snippet using molecule_python
ffmpeg.py
Source:ffmpeg.py  
1""" Plugin that uses ffmpeg to read and write series of images to2a wide range of video formats.3"""4# Heavily inspired from Almar Klein's imageio code5# Copyright (c) 2015, imageio contributors6# distributed under the terms of the BSD License (included in release).7import subprocess as sp8import cv29import numpy as np10from .abstract import VideoReaderAbstract, VideoWriterAbstract11from .ffprobe import ffprobe12from .. import _FFMPEG_APPLICATION13from .. import _FFMPEG_PATH14from .. import _FFMPEG_SUPPORTED_DECODERS15from .. import _FFMPEG_SUPPORTED_ENCODERS16from .. import _HAS_FFMPEG17from ..utils import *18startupinfo = sp.STARTUPINFO()19startupinfo.dwFlags = sp.CREATE_NEW_CONSOLE | sp.STARTF_USESHOWWINDOW20startupinfo.wShowWindow = sp.SW_HIDE21# uses FFmpeg to read the given file with parameters22class FFmpegReader(VideoReaderAbstract):23    """Reads frames using FFmpeg24    Using FFmpeg as a backend, this class25    provides sane initializations meant to26    handle the default case well.27    """28    INFO_AVERAGE_FRAMERATE = "@r_frame_rate"29    INFO_WIDTH = "@width"30    INFO_HEIGHT = "@height"31    INFO_PIX_FMT = "@pix_fmt"32    INFO_DURATION = "@duration"33    INFO_NB_FRAMES = "@nb_frames"34    OUTPUT_METHOD = "image2pipe"35    def __init__(self, *args, **kwargs):36        assert _HAS_FFMPEG, "Cannot find installation of real FFmpeg (which comes with ffprobe)."37        super(FFmpegReader, self).__init__(*args, **kwargs)38    def _createProcess(self, inputdict, outputdict, verbosity):39        if '-vcodec' not in outputdict:40            outputdict['-vcodec'] = "rawvideo"41        iargs = self._dict2Args(inputdict)42        oargs = self._dict2Args(outputdict)43        if verbosity > 0:44            cmd = [_FFMPEG_PATH + "/" + _FFMPEG_APPLICATION] + iargs + ['-i', self._filename] + oargs + ['-']45            try:46                print("FFmpeg Read Command:", " ".join(cmd))47            except UnicodeEncodeError:48                print("FFmpeg Read Command: NON-ASCII character exists in command, not shown")49            self._proc = sp.Popen(cmd, stdin=sp.PIPE,50                                  stdout=sp.PIPE, stderr=None, startupinfo=startupinfo)51        else:52            cmd = [_FFMPEG_PATH + "/" + _FFMPEG_APPLICATION, "-nostats", "-loglevel", "0"] + iargs + ['-i',53                                                                                                      self._filename] + oargs + [54                      '-']55        self._proc = sp.Popen(cmd, stdin=sp.PIPE,56                              stdout=sp.PIPE, stderr=sp.PIPE, startupinfo=startupinfo)57        self._cmd = " ".join(cmd)58    def _probCountFrames(self):59        # open process, grabbing number of frames using ffprobe60        probecmd = [_FFMPEG_PATH + "/ffprobe"] + ["-v", "error", "-count_frames", "-select_streams", "v:0",61                                                  "-show_entries", "stream=nb_read_frames", "-of",62                                                  "default=nokey=1:noprint_wrappers=1", self._filename]63        return np.int(check_output(probecmd).decode().split('\n')[0])64    def _probe(self):65        return ffprobe(self._filename)66    def _getSupportedDecoders(self):67        return _FFMPEG_SUPPORTED_DECODERS68startupinfo = sp.STARTUPINFO()69startupinfo.dwFlags = sp.CREATE_NEW_CONSOLE | sp.STARTF_USESHOWWINDOW70startupinfo.wShowWindow = sp.SW_HIDE71class FFmpegWriter(VideoWriterAbstract):72    """Writes frames using FFmpeg73    Using FFmpeg as a backend, this class74    provides sane initializations for the default case.75    """76    def __init__(self, *args, **kwargs):77        assert _HAS_FFMPEG, "Cannot find installation of real FFmpeg (which comes with ffprobe)."78        super(FFmpegWriter, self).__init__(*args, **kwargs)79    def _getSupportedEncoders(self):80        return _FFMPEG_SUPPORTED_ENCODERS81    def _createProcess(self, inputdict, outputdict, verbosity):82        iargs = self._dict2Args(inputdict)83        oargs = self._dict2Args(outputdict)84        cmd = [_FFMPEG_PATH + "/" + _FFMPEG_APPLICATION, "-y"] + iargs + ["-i", "-"] + oargs + [self._filename]85        self._cmd = " ".join(cmd)86        # Launch process87        if self.verbosity > 0:88            try:89                print("FFmpeg Write Command:", self._cmd)90            except UnicodeEncodeError:91                print("FFmpeg Write Command: NON-ASCII character exists in command, not shown")92            self._proc = sp.Popen(cmd, stdin=sp.PIPE,93                                  stdout=sp.PIPE, stderr=None, startupinfo=startupinfo)94        else:95            self._proc = sp.Popen(cmd, stdin=sp.PIPE,96                                  stdout=self.DEVNULL, stderr=sp.STDOUT, startupinfo=startupinfo)97class EnccWriter(VideoWriterAbstract):98    """Writes frames using NVencc99    Using NVencc as a backend, this class100    provides sane initializations for the default case.101    """102    def __init__(self, *args, **kwargs):103        assert _HAS_FFMPEG, "Cannot find installation of real NVencc (which comes with ffmpeg)."104        super(EnccWriter, self).__init__(*args, **kwargs)105        self.bit_depth = 8106    def _getSupportedEncoders(self):107        return _FFMPEG_SUPPORTED_ENCODERS108    def _createProcess(self, inputdict, outputdict, verbosity):109        if inputdict['encc'] == "NVENCC":110            _ENCC_APPLICATION = "NVEncC64"111        else:112            _ENCC_APPLICATION = "QSVEncC64"113        _ENCC_APPLICATION += ".exe"114        inputdict.pop('encc')115        n_inputdict = self._dealWithFFmpegArgs(inputdict)116        n_outputdict = self._dealWithFFmpegArgs(outputdict)117        if '-s' in inputdict:118            n_inputdict.update({'--input-res': inputdict['-s']})119            n_inputdict.pop('-s')120        if '--output-depth' in outputdict:121            if outputdict['--output-depth'] == '10':122                self.bit_depth = 10123        """124        !!!Attention!!!125        "--input-csp", "yv12" Yes126        "--input-csp yv12" No127        """128        iargs = self._dict2Args(inputdict)129        oargs = self._dict2Args(outputdict)130        n_iargs = self._dict2Args(n_inputdict)131        n_oargs = self._dict2Args(n_outputdict)132        cmd = [_FFMPEG_PATH + "/" + _ENCC_APPLICATION] + ["--raw", "--input-csp", "yv12"] + n_iargs + ["-i",133                                                                                                       "-", ] + n_oargs + [134                  "-o"] + [135                  self._filename]136        # cmd = ['D:\\60-fps-Project\\ffmpeg/QSVEncC64.exe', '--raw', '--input-csp', 'yv12', '--fps', '24.0',137        #        '--input-res', '1920x1080', '-i', '-', '--la-depth', '50', '--la-quality', 'slow', '--extbrc', '--mbbrc',138        #        '--i-adapt', '--b-adapt', '--gop-len', '250', '-b', '6', '--ref', '8', '--b-pyramid', '--weightb',139        #        '--weightp', '--adapt-ltr', '--colorrange', 'tv', '--colormatrix', 'bt709', '--transfer', 'bt709',140        #        '--colorprim', 'bt709', '-c', 'hevc', '--profile', 'main', '--tier', 'main', '--sao', 'luma', '--ctu',141        #        '64', '--la-icq', '16', '-o',142        #        'D:\\60-fps-Project\\input_or_ref\\Test\\è¾åºoutput\\ã3ãæ¹å¤ç1_SVFI_Render_847013.mp4']143        # H264 8bit QSVEncC64144        # cmd = ['D:\\60-fps-Project\\ffmpeg/QSVEncC64.exe', '--raw', '--input-csp', 'yv12', '--fps', '24.0', '--input-res', '1920x1080', '-i', '-',145        #         '--la-depth', '50', '--la-quality', 'slow', '--extbrc', '--mbbrc', '--i-adapt', '--b-adapt', '--gop-len', '250',146        #         '-b', '6', '--ref', '8', '--b-pyramid', '--weightb', '--weightp', '--adapt-ltr',147        #         '--colorrange', 'tv', '--colormatrix', 'bt709', '--transfer', 'bt709', '--colorprim', 'bt709',148        #         '-c', 'h264', '--profile', 'high', '--repartition-check', '--trellis', 'all', 149        #         '--la-icq', '16',150        #        '-o','D:\\60-fps-Project\\input_or_ref\\Test\\è¾åºoutput\\ã3ãæ¹å¤ç1_SVFI_Render_847013.mp4']151        self._cmd = " ".join(cmd)152        # Launch process153        if self.verbosity > 0:154            try:155                print("EnCc Write Command:", self._cmd)156            except UnicodeEncodeError:157                print("EnCc Write Command: NON-ASCII character exists in command, not shown")158            self._proc = sp.Popen(cmd, stdin=sp.PIPE,159                                  stdout=sp.PIPE, stderr=None, startupinfo=startupinfo)160        else:161            self._proc = sp.Popen(cmd, stdin=sp.PIPE,162                                  stdout=self.DEVNULL, stderr=sp.STDOUT, startupinfo=startupinfo, shell=True)163    def writeFrame(self, im: np.ndarray):164        """Sends ndarray frames to NVENCC165        """166        vid = vshape(im)167        T, M, N, C = vid.shape168        if not self.warmStarted:169            self._warmStart(M, N, C, im.dtype)170        assert self._proc is not None  # Check status171        try:172            im = im.astype(np.uint8)173            vid = cv2.cvtColor(im, cv2.COLOR_BGR2YUV_YV12)174            # if self.bit_depth == 10:175            #     vid <<= 2176            self._proc.stdin.write(vid.tostring())177        except IOError as e:178            # Show the command and stderr from pipe179            msg = '{0:}\n\nENCODE COMMAND:\n{1:}\n\nENCODE STDERR ' \180                  'OUTPUT:{2:}\n'.format(e, self._cmd, sp.STDOUT)181            raise IOError(msg)182    def _dealWithFFmpegArgs(self, args: dict):183        input_args = args.copy()184        pop_list = ['-f', '-pix_fmt']185        for p in pop_list:186            if p in input_args:187                input_args.pop(p)188        return input_args189        pass190class SVTWriter(EnccWriter):191    """Writes frames using SVT192    Using SVT as a backend, this class193    provides sane initializations for the default case.194    """195    def __init__(self, *args, **kwargs):196        assert _HAS_FFMPEG, "Cannot find installation of real SVT (which comes with ffmpeg)."197        super(SVTWriter, self).__init__(*args, **kwargs)198    def _createProcess(self, inputdict, outputdict, verbosity):199        if outputdict['encc'] == "hevc":200            _SVT_APPLICATION = "SvtHevcEncApp"201        elif outputdict['encc'] == "vp9":202            _SVT_APPLICATION = "SvtVp9EncApp"203        else:  # av1204            _SVT_APPLICATION = "SvtAv1EncApp"205        _SVT_APPLICATION += ".exe"206        outputdict.pop('encc')207        n_inputdict = self._dealWithFFmpegArgs(inputdict)208        n_outputdict = self._dealWithFFmpegArgs(outputdict)209        if '-s' in inputdict:210            input_resolution = inputdict['-s'].split('x')211            n_inputdict.update({'-w': input_resolution[0], '-h': input_resolution[1]})212            n_inputdict.pop('-s')213        if '-s' in outputdict:214            n_outputdict.pop('-s')215        if '-bit-depth' in outputdict:216            if outputdict['-bit-depth'] == '10':217                self.bit_depth = 10218        """219        !!!Attention!!!220        "--input-csp", "yv12" Yes221        "--input-csp yv12" No222        """223        iargs = self._dict2Args(inputdict)224        oargs = self._dict2Args(outputdict)225        n_iargs = self._dict2Args(n_inputdict)226        n_oargs = self._dict2Args(n_outputdict)227        # _cmd = [_FFMPEG_PATH + "/" + _FFMPEG_APPLICATION] + ["-i", "-"]228        # if outputdict['-bit-depth'] in ['8']:229        #     _cmd += ["-pix_fmt", "yuv420p"]230        # else:231        #     """10bit"""232        #     _cmd += ["-pix_fmt", "yuv420p10le"]233        # _cmd += ["-f", "rawvideo", "-", "|"]234        cmd = [_FFMPEG_PATH + "/" + _SVT_APPLICATION] + ["-i", "stdin"] + n_iargs + n_oargs + ["-b",235                                                                                                           self._filename]236        self._cmd = " ".join(cmd)237        # self._cmd = r"D:\60-fps-Project\ffmpeg\ffmpeg.exe -i D:\60-fps-Project\input_or_ref\Test\ã4ãæåº+é»è¾¹è£å+æ¶é¿ç段+åå¹è½¨åå¹¶.mkv -pix_fmt yuv420p10le -f rawvideo - | D:\60-fps-Project\ffmpeg\SvtHevcEncApp.exe -i stdin -fps 25 -n 241 -w 3840 -h 2160 -brr 1 -sharp 1 -q 16 -bit-depth 10 -b D:\60-fps-Project\input_or_ref\Test\svt_output.mp4"238        # cmd = [_FFMPEG_PATH + "/" + _SVT_APPLICATION] + ["-i", "stdin"] + n_iargs + n_oargs + ["-b", self._filename]239        # self._cmd = " ".join(cmd)240        # Launch process241        if self.verbosity > 0:242            try:243                print("SVT Write Command:", self._cmd)244            except UnicodeEncodeError:245                print("SVT Write Command: NON-ASCII character exists in command, not shown")246            self._proc = sp.Popen(cmd, stdin=sp.PIPE,247                                  stdout=sp.PIPE, stderr=None, startupinfo=startupinfo)248        else:249            self._proc = sp.Popen(cmd, stdin=sp.PIPE,...test_rseek.py
Source:test_rseek.py  
...10DATA_TSAMP = 256e-611PARSER = get_parser()12EXPECTED_COLUMNS = {'period', 'freq', 'width', 'ducy', 'dm', 'snr'}13DEFAULT_OPTIONS = dict(Pmin=0.5, Pmax=2.0, bmin=480, bmax=520, smin=7.0, format='presto')14def dict2args(d):15    """16    Convert dictionary of options to command line argument list17    """18    args = []19    for k, v in d.items():20        args.append(f'--{k}')21        args.append(str(v))22    return args23def test_rseek_fakepsr():24    with tempfile.TemporaryDirectory() as outdir:25        generate_data_presto(26            outdir, 'data', tobs=DATA_TOBS, tsamp=DATA_TSAMP, period=SIGNAL_PERIOD,27            dm=0.0, amplitude=20.0, ducy=0.0228        )29        fname = os.path.join(outdir, 'data.inf')30        cmdline_args = dict2args(DEFAULT_OPTIONS) + [fname]31        args = PARSER.parse_args(cmdline_args)32        df = run_program(args)33    assert df is not None34    assert set(df.columns) == EXPECTED_COLUMNS35    # Results must be sorted in decreasing S/N order36    assert all(df.snr == df.sort_values('snr', ascending=False).snr)37    # Check parameters of the top candidate38    # NOTE: these checks depend on the RNG seed and the program options39    topcand = df.iloc[0]40    assert abs(topcand.freq - SIGNAL_FREQ) < 0.1 / DATA_TOBS41    assert abs(topcand.snr - 18.5) < 0.1542    assert topcand.dm == 043    assert topcand.width == 1344    45def test_rseek_purenoise():46    with tempfile.TemporaryDirectory() as outdir:47        generate_data_presto(48            outdir, 'data', tobs=DATA_TOBS, tsamp=DATA_TSAMP, period=SIGNAL_PERIOD,49            dm=0.0, amplitude=0.050        )51        fname = os.path.join(outdir, 'data.inf')52        cmdline_args = dict2args(DEFAULT_OPTIONS) + [fname]53        args = PARSER.parse_args(cmdline_args)54        df = run_program(args)55    ...test_util.py
Source:test_util.py  
...6    ({"basename": "base", "name": "sub"}, "base:sub"),7])8def test_normalize_task_name(task, expected):9    assert di.util.normalize_task_name(task) == expected10def test_dict2args():11    assert di.dict2args({"hello": "world"}, foo="bar") == ["--hello=world", "--foo=bar"]12    with pytest.raises(ValueError):...Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
