Python tips

argparse

https://docs.python.org/release/3.8.2/library/argparse.html

インスタンス化

モジュールdocstring書いといてそれをdescriptionにすると良さげ

parser = argparse.ArgumentParser(description=__doc__)

parser = argparse.ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)

具体例1

def parse_args():
    parser = argparse.ArgumentParser(
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        description=__doc__
    )
    parser.add_argument('--Z_file', type=str, required=True, help='Pickled file of latent variables.')
    parser.add_argument('--k_init', type=int, default=2, help='Initial number of clusters in X-Means')
    parser.add_argument('--k_max', type=int, default=20, help='Max number of clusters in X-Means')
    parser.add_argument('--criterion', type=str, default='bic', choices=['bic', 'mndl'], help='Splitting criterion. bic: bayesian information criterion, mndl: minimum noiseless description length.')
    parser.add_argument('--random_state', type=int, default=None, help='Random seed for X-Means initialization (K-Means++)')
    parser.add_argument('--no_ccore', action='store_true', help='Use python implementation of PyClustering library, instead of C++.')
    parser.add_argument('--outdir', required=True, help='Output directory.')
    parser.add_argument('--input_star', type=str, required=True, help='Input star file.')
    parser.add_argument('--output_star_rootname', type=str, default='cluster', help='Output file rootname for each cluster\'s star file.')
    parser.add_argument('--output_model', type=str, default='xmeans_model.pkl', help='Output X-Means model file name (pickled).')

    args = parser.parse_args()

    logger.info('##### Command #####\n\t' + ' '.join(sys.argv))
    args_print_str = '##### Input parameters #####\n'
    for opt, val in vars(args).items():
        args_print_str += '\t{} : {}\n'.format(opt, val)
    return args

logging

https://docs.python.org/release/3.8.2/library/logging.html

ロガー生成スニペット

logging_fmt = logging.Formatter(
    '\n[%(asctime)s - %(name)s - %(levelname)s] %(message)s')
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(logging_fmt)
logger.addHandler(stream_handler)


def set_logger(log_file, is_debug):
    if log_file is not None:
        log_file_dir = os.path.dirname(log_file)
        if log_file_dir:
            if not os.path.exists(log_file_dir):
                logger.debug('Creating log file directory...')
                os.makedirs(log_file_dir)
            elif not os.path.isdir(log_file_dir):
                logger.error('{} is not a directory.'.format(log_file_dir))
                sys.exit(1)
        file_handler = logging.FileHandler(log_file, mode='w')
        file_handler.setFormatter(logging_fmt)
        logger.addHandler(file_handler)

    if not is_debug:
        logger.setLevel(logging.INFO)

serialize

pickle

保存

pickle.dump(\
    object_to_be_serialized,
    open(outfile, 'wb'),
    protocol=4)

読み込み

with open(pickled_file, 'rb') as f:
    pickled_content = pickle.load(f)

subprocess

run - 同期型

ret = subprocess.run("シェルコマンド", shell=True, capture_output=True)

投げたシェルコマンドが終わるまで呼び出し側コードはブロックされる。

retが持つ属性は ret.args, ret.returncode, ret.stdout, ret.stderr。

Popen - 非同期型

参考
- https://qiita.com/HidKamiya/items/e192a55371a2961ca8a4
- https://qiita.com/megmogmog1965/items/5f95b35539ed6b3cfa17

p = subprocess.Popen("シェルコマンド", shell=True, stderr=subprocess.STDOUT, stdout=subprocess.PIPE)

Popenはコマンド投げっぱなしで、呼び出し側コードは遠慮なく先へ進む。

処理の終了 (p.poll() is not None)を待ちつつ、出力をリアルタイム表示するには、例えば以下の様にする。

while True:
    if p.poll() is not None:
        break
    line = p.stdout.readline()
    if line is not None:
        print(line, end='')

build-in

print

改行無しで表示

print('hogefuga', end='')