def _record_perf_async(
        loop: asyncio.BaseEventLoop, event: str, message: str) -> None:
    """Record timing metric async
    :param asyncio.BaseEventLoop loop: event loop
    :param str event: event
    :param str message: message
    """
    if not _RECORD_PERF:
        return
    proc = await asyncio.create_subprocess_shell(
        './perf.py cascade {ev} --prefix {pr} --message "{msg}"'.format(
            ev=event, pr=_PREFIX, msg=message), loop=loop)
    await proc.wait()
    if proc.returncode != 0:
        logger.error(
            'could not record perf to storage for event: {}'.format(event)) 

def start(self):
        if self.is_active:
            print(f"[Cluster {self.id}] Already active.")
            return
        self.started_at = time.time()
        self._process = await asyncio.create_subprocess_shell(
            self.command,
            stdin=asyncio.subprocess.DEVNULL,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            preexec_fn=os.setsid,
            limit=1024 * 256,
        )
        self.status = "running"
        self.started_at = time.time()
        print(f"[Cluster {self.id}] The cluster is starting.")
        await asyncio.wait([self.read_stream(self._process.stdout), self.read_stream(self._process.stderr)])
        return self 

def wait_for_pulseaudio_startup(self, vm):
        self.loop.run_until_complete(
            self.wait_for_session(self.testvm1))
        try:
            self.loop.run_until_complete(vm.run_for_stdio(
                "timeout 30s sh -c 'while ! pactl info; do sleep 1; done'"
            ))
        except subprocess.CalledProcessError as e:
            self.fail('Timeout waiting for pulseaudio start in {}: {}{}'.format(
                vm.name, e.stdout, e.stderr))
        # then wait for the stream to appear in dom0
        local_user = grp.getgrnam('qubes').gr_mem[0]
        p = self.loop.run_until_complete(asyncio.create_subprocess_shell(
            "sudo -E -u {} timeout 30s sh -c '"
            "while ! pactl list sink-inputs | grep -q :{}; do sleep 1; done'".format(
                local_user, vm.name)))
        self.loop.run_until_complete(p.wait())
        # and some more...
        self.loop.run_until_complete(asyncio.sleep(1)) 

def shell(self, request, command, communicate=None,
                    raw=False, chdir=None):
        """Asynchronous execution of a shell command
        """
        if chdir:
            command = 'cd %s && %s' % (chdir, command)
        request.logger.info('Execute shell command: %s', command)
        stdin = subprocess.PIPE if communicate else None
        proc = await create_subprocess_shell(command,
                                             stdin=stdin,
                                             stdout=subprocess.PIPE,
                                             stderr=subprocess.PIPE)
        if communicate:
            msg, err = await proc.communicate(to_bytes(communicate))
        else:
            await proc.wait()
            msg = await proc.stdout.read()
            err = await proc.stderr.read()
        if proc.returncode:
            err = err.decode('utf-8').strip()
            raise ShellError(err, proc.returncode)
        return msg if raw else msg.decode('utf-8').strip() 

def compile(self, prior_output: FilePath, compile_command: str) -> FilePath:
        # prior_output is the location where our new file will be created after compiling
        # compile_command is the thing we're going to execute (hopefully after some pre-processing is done)
        proc = await asyncio.create_subprocess_shell(compile_command,
                                                     stdout=asyncio.subprocess.PIPE,
                                                     stderr=asyncio.subprocess.PIPE,
                                                     cwd=self.working_dir)
        stdout, stderr = await proc.communicate()
        if stdout:
            print(f'[stdout]\n{stdout.decode()}')
        if stderr:
            print(f'[stderr]\n{stderr.decode()}')
            raise Exception(stderr.decode())
        # we return the status (in case that's something you want to print out) and where the new file is located
        print("called compile and returned final path of: {}".format(prior_output))
        return FilePath(prior_output) 

