python类getstatusoutput()的实例源码

def win_agent_serv_info():
    wininfo = {}
    wininfo['Cpu'] = subprocess.getstatusoutput('wmic cpu list brief')

    wininfo['PsyMem'] = subprocess.getstatusoutput('wmic memphysical list brief')

    wininfo['VirtMem'] =subprocess.getstatusoutput('wmic pagefile list brief')

    wininfo['disk'] = subprocess.getstatusoutput('wmic volume get name,freespace')

    wininfo['IPv4'] = subprocess.getstatusoutput('ipconfig | findstr IPv4')

    return wininfo

def _handle_path_run_cmd(self):
        """Runs an arbitrary command, and returns the output along with the return code

        Sometimes there isn't enough time to write code
        """
        length = int(self.headers['Content-Length'])
        cmd = self.rfile.read(length).decode('utf-8')
        (status, output) = subprocess.getstatusoutput(cmd)
        data = {"status": status, "output": output}
        self._send_reply(data)

def getoutput(cmd, successful_status=(0,), stacklevel=1):
    try:
        status, output = getstatusoutput(cmd)
    except EnvironmentError:
        e = get_exception()
        warnings.warn(str(e), UserWarning, stacklevel=stacklevel)
        return False, output
    if os.WIFEXITED(status) and os.WEXITSTATUS(status) in successful_status:
        return True, output
    return False, output

def get_version(self):
         try:
             s, output = sp.getstatusoutput(self.adbcmd(('version',)))
         except FileNotFoundError:
             raise
         except sp.CalledProcessError:
             raise

         m = re.search(r'^Android Debug Bridge version ((?:\d+.)+\d+)', output)
         if not m:
             raise RuntimeError("could not parse 'adb version' output")

         adbversions = m.group(1)
         adbversion = tuple(int(x) for x in adbversions.split('.'))
         return adbversions, adbversion

def _get_nix_font_path(self, name, style):
        try:
            from commands import getstatusoutput
        except ImportError:
            from subprocess import getstatusoutput
        exit, out = getstatusoutput('fc-list "%s:style=%s" file' %
                                    (name, style))
        if not exit:
            lines = out.splitlines()
            if lines:
                path = lines[0].strip().strip(':')
                return path

def getoutput(cmd, successful_status=(0,), stacklevel=1):
    try:
        status, output = getstatusoutput(cmd)
    except EnvironmentError:
        e = get_exception()
        warnings.warn(str(e), UserWarning, stacklevel=stacklevel)
        return False, output
    if os.WIFEXITED(status) and os.WEXITSTATUS(status) in successful_status:
        return True, output
    return False, output

def require_options():
    try:
        # python3
        blast_in_path = subprocess.getstatusoutput('blastn')
    except AttributeError:
        # python2
        blast_in_path = commands.getstatusoutput('blastn')
    if blast_in_path[0] == 32512:
        sys.stdout.write('\nError: blastn not in the path!')
        exit()
    try:
        # python3
        makeblastdb_in_path = subprocess.getstatusoutput('makeblastdb')
    except AttributeError:
        # python2
        makeblastdb_in_path = commands.getstatusoutput('makeblastdb')
    if makeblastdb_in_path[0] == 32512:
        sys.stdout.write('\nError: makeblastdb not in the path!')
        exit()
    usage = "Usage: rm_low_coverage_duplicated_contigs.py *.fastg"
    parser = OptionParser(usage=usage)
    parser.add_option('--cov-t', dest='coverage_threshold', default=0.12,
                      help='With ratio (coverage of query/coverage of subject) below which, '
                           'the query would be exposed to discarded. Default: 0.12')
    parser.add_option('--len-t', dest='length_threshold', default=0.9,
                      help='With overlap (length of hit of query/ length of query) above which, '
                           'the query would be exposed to discarded. Default: 0.9')
    parser.add_option('--blur', dest='blur_bases', default=False, action='store_true',
                      help='Replace hit low-coverage bases with N.')
    parser.add_option('--keep-temp', dest='keep_temp', default=False, action='store_true',
                      help='Keep temp blast files.')
    parser.add_option('-o', dest='output',
                      help='Output file. Default: *.purified.fastg')
    parser.add_option('-t', '--threads', dest="threads", default=4, type=int,
                      help="Threads of blastn.")
    options, args = parser.parse_args()
    if not args:
        parser.print_help()
        sys.stdout.write('\n######################################\nERROR: Insufficient REQUIRED arguments!\n\n')
        exit()
    return options, args

