# -*- coding: utf-8 -*- """ Created on Tue Aug 8 14:26:00 2017 @author: ofn77899 """ import regularizers import numpy as np from enum import Enum import timeit class Regularizer(): '''Class to handle regularizer algorithms to be used during reconstruction Currently 5 CPU (OMP) regularization algorithms are available: 1) SplitBregman_TV 2) FGP_TV 3) LLT_model 4) PatchBased_Regul 5) TGV_PD Usage: the regularizer can be invoked as object or as static method Depending on the actual regularizer the input parameter may vary, and a different default setting is defined. reg = Regularizer(Regularizer.Algorithm.SplitBregman_TV) out = reg(input=u0, regularization_parameter=10., number_of_iterations=30, tolerance_constant=1e-4, TV_Penalty=Regularizer.TotalVariationPenalty.l1) out2 = Regularizer.SplitBregman_TV(input=u0, regularization_parameter=10., number_of_iterations=30, tolerance_constant=1e-4, TV_Penalty=Regularizer.TotalVariationPenalty.l1) A number of optional parameters can be passed or skipped out2 = Regularizer.SplitBregman_TV(input=u0, regularization_parameter=10. ) ''' class Algorithm(Enum): SplitBregman_TV = regularizers.SplitBregman_TV FGP_TV = regularizers.FGP_TV LLT_model = regularizers.LLT_model PatchBased_Regul = regularizers.PatchBased_Regul TGV_PD = regularizers.TGV_PD # Algorithm class TotalVariationPenalty(Enum): isotropic = 0 l1 = 1 # TotalVariationPenalty def __init__(self , algorithm, debug = True): self.setAlgorithm ( algorithm ) self.debug = debug # __init__ def setAlgorithm(self, algorithm): self.algorithm = algorithm self.pars = self.getDefaultParsForAlgorithm(algorithm) # setAlgorithm def getDefaultParsForAlgorithm(self, algorithm): pars = dict() if algorithm == Regularizer.Algorithm.SplitBregman_TV : pars['algorithm'] = algorithm pars['input'] = None pars['regularization_parameter'] = None pars['number_of_iterations'] = 35 pars['tolerance_constant'] = 0.0001 pars['TV_penalty'] = Regularizer.TotalVariationPenalty.isotropic elif algorithm == Regularizer.Algorithm.FGP_TV : pars['algorithm'] = algorithm pars['input'] = None pars['regularization_parameter'] = None pars['number_of_iterations'] = 50 pars['tolerance_constant'] = 0.001 pars['TV_penalty'] = Regularizer.TotalVariationPenalty.isotropic elif algorithm == Regularizer.Algorithm.LLT_model: pars['algorithm'] = algorithm pars['input'] = None pars['regularization_parameter'] = None pars['time_step'] = None pars['number_of_iterations'] = None pars['tolerance_constant'] = None pars['restrictive_Z_smoothing'] = 0 elif algorithm == Regularizer.Algorithm.PatchBased_Regul: pars['algorithm'] = algorithm pars['input'] = None pars['searching_window_ratio'] = None pars['similarity_window_ratio'] = None pars['PB_filtering_parameter'] = None pars['regularization_parameter'] = None elif algorithm == Regularizer.Algorithm.TGV_PD: pars['algorithm'] = algorithm pars['input'] = None pars['first_order_term'] = None pars['second_order_term'] = None pars['number_of_iterations'] = None pars['regularization_parameter'] = None else: raise Exception('Unknown regularizer algorithm') return pars # parsForAlgorithm def setParameter(self, **kwargs): '''set named parameter for the regularization engine raises Exception if the named parameter is not recognized Typical usage is: reg = Regularizer(Regularizer.Algorithm.SplitBregman_TV) reg.setParameter(input=u0) reg.setParameter(regularization_parameter=10.) it can be also used as reg = Regularizer(Regularizer.Algorithm.SplitBregman_TV) reg.setParameter(input=u0 , regularization_parameter=10.) ''' for key , value in kwargs.items(): if key in self.pars.keys(): self.pars[key] = value else: raise Exception('Wrong parameter {0} for regularizer algorithm'.format(key)) # setParameter def getParameter(self, **kwargs): ret = {} for key , value in kwargs.items(): if key in self.pars.keys(): ret[key] = self.pars[key] else: raise Exception('Wrong parameter {0} for regularizer algorithm'.format(key)) # setParameter def __call__(self, input = None, regularization_parameter = None, **kwargs): '''Actual call for the regularizer. One can either set the regularization parameters first and then call the algorithm or set the regularization parameter during the call (as is done in the static methods). ''' if kwargs is not None: for key, value in kwargs.items(): #print("{0} = {1}".format(key, value)) self.pars[key] = value if input is not None: self.pars['input'] = input if regularization_parameter is not None: self.pars['regularization_parameter'] = regularization_parameter if self.debug: print ("--------------------------------------------------") for key, value in self.pars.items(): if key== 'algorithm' : print("{0} = {1}".format(key, value.__name__)) elif key == 'input': print("{0} = {1}".format(key, np.shape(value))) else: print("{0} = {1}".format(key, value)) if None in self.pars: raise Exception("Not all parameters have been provided") input = self.pars['input'] regularization_parameter = self.pars['regularization_parameter'] if self.algorithm == Regularizer.Algorithm.SplitBregman_TV : return self.algorithm(input, regularization_parameter, self.pars['number_of_iterations'], self.pars['tolerance_constant'], self.pars['TV_penalty'].value ) elif self.algorithm == Regularizer.Algorithm.FGP_TV : return self.algorithm(input, regularization_parameter, self.pars['number_of_iterations'], self.pars['tolerance_constant'], self.pars['TV_penalty'].value ) elif self.algorithm == Regularizer.Algorithm.LLT_model : #LLT_model(np::ndarray input, double d_lambda, double d_tau, int iter, double d_epsil, int switcher) # no default return self.algorithm(input, regularization_parameter, self.pars['time_step'] , self.pars['number_of_iterations'], self.pars['tolerance_constant'], self.pars['restrictive_Z_smoothing'] ) elif self.algorithm == Regularizer.Algorithm.PatchBased_Regul : #LLT_model(np::ndarray input, double d_lambda, double d_tau, int iter, double d_epsil, int switcher) # no default return self.algorithm(input, regularization_parameter, self.pars['searching_window_ratio'] , self.pars['similarity_window_ratio'] , self.pars['PB_filtering_parameter']) elif self.algorithm == Regularizer.Algorithm.TGV_PD : #LLT_model(np::ndarray input, double d_lambda, double d_tau, int iter, double d_epsil, int switcher) # no default if len(np.shape(input)) == 2: return self.algorithm(input, regularization_parameter, self.pars['first_order_term'] , self.pars['second_order_term'] , self.pars['number_of_iterations']) elif len(np.shape(input)) == 3: #assuming it's 3D # run independent calls on each slice out3d = input.copy() for i in range(np.shape(input)[2]): out = self.algorithm(input, regularization_parameter, self.pars['first_order_term'] , self.pars['second_order_term'] , self.pars['number_of_iterations']) # copy the result in the 3D image out3d.T[i] = out[0].copy() # append the rest of the info that the algorithm returns output = [out3d] for i in range(1,len(out)): output.append(out[i]) return output # __call__ @staticmethod def SplitBregman_TV(input, regularization_parameter , **kwargs): start_time = timeit.default_timer() reg = Regularizer(Regularizer.Algorithm.SplitBregman_TV) out = list( reg(input, regularization_parameter, **kwargs) ) out.append(reg.pars) txt = reg.printParametersToString() txt += "%s = %.3fs" % ('elapsed time',timeit.default_timer() - start_time) out.append(txt) return out @staticmethod def FGP_TV(input, regularization_parameter , **kwargs): start_time = timeit.default_timer() reg = Regularizer(Regularizer.Algorithm.FGP_TV) out = list( reg(input, regularization_parameter, **kwargs) ) out.append(reg.pars) txt = reg.printParametersToString() txt += "%s = %.3fs" % ('elapsed time',timeit.default_timer() - start_time) out.append(txt) return out @staticmethod def LLT_model(input, regularization_parameter , time_step, number_of_iterations, tolerance_constant, restrictive_Z_smoothing=0): start_time = timeit.default_timer() reg = Regularizer(Regularizer.Algorithm.LLT_model) out = list( reg(input, regularization_parameter, time_step=time_step, number_of_iterations=number_of_iterations, tolerance_constant=tolerance_constant, restrictive_Z_smoothing=restrictive_Z_smoothing) ) out.append(reg.pars) txt = reg.printParametersToString() txt += "%s = %.3fs" % ('elapsed time',timeit.default_timer() - start_time) out.append(txt) return out @staticmethod def PatchBased_Regul(input, regularization_parameter, searching_window_ratio, similarity_window_ratio, PB_filtering_parameter): start_time = timeit.default_timer() reg = Regularizer(Regularizer.Algorithm.PatchBased_Regul) out = list( reg(input, regularization_parameter, searching_window_ratio=searching_window_ratio, similarity_window_ratio=similarity_window_ratio, PB_filtering_parameter=PB_filtering_parameter ) ) out.append(reg.pars) txt = reg.printParametersToString() txt += "%s = %.3fs" % ('elapsed time',timeit.default_timer() - start_time) out.append(txt) return out @staticmethod def TGV_PD(input, regularization_parameter , first_order_term, second_order_term, number_of_iterations): start_time = timeit.default_timer() reg = Regularizer(Regularizer.Algorithm.TGV_PD) out = list( reg(input, regularization_parameter, first_order_term=first_order_term, second_order_term=second_order_term, number_of_iterations=number_of_iterations) ) out.append(reg.pars) txt = reg.printParametersToString() txt += "%s = %.3fs" % ('elapsed time',timeit.default_timer() - start_time) out.append(txt) return out def printParametersToString(self): txt = r'' for key, value in self.pars.items(): if key== 'algorithm' : txt += "{0} = {1}".format(key, value.__name__) elif key == 'input': txt += "{0} = {1}".format(key, np.shape(value)) else: txt += "{0} = {1}".format(key, value) txt += '\n' return txt