def call_method(
            self, method, arg_signature, args, return_signature, returns
    ):
        cmd = 'busctl --user -- call '
        cmd += f'{service_name} {object_path} {object_interface} {method}'
        cmd += f' "{arg_signature}"'
        for i in args:
            cmd += f' {i}'

        create = asyncio.create_subprocess_shell(
                cmd, stdout=asyncio.subprocess.PIPE
        )

        proc = await create

        # Read one line of output
        data = await proc.stdout.readline()
        line = data.decode('ascii').rstrip()

        self.assertEqual(line, f'{return_signature} {returns}')

        await proc.wait() 

def cat(loop):

    #
    # 异步返回:
    #   - 调用接口:
    #
    proc = yield from asyncio.create_subprocess_shell("cat",
                                                      stdin=PIPE,
                                                      stdout=PIPE)
    print("pid: %s" % proc.pid)

    message = "Hello World!"
    print("cat write: %r" % message)

    stdout, stderr = yield from proc.communicate(message.encode('ascii'))
    print("cat read: %r" % stdout.decode('ascii'))

    exitcode = yield from proc.wait()
    print("(exit code %s)" % exitcode) 

def aria_start(event):
    process = await asyncio.create_subprocess_shell(
        aria2_daemon_start_cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await process.communicate()
    global aria2
    aria2 = aria2p.API(
        aria2p.Client(
            host="http://localhost",
            port=ARIA2_STARTED_PORT,
            secret=""
        )
    )
    OUTPUT = f"**ARIA TWO C:**\n__PID:__\n`{process.pid}`\n\n**ARIA TWO STARTED**"
    await event.edit(OUTPUT) 

def sysdetails(sysd):
    """ a. """
    if not sysd.text[0].isalpha() and sysd.text[0] not in ("/", "#", "@", "!"):
        try:
            neo = "neofetch/neofetch --on --color_blocks on --bold on --cpu_temp=C \
                    --cpu_speed on --cpu_cores physical --kernel_shorthand on \
                    --gpu_brand on --refresh_rate on --gtk_shorthand on --colors=distro  --backend ascii \
                    --source=auto --Redhat source --stdout"
            fetch = await asyncrunapp(
                neo,
                stdout=asyncPIPE,
                stderr=asyncPIPE,
            )

            stdout, stderr = await fetch.communicate()
            result = str(stdout.decode().strip()) \
                + str(stderr.decode().strip())

            await sysd.edit("sysd Result: `" + result + "`")
        except FileNotFoundError:
            await sysd.edit("`Hey, on mkaraniya/BotHub install .neofetch first kthx`") 

def runcmd(self, message, cmd, editor=None):
        if len(cmd.split(" ")) > 1 and cmd.split(" ")[0] == "sudo":
            needsswitch = True
            for word in cmd.split(" ", 1)[1].split(" "):
                if word[0] != "-":
                    break
                if word == "-S":
                    needsswitch = False
            if needsswitch:
                cmd = " ".join([cmd.split(" ", 1)[0], "-S", cmd.split(" ", 1)[1]])
        sproc = await asyncio.create_subprocess_shell(cmd, stdin=asyncio.subprocess.PIPE,
                                                      stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
                                                      cwd=utils.get_base_dir())
        if editor is None:
            editor = SudoMessageEditor(message, cmd, self.config)
        editor.update_process(sproc)
        self.activecmds[hash_msg(message)] = sproc
        await editor.redraw(True)
        await asyncio.gather(read_stream(editor.update_stdout, sproc.stdout, self.config["FLOOD_WAIT_PROTECT"]),
                             read_stream(editor.update_stderr, sproc.stderr, self.config["FLOOD_WAIT_PROTECT"]))
        await editor.cmd_ended(await sproc.wait())
        del self.activecmds[hash_msg(message)] 

def start(self):
        """ Launch the task and readers of stdout, stderr """
        try:
            self.command = ['amass', 'enum' '-d'] + [self.target] + [self.params['program']['argv']]
            print(' '.join(self.command))
            self.proc = await asyncio.create_subprocess_shell(' '.join(self.command), stdout=PIPE, stderr=PIPE)

            # 1337 is hardcoded to show frontend that we don't track progress here
            await self.set_status("Working", progress=1337)
    
            # Launch readers
            loop = asyncio.get_event_loop()
            loop.create_task(self.read_stdout())
            loop.create_task(self.read_stderr())

        except Exception as exc:
            print(str(exc))
            self.logger.error(str(exc))
            await self.set_status("Aborted", 0, str(exc)) 

def start(self):
        """ Launch the task and readers of stdout, stderr """
        print(self.params, self.target)
        try:
            self.command = ['sudo', 'masscan'] + [','.join(self.target)] + ['-oX', '-'] + self.params['program']

            self.proc = await asyncio.create_subprocess_shell(' '.join(self.command), stdout=PIPE, stderr=PIPE)

            await self.set_status("Working", progress=0)
    
            # Launch readers
            loop = asyncio.get_event_loop()
            loop.create_task(self.read_stdout())
            loop.create_task(self.read_stderr())

            # Launch status poller
            loop.create_task(self.spawn_status_poller())
        except Exception as exc:
            await self.set_status("Aborted", 0, str(exc)) 

def _(event):
    if event.fwd_from:
        return
    await event.edit("Processing ...")
    PROCESS_RUN_TIME = 100
    input_str = event.pattern_match.group(1)
    selected_transfer = event.pattern_match.group(2)
    if input_str:
        file_name = input_str
    else:
        reply = await event.get_reply_message()
        file_name = await bot.download_media(reply.media, Var.TEMP_DOWNLOAD_DIRECTORY)
    reply_to_id = event.message.id
    CMD_WEB = {"anonfiles": "curl -F \"file=@{}\" https://anonfiles.com/api/upload", "transfer": "curl --upload-file \"{}\" https://transfer.sh/{os.path.basename(file_name)}", "filebin": "curl -X POST --data-binary \"@test.png\" -H \"filename: {}\" \"https://filebin.net\"", "anonymousfiles": "curl -F file=\"@{}\" https://api.anonymousfiles.io/", "megaupload": "curl -F \"file=@{}\" https://megaupload.is/api/upload", "bayfiles": ".exec curl -F \"file=@{}\" https://bayfiles.com/api/upload"}
    try:
        selected_one = CMD_WEB[selected_transfer].format(file_name)
    except KeyError:
        await event.edit("Invalid selected Transfer")
    cmd = selected_one
    start_time = time.time() + PROCESS_RUN_TIME
    process = await asyncio.create_subprocess_shell(
        cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await process.communicate()
    await event.edit(f"{stdout.decode()}") 

def output_of(*cmd: Optional[str], **kwargs) -> str:
    """Invokes a subprocess and returns its output as a string.

    Args:
        cmd: Components of the command to execute, e.g. ["echo", "dog"].
        **kwargs: Extra arguments for asyncio.create_subprocess_shell, such as
            a cwd (current working directory) argument.

    Returns:
        A (captured output, captured error output, return code) triplet. The
        captured outputs will be None if the out or err parameters were not set
        to an instance of TeeCapture.

    Raises:
         subprocess.CalledProcessError: The process returned a non-zero error
            code and raise_on_fail was set.
    """
    result = cast(str, run_cmd(*cmd,
                               log_run_to_stderr=False,
                               out=TeeCapture(),
                               **kwargs).out)

    # Strip final newline.
    if result.endswith('\n'):
        result = result[:-1]

    return result 

def close(self) -> None:
        try:
            p = await asyncio.create_subprocess_shell('git pull')
            await p.wait()
            p = await asyncio.create_subprocess_shell(f'{sys.executable} -m pip install -U -r requirements.txt --no-cache')
            await p.wait()
        except Exception as c: # pylint: disable=broad-except
            repo.create_issue('Bot error while closing', 'discord user', 'discordbot', 'PennyDreadfulMTG/perf-reports', exception=c)
        await super().close() 

def run_cmd(semaphore, cmd, target, tag='?', patterns=[]):
    async with semaphore:
        address = target.address
        scandir = target.scandir

        info('Running task {bgreen}{tag}{rst} on {byellow}{address}{rst}' + (' with {bblue}{cmd}{rst}' if verbose >= 1 else ''))

        async with target.lock:
            with open(os.path.join(scandir, '_commands.log'), 'a') as file:
                file.writelines(e('{cmd}\n\n'))

        start_time = time.time()
        process = await asyncio.create_subprocess_shell(cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, executable='/bin/bash')
        async with target.lock:
            target.running_tasks.append(tag)

        await asyncio.wait([
            read_stream(process.stdout, target, tag=tag, patterns=patterns),
            read_stream(process.stderr, target, tag=tag, patterns=patterns, color=Fore.RED)
        ])

        await process.wait()
        async with target.lock:
            target.running_tasks.remove(tag)
        elapsed_time = calculate_elapsed_time(start_time)

    if process.returncode != 0:
        error('Task {bred}{tag}{rst} on {byellow}{address}{rst} returned non-zero exit code: {process.returncode}')
        async with target.lock:
            with open(os.path.join(scandir, '_errors.log'), 'a') as file:
                file.writelines(e('[*] Task {tag} returned non-zero exit code: {process.returncode}. Command: {cmd}\n'))
    else:
        info('Task {bgreen}{tag}{rst} on {byellow}{address}{rst} finished successfully in {elapsed_time}')

    return {'returncode': process.returncode, 'name': 'run_cmd'} 

def run(args, universal_newlines=False, check=False, shell=False, **kwargs):
    """Provide the subprocess.run() function as asyncio corouting.

    This takes care of the missing 'universal_newlines' and 'check' options.
    Everything else is passed through. Will also raise the same exceptions as
    subprocess.run() to act as a drop-in replacement.
    """
    import asyncio
    import io
    import locale
    import subprocess

    if shell:
        proc = await asyncio.create_subprocess_shell(args, **kwargs)
    else:
        proc = await asyncio.create_subprocess_exec(*args, **kwargs)
    stdout, stderr = await proc.communicate()

    if universal_newlines and (stdout is not None):
        stdout = io.TextIOWrapper(io.BytesIO(stdout)).read()
    if universal_newlines and (stderr is not None):
        stderr = io.TextIOWrapper(io.BytesIO(stderr)).read()

    if check and (proc.returncode != 0):
        raise subprocess.CalledProcessError(proc.returncode, args,
            stdout, stderr)

    return subprocess.CompletedProcess(args, proc.returncode, stdout,
        stderr) 

def output_of(*cmd: Optional[str], **kwargs) -> str:
    """Invokes a subprocess and returns its output as a string.

    Args:
        cmd: Components of the command to execute, e.g. ["echo", "dog"].
        **kwargs: Extra arguments for asyncio.create_subprocess_shell, such as
            a cwd (current working directory) argument.

    Returns:
        A (captured output, captured error output, return code) triplet. The
        captured outputs will be None if the out or err parameters were not set
        to an instance of TeeCapture.

    Raises:
         subprocess.CalledProcessError: The process returned a non-zero error
            code and raise_on_fail was set.
    """
    result = cast(
        str,
        run_cmd(*cmd, log_run_to_stderr=False, out=TeeCapture(), **kwargs).out)

    # Strip final newline.
    if result.endswith('\n'):
        result = result[:-1]

    return result 

def test_090_qrexec_service_socket_dom0(self):
        """Basic test socket services (dom0) - data receive"""
        self.loop.run_until_complete(self.testvm1.start())

        self.service_proc = self.loop.run_until_complete(
            asyncio.create_subprocess_shell(
                'socat -u UNIX-LISTEN:/etc/qubes-rpc/test.Socket,mode=666 -',
                stdout=subprocess.PIPE, stdin=subprocess.PIPE))

        try:
            with self.qrexec_policy('test.Socket', self.testvm1, '@adminvm'):
                (stdout, stderr) = self.loop.run_until_complete(asyncio.wait_for(
                    self.testvm1.run_for_stdio(
                        'qrexec-client-vm @adminvm test.Socket', input=TEST_DATA),
                    timeout=10))
        except subprocess.CalledProcessError as e:
            self.fail('{} exited with non-zero code {}; stderr: {}'.format(
                e.cmd, e.returncode, e.stderr))
        except asyncio.TimeoutError:
            self.fail(
                "service timeout, probably EOF wasn't transferred to the VM process")

        try:
            (service_stdout, service_stderr) = self.loop.run_until_complete(
                asyncio.wait_for(
                    self.service_proc.communicate(),
                    timeout=10))
        except asyncio.TimeoutError:
            self.fail(
                "socat timeout, probably EOF wasn't transferred to the VM process")

        service_descriptor = b'test.Socket+ test-inst-vm1 keyword adminvm\0'
        self.assertEqual(service_stdout, service_descriptor + TEST_DATA,
            'Received data differs from what was sent')
        self.assertFalse(stderr,
            'Some data was printed to stderr')
        self.assertFalse(service_stderr,
            'Some data was printed to stderr') 

def test_091_qrexec_service_socket_dom0_send(self):
        """Basic test socket services (dom0) - data send"""
        self.loop.run_until_complete(self.testvm1.start())

        self.create_local_file('/tmp/service-input', TEST_DATA.decode())

        self.service_proc = self.loop.run_until_complete(
            asyncio.create_subprocess_shell(
                'socat -u OPEN:/tmp/service-input UNIX-LISTEN:/etc/qubes-rpc/test.Socket,mode=666'))

        try:
            with self.qrexec_policy('test.Socket', self.testvm1, '@adminvm'):
                stdout, stderr = self.loop.run_until_complete(asyncio.wait_for(
                    self.testvm1.run_for_stdio(
                        'qrexec-client-vm @adminvm test.Socket'),
                    timeout=10))
        except subprocess.CalledProcessError as e:
            self.fail('{} exited with non-zero code {}; stderr: {}'.format(
                e.cmd, e.returncode, e.stderr))
        except asyncio.TimeoutError:
            self.fail(
                "service timeout, probably EOF wasn't transferred to the VM process")

        try:
            (service_stdout, service_stderr) = self.loop.run_until_complete(
                asyncio.wait_for(
                    self.service_proc.communicate(),
                    timeout=10))
        except asyncio.TimeoutError:
            self.fail(
                "socat timeout, probably EOF wasn't transferred to the VM process")

        self.assertEqual(stdout, TEST_DATA,
            'Received data differs from what was sent')
        self.assertFalse(stderr,
            'Some data was printed to stderr')
        self.assertFalse(service_stderr,
            'Some data was printed to stderr') 

def test_092_qrexec_service_socket_dom0_eof_reverse(self):
        """Test for EOF transmission dom0(socket)->VM"""

        self.loop.run_until_complete(self.testvm1.start())

        self.create_local_file(
            '/tmp/service_script',
            '#!/usr/bin/python3\n'
            'import socket, os, sys, time\n'
            's = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)\n'
            'os.umask(0)\n'
            's.bind("/etc/qubes-rpc/test.Socket")\n'
            's.listen(1)\n'
            'conn, addr = s.accept()\n'
            'conn.send(b"test\\n")\n'
            'conn.shutdown(socket.SHUT_WR)\n'
            # wait longer than the timeout below
            'time.sleep(15)\n'
        )

        self.service_proc = self.loop.run_until_complete(
            asyncio.create_subprocess_shell('python3 /tmp/service_script',
                stdout=subprocess.PIPE, stdin=subprocess.PIPE))

        try:
            with self.qrexec_policy('test.Socket', self.testvm1, '@adminvm'):
                p = self.loop.run_until_complete(self.testvm1.run(
                        'qrexec-client-vm @adminvm test.Socket',
                        stdout=subprocess.PIPE, stdin=subprocess.PIPE))

                stdout = self.loop.run_until_complete(asyncio.wait_for(
                    p.stdout.read(),
                    timeout=10))
        except asyncio.TimeoutError:
            self.fail(
                "service timeout, probably EOF wasn't transferred from the VM process")

        self.assertEqual(stdout, b'test\n',
            'Received data differs from what was expected') 

def test_shell(self):
        create = asyncio.create_subprocess_shell('exit 7',
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 7) 

def test_start_new_session(self):
        # start the new process in a new session
        create = asyncio.create_subprocess_shell('exit 8',
                                                 start_new_session=True,
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 8) 

def update_handler(event):
    message = await event.respond("Updating...")
    process = await create_subprocess_shell("git pull origin master", stdin=PIPE, stdout=PIPE)
    output = await process.stdout.read()
    output = output.decode().strip()
    await process.wait()
    if output:
        await message.edit(output)
    await sleep(2)
    await message.edit("Restarting...")
    await restart(message) 

def execution(_, message: Message):
    cmd = message.text.split(" ", maxsplit=1)[1]

    reply_to_id = message.message_id
    if message.reply_to_message:
        reply_to_id = message.reply_to_message.message_id

    process = await asyncio.create_subprocess_shell(
        cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await process.communicate()
    e = stderr.decode()
    if not e:
        e = "No errors"
    o = stdout.decode()
    if not o:
        o = "No output"

    OUTPUT = ""
    OUTPUT += f"<b>Command:</b>\n<code>{cmd}</code>\n\n"
    OUTPUT += f"<b>Output</b>: \n<code>{o}</code>\n"
    OUTPUT += f"<b>Errors</b>: \n<code>{e}</code>"

    if len(OUTPUT) > 4096:
        with open("exec.text", "w+", encoding="utf8") as out_file:
            out_file.write(str(OUTPUT))
        await message.reply_document(
            document="exec.text",
            caption=cmd,
            disable_notification=True,
            reply_to_message_id=reply_to_id
        )
        os.remove("exec.text")
    else:
        await message.reply_text(OUTPUT) 

def run_script():
    process = await asyncio.create_subprocess_shell(
        'python3',
        stdout=asyncio.subprocess.PIPE,
        stdin=asyncio.subprocess.PIPE,
    )

    # Write a simple Python script to the interpreter
    process.stdin.write(b'\n'.join((
        b'import math',
        b'x = 2 ** 8',
        b'y = math.sqrt(x)',
        b'z = math.sqrt(y)',
        b'print("x: %d" % x)',
        b'print("y: %d" % y)',
        b'print("z: %d" % z)',
        b'for i in range(int(z)):',
        b'    print("i: %d" % i)',
    )))
    # Make sure the stdin is flushed asynchronously
    await process.stdin.drain()
    # And send the end of file so the Python interpreter will
    # start processing the input. Without this the process will
    # stall forever.
    process.stdin.write_eof()

    # Fetch the lines from the stdout asynchronously
    async for out in process.stdout:
        # Decode the output from bytes and strip the whitespace
        # (newline) at the right
        print(out.decode('utf-8').rstrip())

    # Wait for the process to exit
    await process.wait() 

def spawn_process(host, port, counter):
    global stopped
    global active_processes
    path_to_vdf_client = find_vdf_client()
    while not stopped:
        try:
            dirname = path_to_vdf_client.parent
            basename = path_to_vdf_client.name
            proc = await asyncio.create_subprocess_shell(
                f"{basename} {host} {port} {counter}",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env={"PATH": dirname},
            )
        except Exception as e:
            log.warning(f"Exception while spawning process {counter}: {(e)}")
            continue
        async with lock:
            active_processes.append(proc)
        stdout, stderr = await proc.communicate()
        if stdout:
            log.info(f"Stdout:\n{stdout.decode().rstrip()}")
        if stderr:
            log.info(f"Stderr:\n{stderr.decode().rstrip()}")
        log.info(f"Process number {counter} ended.")
        async with lock:
            if proc in active_processes:
                active_processes.remove(proc)
        await asyncio.sleep(0.1) 

def test_shell(self):
        create = asyncio.create_subprocess_shell('exit 7',
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 7) 

def test_start_new_session(self):
        # start the new process in a new session
        create = asyncio.create_subprocess_shell('exit 8',
                                                 start_new_session=True,
                                                 loop=self.loop)
        proc = self.loop.run_until_complete(create)
        exitcode = self.loop.run_until_complete(proc.wait())
        self.assertEqual(exitcode, 8) 

def is_ffmpeg_there():
    cmd = await asyncio.create_subprocess_shell(
        'ffmpeg -version',
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    await cmd.communicate()
    return True if cmd.returncode == 0 else False