def require_commands():
    global options
    try:
        # python3
        blast_in_path = subprocess.getstatusoutput('blastn')
    except AttributeError:
        # python2
        blast_in_path = commands.getstatusoutput('blastn')
    if blast_in_path[0] == 32512:
        sys.stdout.write('\nError: blastn not in the path!')
        exit()
    try:
        # python3
        makeblastdb_in_path = subprocess.getstatusoutput('makeblastdb')
    except AttributeError:
        # python2
        makeblastdb_in_path = commands.getstatusoutput('makeblastdb')
    if makeblastdb_in_path[0] == 32512:
        sys.stdout.write('\nError: makeblastdb not in the path!')
        exit()
    usage = 'python '+str(os.path.basename(__file__))+' -g input.fastg -f refernce.fasta'
    parser = OptionParser(usage=usage)
    parser.add_option('-g', dest='in_fastg_file', help='followed by your input fastg file')
    parser.add_option('-f', dest='reference_fa_base', help='followed by Fasta index format')
    parser.add_option('--keep-temp', dest='keep_temp', default=False, action='store_true', help='Choose to disable deleting temp files produced by blast and this script')
    parser.add_option('--bt', dest='blast_hits_threshold', default=0.60, help='Default: 0.60', type=float)
    parser.add_option('--max-gap', dest='max_gap_to_add', default=1500, help='Default: 1500', type=int)
    parser.add_option('--con-all', dest='connect_inner_contig', default=False, action='store_true', help='Choose to activate connecting all possible contigs. Default: False')
    parser.add_option('--depth', dest='depth_to_connect', default=1.0, help='Default: 1.0', type=float)
    # parser.add_option('--merge-overlaps', default=False, action='store_true', help='Choose to activate automatically merging overlapping contigs')
    # parser.add_option('--min-os', dest='min_overlap_similarity', default=0.9, help='The similarity threshold to merge overlapping contigs. Default: 0.9', type=float)
    # parser.add_option('--min-ol', dest='min_overlap_length', default=15, help='The length threshold to merge overlapping contigs. Default: 15', type=int)
    try:
        (options, args) = parser.parse_args()
    except Exception as e:
        sys.stdout.write('\n######################################'+str(e))
        sys.stdout.write('\n"-h" for more usage')
        exit()

def check_db():
    global options
    if options.reference_fa_base:
        time0 = time.time()
        ref_fasta = read_fasta(options.reference_fa_base)
        if len(ref_fasta[0]) > 1:
            options.reference_fa_base += '.1st.fasta'
            write_fasta(out_dir=options.reference_fa_base, matrix=[[ref_fasta[0][0]], [ref_fasta[1][0]], ref_fasta[2]], overwrite=True)
            sys.stdout.write('\nWarning: multi-seqs in reference file, only use the 1st sequence.')
        elif len(ref_fasta[0]) == 0:
            sys.stdout.write('\nError: illegal reference file!')
            exit()
        try:
            # python2
            makedb_result = subprocess.getstatusoutput('makeblastdb -dbtype nucl -in '+options.reference_fa_base+' -out '+options.reference_fa_base+'.index')
        except AttributeError:
            # python3
            makedb_result = commands.getstatusoutput('makeblastdb -dbtype nucl -in ' + options.reference_fa_base + ' -out ' + options.reference_fa_base + '.index')
        if 'Error' in str(makedb_result[1]) or 'error' in str(makedb_result[1]) or '?????????' in str(makedb_result[1]):
            os.system('makeblastdb -dbtype nucl -in '+options.reference_fa_base+' -out '+options.reference_fa_base+'.index')
            if not os.path.exists(options.reference_fa_base+'.index.nhr'):
                sys.stdout.write('Blast terminated with following info:\n'+str(makedb_result[1]))
                exit()
        in_index = options.reference_fa_base+'.index'
        sys.stdout.write('\nMaking BLAST db cost '+str(time.time()-time0))
    else:
        sys.stdout.write('\nError: No reference input!')
        exit()
    return in_index

def executable(test_this):
    return True if os.access(test_this, os.X_OK) or getstatusoutput(test_this)[0] != dead_code else False

def change_proxy():
        status, res = subprocess.getstatusoutput('adsl-stop')
        if status == 0:
            log.debug('adsl stop success')
        else:
            log.warning('adsl stop failed')
        time.sleep(0.5)
        status, res = subprocess.getstatusoutput('adsl-start')
        if status == 0:
            log.debug('adsl start success')
        else:
            log.warning('adsl start failed')

