Source code for mit_d3m.utils

# -*- coding: utf-8 -*-

import logging
import os
import subprocess

import psutil


[docs]def make_abs(path, base_dir=None): base_dir = base_dir or os.getcwd() if not os.path.isabs(path): return os.path.join(base_dir, path) return path
[docs]def contains_files(d): """Check if a directory contains any normal files""" for _, _, files in os.walk(d): if files: return True return False
[docs]def used_memory(): return psutil.Process(os.getpid()).memory_info().rss
[docs]def available_memory(): return psutil.virtual_memory().available
[docs]def disk_usage(path, human=False): """disk usage in bytes or human readable format (e.g. '2,1GB')""" command = ['du', '-s', path] if human: command.insert(-1, '-h') return subprocess.check_output(command).split()[0].decode('utf-8')
[docs]def walk(document, transform): if not isinstance(document, dict): return document new_doc = dict() for key, value in document.items(): if isinstance(value, dict): value = walk(value, transform) elif isinstance(value, list): value = [walk(v, transform) for v in value] new_key, new_value = transform(key, value) new_doc[new_key] = new_value return new_doc
[docs]def remove_dots(document): return walk(document, lambda key, value: (key.replace('.', '-'), value))
[docs]def restore_dots(document): return walk(document, lambda key, value: (key.replace('-', '.'), value))
[docs]def logging_setup(verbosity=1, logfile=None, logger_name=None): logger = logging.getLogger(logger_name) log_level = (3 - verbosity) * 10 fmt = '%(asctime)s - %(process)d - %(levelname)s - %(module)s - %(message)s' formatter = logging.Formatter(fmt) logger.setLevel(log_level) logger.propagate = False if logfile: file_handler = logging.FileHandler(logfile) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(formatter) logger.addHandler(file_handler) else: console_handler = logging.StreamHandler() console_handler.setLevel(log_level) console_handler.setFormatter(formatter) logger.addHandler(console_handler)