jw-python/tools/python/jwutils/misc.py

import os, errno
import atexit
import tempfile
import inspect
from jwutils import log
from typing import Set

_tmpfiles: Set[str] = set()

def _cleanup():
    for f in _tmpfiles:
        silentremove(f)

def silentremove(filename): #export
    try:
        os.remove(filename)
    except OSError as e:
        if e.errno != errno.ENOENT:
            raise # re-raise exception if a different error occurred

def pad(token, total_size, right_align = False):
    add = total_size - len(token)
    if add <= 0:
        return token
    space = ' ' * add
    if right_align:
        return space + token
    return token + space

def atomic_store(contents, path): # export
    if path[0:3] == '/dev':
        with open(path, 'w') as outfile:
            outfile.write(contents)
        return
    outfile = tempfile.NamedTemporaryFile(prefix=os.path.basename(path), delete=False, dir=os.path.dirname(path))
    name = outfile.name
    _tmpfiles.add(name)
    outfile.write(contents)
    outfile.close()
    os.rename(name, path)
    _tmpfiles.remove(name)

# see https://stackoverflow.com/questions/2020014
def object_builtin_name(o, full=True): # export
    #if not full:
    #    return o.__class__.__name__
    module = o.__class__.__module__
    if module is None or module == str.__class__.__module__:
        return o.__class__.__name__  # Avoid reporting __builtin__
    return module + '.' + o.__class__.__name__

def get_derived_classes(mod, base): # export
    members = inspect.getmembers(mod, inspect.isclass)
    r = []
    for name, c in members:
        log.slog(log.DEBUG, "found ", name)
        if inspect.isabstract(c):
            log.slog(log.DEBUG, "  is abstract")
            continue
        if not base in inspect.getmro(c):
            log.slog(log.DEBUG, "  is not derived from", base, "only", inspect.getmro(c))
            continue
        r.append(c)
    return r

def commit_tmpfile(tmp, path): # export
    caller = log.get_caller_pos()
    if os.path.isfile(path) and filecmp.cmp(tmp, path):
        log.slog(log.INFO, "{} is up to date".format(path), caller=caller)
        os.unlink(tmp)
    else:
        log.slog(log.NOTICE, "saving {}".format(path), caller=caller)
        os.rename(path + '.tmp', path)

atexit.register(_cleanup)