def test_epoch_monitor():
 dim = 1
 batch_size = 3
 n_batches = 10
 m = (n_batches * batch_size)
 dataset = ArangeDataset(m)
 model = SoftmaxModel(dim)
 monitor = Monitor.get_monitor(model)
 learning_rate = 0.001
 data_specs = (model.get_input_space(), model.get_input_source())
 cost = DummyCost()
 termination_criterion = EpochCounter(1)
 monitor_rate = 3
 em = EpochMonitor(model=model, tick_rate=None, monitor_rate=monitor_rate)
 algorithm = SGD(learning_rate, cost, batch_size=batch_size, train_iteration_mode='sequential', monitoring_dataset=dataset, termination_criterion=termination_criterion, update_callbacks=[em], set_batch_size=False)
 algorithm.setup(dataset=dataset, model=model)
 algorithm.train(dataset)
 for (key, val) in monitor.channels.items():
  assert (len(val.val_record) == (n_batches // monitor_rate))

def test_epoch_monitor ( ) : dim = 1 batch_size = 3 n_batches = 10 m = ( n_batches * batch_size ) dataset = ArangeDataset ( m ) model = SoftmaxModel ( dim ) monitor = Monitor . get_monitor ( model ) learning_rate = 0.001 data_specs = ( model . get_input_space ( ) , model . get_input_source ( ) ) cost = DummyCost ( ) termination_criterion = EpochCounter ( 1 ) monitor_rate = 3 em = EpochMonitor ( model = model , tick_rate = None , monitor_rate = monitor_rate ) algorithm = SGD ( learning_rate , cost , batch_size = batch_size , train_iteration_mode = 'sequential' , monitoring_dataset = dataset , termination_criterion = termination_criterion , update_callbacks = [ em ] , set_batch_size = False ) algorithm . setup ( dataset = dataset , model = model ) algorithm . train ( dataset ) for ( key , val ) in monitor . channels . items ( ) : assert ( len ( val . val_record ) == ( n_batches // monitor_rate ) )
checks that monitored channels contain expected number of values when using epochmonitor for updates within epochs .