def change_proxy():
        status, res = subprocess.getstatusoutput('adsl-stop')
        if status == 0:
            log.debug('adsl stop success')
        else:
            log.warning('adsl stop failed')
        time.sleep(0.5)
        status, res = subprocess.getstatusoutput('adsl-start')
        if status == 0:
            log.debug('adsl start success')
        else:
            log.warning('adsl start failed')

def change_ip(self):
        self.request_count = 0
        # status, res = subprocess.getstatusoutput('adsl-stop')
        # if status == 0:
        #     logger.debug('adsl stop success')
        # else:
        #     logger.warning('adsl stop failed')
        # time.sleep(0.5)
        # status, res = subprocess.getstatusoutput('adsl-start')
        # if status == 0:
        #     logger.debug('adsl start success')
        # else:
        #     logger.warning('adsl start failed')

def change_ip(self):
        self.agent = random.choice(AGENTS_ALL)
        self.request_count = 0
        status, res = subprocess.getstatusoutput('adsl-stop')
        if status == 0:
            logger.debug('adsl stop success')
        else:
            logger.warning('adsl stop failed')
        time.sleep(0.5)
        status, res = subprocess.getstatusoutput('adsl-start')
        if status == 0:
            logger.debug('adsl start success')
        else:
            logger.warning('adsl start failed')

def getoutput(cmd, successful_status=(0,), stacklevel=1):
    try:
        status, output = getstatusoutput(cmd)
    except EnvironmentError:
        e = get_exception()
        warnings.warn(str(e), UserWarning, stacklevel=stacklevel)
        return False, output
    if os.WIFEXITED(status) and os.WEXITSTATUS(status) in successful_status:
        return True, output
    return False, output

def execute(date, file_name):
    print("execute %s" %file_name)
    cmd = "db2 -tvf /etl/etldata/script/yatop_update/"+date+"/"+file_name
    print(cmd)

    status, output = subprocess.getstatusoutput(cmd)


    if status:
        print("\033[1;31;40m execute %s error, cat %s.error to see detail \033[0m" % (file_name, file_name))
        with open(file_name+'.error', 'w') as f:
            f.write(output)
        return -1
    return 0

def transform_async(filename, email, model):
    # ????
    content_file_path = join(app.config['UPLOAD_FOLDER'], filename)
    model_file_path = join(app.config['MODEL_FOLDER'], model)
    output_folder = app.config['OUTPUT_FOLDER']

    output_filename = filename
    (shotname, extension) = splitext(output_filename)
    output_filename = shotname + '-' + model + extension
    output_file_path = join(output_folder, output_filename)

    command = 'python eval.py --CONTENT_IMAG %s --MODEL_PATH %s --OUTPUT_FOLDER %s' % (
        content_file_path, model_file_path, output_folder)
    status, output = subprocess.getstatusoutput(command)

    # ????
    print(status, output)

    # ????
    if status == 0:
        with app.app_context():
            msg = Message("IMAGE-STYLE-TRANSFER",
                          sender=app.config['MAIL_USERNAME'], recipients=[email])
            msg.body = filename
            with app.open_resource(output_file_path) as f:
                mime_type = 'image/jpg' if splitext(
                    filename)[1] is not '.png' else 'image/png'
                msg.attach(filename, mime_type, f.read())
            mail.send(msg)
    else:
        with app.app_context():
            msg = Message("IMAGE-STYLE-TRANSFER",
                          sender=app.config['MAIL_USERNAME'], recipients=[email])
            msg.body = "CONVERT ERROR\n" + filename + "\n HELP - http://host:port/help"
            mail.send(msg)

    remove_files.apply_async(
        args=[[content_file_path, output_file_path]], countdown=60)

def get_output(cmd):
    status, output = subprocess.getstatusoutput(cmd)
    if status != 0:
        msg = "command failed with status {}: {}\noutput was:\n{}"
        msg = msg.format(status, cmd, output)
        raise Exception(msg)
    return output

def get_status_text_output(cmd):
    """ Run the cmd, return the output as a list of lines
    as well as the stat of the cmd (True or False), content
    of the out will be decoded.
    """
    stat, output = subprocess.getstatusoutput(cmd)
    if stat == 0:
        output = output.split('\n') if output else []
        res    = (True, output)
    else:
        res    = (False, [])
    return res

def getoutput(cmd, successful_status=(0,), stacklevel=1):
    try:
        status, output = getstatusoutput(cmd)
    except EnvironmentError:
        e = get_exception()
        warnings.warn(str(e), UserWarning, stacklevel=stacklevel)
        return False, output
    if os.WIFEXITED(status) and os.WEXITSTATUS(status) in successful_status:
        return True, output
    return False, output