<s> import numpy as np <EOL> from os . path import join as pjoin <EOL> from . import utils <EOL> from . interfaces import Task <EOL> class TrainingExit ( Exception ) : <EOL> def __init__ ( self , status ) : <EOL> self . status = status <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" + repr ( self . status ) <EOL> class MaxEpochStopping ( Task ) : <EOL> def __init__ ( self , nb_epochs_max ) : <EOL> super ( ) . __init__ ( ) <EOL> self . nb_epochs_max = nb_epochs_max <EOL> def post_epoch ( self , status ) : <EOL> if status . current_epoch >= self . nb_epochs_max : <EOL> raise TrainingExit ( status ) <EOL> class EarlyStopping ( Task ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cost , lookahead , eps = <NUM_LIT:0.> , min_nb_epochs = <NUM_LIT:0> , callback = None ) : <EOL> super ( ) . __init__ ( ) <EOL> self . cost = cost <EOL> self . best_epoch = <NUM_LIT:0> <EOL> self . best_cost = np . inf <EOL> self . lookahead = lookahead <EOL> self . callback = callback <EOL> self . eps = eps <EOL> self . min_nb_epochs = min_nb_epochs <EOL> self . stash = None <EOL> def init ( self , status ) : <EOL> self . best_cost = self . cost . view ( status ) <EOL> self . _stash_model ( status ) <EOL> def post_epoch ( self , status ) : <EOL> cost = float ( self . cost . view ( status ) ) <EOL> if cost + self . eps < self . best_cost : <EOL> self . best_epoch = status . current_epoch <EOL> self . best_cost = cost <EOL> self . _stash_model ( status ) <EOL> if self . callback is not None : <EOL> self . callback ( self , status ) <EOL> if status . current_epoch - self . best_epoch >= self . lookahead and status . current_epoch >= self . min_nb_epochs : <EOL> raise TrainingExit ( status ) <EOL> def finished ( self , status ) : <EOL> self . _restore_model ( status ) <EOL> print ( "<STR_LIT>" . format ( self . best_cost , self . best_epoch ) ) <EOL> def _stash_model ( self , status ) : <EOL> model = status . trainer . _optimizer . loss . model <EOL> self . stash = [ p . get_value ( ) for p in model . parameters ] <EOL> def _restore_model ( self , status ) : <EOL> model = status . trainer . _optimizer . loss . model <EOL> for param , value in zip ( model . parameters , self . stash ) : <EOL> param . set_value ( value ) <EOL> def save ( self , path ) : <EOL> state = { "<STR_LIT:version>" : <NUM_LIT:1> , <EOL> "<STR_LIT>" : self . best_epoch , <EOL> "<STR_LIT>" : self . best_cost , <EOL> "<STR_LIT>" : self . lookahead , <EOL> "<STR_LIT>" : self . eps , <EOL> "<STR_LIT>" : self . min_nb_epochs , <EOL> "<STR_LIT>" : self . stash } <EOL> utils . save_dict_to_json_file ( pjoin ( path , "<STR_LIT>" ) , state ) <EOL> def load ( self , path ) : <EOL> state = utils . load_dict_from_json_file ( pjoin ( path , "<STR_LIT>" ) ) <EOL> self . best_epoch = state [ "<STR_LIT>" ] <EOL> self . best_cost = state [ "<STR_LIT>" ] <EOL> self . stash = state [ "<STR_LIT>" ] </s>
