Homec4science
Diffusion networkTraining (master) Browse

R8206/utils.py
b77bb05417bbmaster

Last Change
Raw File

utils.py
View Options

import sys
import os
import numpy as np
import json
import yaml
import re
import pickle
import torch
import logging
from collections import namedtuple
import multiprocessing
import itertools
import glob
from . import cropping
__all__ = ["torch_version_major", "torch_version_minor", "torch_no_grad",
"sigmoid", "softmax", "rgb2gray", "sort_nicely", "mkdir", "rmdir",
"json_read", "json_write", "yaml_read", "yaml_write",
"pickle_read", "pickle_write", "isnan", "torch_save",
"downsample_image", "upsample_image", "inbounds",
"Parallel", "find_files", "find_images", "config_logger",
"BinCounter", "invfreq_lossweights", "process_in_chuncks",
"to_torch", "from_torch"]
torch_version_major = int(torch.__version__.split('.')[0])
torch_version_minor = int(torch.__version__.split('.')[1])
class Dummysink(object):
def write(self, data):
pass # ignore the data
def __enter__(self): return self
def __exit__(*x): pass
torch_no_grad = Dummysink() if torch_version_major==0 and torch_version_minor<4 else torch.no_grad()
def to_torch(ndarray, volatile=False):
if torch_version_major>=1:
return torch.from_numpy(ndarray)
else:
from torch.autograd import Variable
return Variable(torch.from_numpy(ndarray), volatile=volatile)
def from_torch(tensor, num=False):
return tensor.data.cpu().numpy()
'''
if num and torch_version_major==0 and torch_version_minor<4:
return tensor.data.cpu().numpy()[0]
else:
return tensor.data.cpu().numpy()
'''
def sigmoid(x):
e = np.exp(x)
return e/(e + 1)
def softmax(x):
e_x = np.exp(x - np.max(x,0))
return e_x / e_x.sum(axis=0)
def rgb2gray(image):
dtype = image.dtype
gray = np.dot(image[...,:3], [0.299, 0.587, 0.114])
return gray.astype(dtype)
def sort_nicely(l):
""" Sort the given list in the way that humans expect.
"""
convert = lambda text: int(text) if text.isdigit() else text
alphanum_key = lambda key: [ convert(c) for c in re.split('([0-9]+)', key) ]
return sorted(l, key=alphanum_key)
def mkdir(directory):
directory = os.path.abspath(directory)
if not os.path.exists(directory):
os.makedirs(directory)
def rmdir(directory):
directory = os.path.abspath(directory)
if os.path.exists(directory):
shutil.rmtree(directory)
def json_read(filename):
try:
with open(os.path.abspath(filename)) as f:
data = json.load(f)
return data
except:
raise ValueError("Unable to read JSON {}".format(filename))
def json_write(data, filename):
try:
directory = os.path.dirname(os.path.abspath(filename))
if not os.path.exists(directory):
os.makedirs(directory)
with open(os.path.abspath(filename), 'w') as f:
json.dump(data, f, indent=2)
except:
raise ValueError("Unable to write JSON {}".format(filename))
def yaml_read(filename):
try:
with open(filename, 'r') as f:
try:
data = yaml.load(f, Loader=yaml.FullLoader)
except AttributeError as e:
data = yaml.load(f)
return data
except:
raise ValueError("Unable to read YAML {}".format(filename))
def yaml_write(data, filename):
try:
directory = os.path.dirname(os.path.abspath(filename))
if not os.path.exists(directory):
os.makedirs(directory)
with open(os.path.abspath(filename), 'w') as f:
yaml.dump(data, f, default_flow_style=False, width=1000)
except:
raise ValueError("Unable to write YAML {}".format(filename))
def pickle_read(filename):
with open(filename, "rb") as f:
data = pickle.load(f)
return data
def pickle_write(filename, data):
directory = os.path.dirname(os.path.abspath(filename))
if not os.path.exists(directory):
os.makedirs(directory)
with open(filename, 'wb') as f:
pickle.dump(data, f)
def isnan(x):
return x!=x
def torch_save(filename, dictionary):
mkdir(os.path.dirname(filename))
torch.save(dictionary, filename)
def downsample_image(img, size, msigma=1.0, interpolation='area'):
import cv2
scale_h = size[0]/img.shape[0]
scale_w = size[1]/img.shape[1]
if interpolation == 'cubic':
interpolation=cv2.INTER_CUBIC
elif interpolation == 'area':
interpolation=cv2.INTER_AREA
elif interpolation == 'linear':
interpolation=cv2.INTER_LINEAR
if msigma is not None:
img = cv2.GaussianBlur(img, ksize=(0,0), sigmaX=1.0/scale_w*msigma, sigmaY=1.0/scale_h*msigma)
img = cv2.resize(img, dsize=(size[1], size[0]), interpolation=interpolation)
return img
def upsample_image(img, size, interpolation='cubic'):
import cv2
if interpolation == 'cubic':
interpolation=cv2.INTER_CUBIC
elif interpolation == 'area':
interpolation=cv2.INTER_AREA
elif interpolation == 'linear':
interpolation=cv2.INTER_LINEAR
img = cv2.resize(img, dsize=(size[1], size[0]), interpolation=interpolation)
return img
def inbounds(coords, image_shape):
"""
Returns a mask of the coordinates that are inside an image with the given
shape.
coords: [x,y]
image_shape: [height, width]
x->width
y->height
"""
mask = np.ones(len(coords))
image_shape = np.array(image_shape)[[1,0]]
for coords_i, sh_i in zip(coords.T, image_shape):
aux = np.logical_and(coords_i >= 0, coords_i < sh_i)
mask = np.logical_and(mask, aux)
return mask
class Parallel(object):
def __init__(self, threads=8):
self.threads = threads
self.p = multiprocessing.Pool(threads)
def __call__(self, f, iterable, *arg):
if len(arg):
res = self.p.starmap(f, itertools.product(iterable, *[[x] for x in arg]))
else:
res = self.p.map(f, iterable)
self.p.close()
self.p.join()
return res
@staticmethod
def split_iterable(iterable, n):
if isinstance(iterable, (list,tuple)):
s = len(iterable)//n
return [iterable[i:i + s] for i in range(0, len(iterable), s)]
elif isinstance(iterable, np.ndarray):
return np.array_split(iterable, n)
@staticmethod
def join_iterable(iterable):
if isinstance(iterable, (list,tuple)):
return list(itertools.chain.from_iterable(iterable))
elif isinstance(iterable, np.ndarray):
return np.concatenate(iterable)
def find_files(file_or_folder, hint=None):
if hint is not None:
file_or_folder = os.path.join(file_or_folder, hint)
filenames = [f for f in glob.glob(file_or_folder)]
filenames = sort_nicely(filenames)
filename_files = []
for filename in filenames:
if os.path.isfile(filename):
filename_files.append(filename)
return filename_files
def find_images(file_or_folder, hint=None):
filenames = find_files(file_or_folder, hint)
for filename in filenames:
_, extension = os.path.splitext(filename)
if extension.lower() in [".jpg",".jpeg",".bmp",".tiff",".png",".gif"]:
filename_images.append(filename)
return filename_images
def config_logger(log_file=None):
"""
Basic configuration of the logging system. Support logging to a file.
Log messages can be submitted from any script.
config_logger(.) is called once from the main script.
Example
-------
import logging
logger = logging.getLogger(__name__)
utils.config_logger("main.log")
logger.info("this is a log.")
"""
class MyFormatter(logging.Formatter):
info_format = "\x1b[32;1m%(asctime)s [%(name)s]\x1b[0m %(message)s"
error_format = "\x1b[31;1m%(asctime)s [%(name)s] [%(levelname)s]\x1b[0m %(message)s"
def format(self, record):
if record.levelno > logging.INFO:
self._style._fmt = self.error_format
else:
self._style._fmt = self.info_format
return super(MyFormatter, self).format(record)
rootLogger = logging.getLogger()
if log_file is not None:
fileHandler = logging.FileHandler(log_file)
fileFormatter = logging.Formatter("%(asctime)s [%(name)s] [%(levelname)s]> %(message)s")
fileHandler.setFormatter(fileFormatter)
rootLogger.addHandler(fileHandler)
consoleHandler = logging.StreamHandler()
consoleFormatter = MyFormatter()
consoleHandler.setFormatter(consoleFormatter)
rootLogger.addHandler(consoleHandler)
rootLogger.setLevel(logging.INFO)
class BinCounter(object):
"""Counter of elements in NumPy arrays."""
def __init__(self, minlength=0, x=None, weights=None):
self.minlength = minlength
self.counts = np.zeros(minlength, dtype=np.int_)
if x is not None and len(x) > 0:
self.update(x, weights)
def update(self, x, weights=None):
if weights is not None:
weights = weights.flatten()
current_counts = np.bincount(np.ravel(x), weights=weights, minlength=self.minlength)
current_counts[:len(self.counts)] += self.counts
self.counts = current_counts
def frequencies(self):
return self.counts / np.float_(np.sum(self.counts))
def invfreq_lossweights(labels, num_classes):
bc = BinCounter(num_classes + 1)
for labels_i in labels:
bc.update(labels_i)
class_weights = 1.0 / (num_classes * bc.frequencies)[:num_classes]
class_weights = np.hstack([class_weights, 0])
class_weights[np.isinf(class_weights)] = np.max(class_weights)
return np.float32(class_weights)
def process_in_chuncks(image, output, process, patch_size, patch_margin):
"""
N,C,D1,D2,...,Dn
"""
assert len(image.shape)==len(output.shape)
assert (len(image.shape)-2)==len(patch_size)
assert len(patch_margin)==len(patch_size)
chunck_coords = cropping.split_with_margin(image.shape[2:], patch_size, patch_margin)
semicol = (slice(None,None),) # this mimicks :
for source_c, valid_c, destin_c in zip(*chunck_coords):
crop = image[semicol+semicol+source_c]
proc_crop = process(crop)
output[semicol+semicol+destin_c] = proc_crop[semicol+semicol+valid_c]
return output
c4science · Help