<commit_before>package be.iminds.iot.dianne.nn.learn;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Reference;
import org.osgi.service.component.annotations.ReferenceCardinality;
import org.osgi.service.component.annotations.ReferencePolicy;

import be.iminds.iot.dianne.api.dataset.Dataset;
import be.iminds.iot.dianne.api.nn.learn.Learner;
import be.iminds.iot.dianne.api.nn.learn.Processor;
import be.iminds.iot.dianne.api.nn.module.Input;
import be.iminds.iot.dianne.api.nn.module.Module;
import be.iminds.iot.dianne.api.nn.module.Output;
import be.iminds.iot.dianne.api.nn.module.Preprocessor;
import be.iminds.iot.dianne.api.nn.module.Trainable;
import be.iminds.iot.dianne.api.nn.module.dto.ModuleDTO;
import be.iminds.iot.dianne.api.nn.module.dto.ModuleInstanceDTO;
import be.iminds.iot.dianne.api.nn.module.dto.NeuralNetworkDTO;
import be.iminds.iot.dianne.api.nn.module.dto.NeuralNetworkInstanceDTO;
import be.iminds.iot.dianne.api.nn.runtime.ModuleManager;
import be.iminds.iot.dianne.api.repository.DianneRepository;
import be.iminds.iot.dianne.nn.learn.processors.MomentumProcessor;
import be.iminds.iot.dianne.nn.learn.processors.RandomBatchProcessor;
import be.iminds.iot.dianne.nn.learn.processors.RegularizationProcessor;
import be.iminds.iot.dianne.tensor.TensorFactory;

@Component
public class SimpleLearner implements Learner {

	private TensorFactory factory;
	private DianneRepository repository;
	private ModuleManager runtime;
	private Map<String, Dataset> datasets = new HashMap<String, Dataset>();
	
	private Thread learnerThread = null;
	private volatile boolean learning = false;
	
	private int updateInterval = 0;
	
	@Override
	public void learn(String nnName, String dataset,
			Map<String, String> config) throws Exception {
		if(learning){
			throw new Exception("Already running a learning session here");
		}
		
		// Fetch the dataset
		Dataset d = datasets.get(dataset);
		if(d==null){
			throw new Exception("Dataset "+dataset+" not available");
		}
		
		// Deploy an instance of nnName to train on
		NeuralNetworkDTO nn = repository.loadNeuralNetwork(nnName);
		UUID nnId = UUID.randomUUID();
		List<ModuleInstanceDTO> moduleInstances = new ArrayList<ModuleInstanceDTO>();
		for(ModuleDTO module : nn.modules){
			ModuleInstanceDTO instance = runtime.deployModule(module, nnId);
			moduleInstances.add(instance);
		}
		NeuralNetworkInstanceDTO nni = new NeuralNetworkInstanceDTO(nnId, nnName, moduleInstances);

		Input input = null;
		Output output = null;
		Map<UUID, Trainable> toTrain = new HashMap<>();
		Set<Preprocessor> preprocessing = new HashSet<>();
		
		for(ModuleInstanceDTO mi : nni.modules){
			Module m = runtime.getModule(mi.moduleId, mi.nnId);
			if(m instanceof Input){
				input = (Input) m;
			} else if(m instanceof Output){
				output = (Output) m;
			} else if(m instanceof Trainable){
				toTrain.put(mi.moduleId, (Trainable)m);
			} else if(m instanceof Preprocessor){
				preprocessing.add((Preprocessor) m);
			}	
		}
		
		// first get parameters for preprocessing?
		preprocessing.stream().forEach(p -> {
			if(!p.isPreprocessed())
				p.preprocess(d);
			}
		);
		
		// create a Processor from config
		// for now just fixed
		Processor p = new MomentumProcessor(new RegularizationProcessor(new RandomBatchProcessor(factory, input, output, toTrain, d, config)));
		
		
		learnerThread = new Thread(new Runnable() {
			
			@Override
			public void run() {
				learning = true;
				int i = 0;
				float runningAvg = 0;
				float error = 0;

				do {
					i++;
					error = p.processNext();
					runningAvg+= error;
					
					System.out.println(error+" - "+runningAvg/i);
					
					if(error >= 0){
						
						toTrain.entrySet().stream().forEach(e -> {
							e.getValue().updateParameters(1.0f);
							e.getValue().zeroDeltaParameters();
						});
						
						if(updateInterval>0){
							if(i % updateInterval == 0){
								// publish weights
								publishParameters();
							}
						}
					}
					

				} while(learning && error >= 0);
				System.out.println("Stopped learning");
				publishParameters();
			}
		});
		learnerThread.start();
	}
	
	@Override
	public void stop() {
		if(learning){
			learning = false;
			learnerThread.interrupt();
			learnerThread = null;
		}
	}

	private void publishParameters(){
		System.out.println("Publish parameters");
		
		// TODO collect all parameter deltas to update repository
	}
	
	@Reference
	void setTensorFactory(TensorFactory f){
		this.factory = f;
	}
	
	@Reference
	void setDianneRepository(DianneRepository r){
		this.repository = r;
	}
	
	@Reference
	void setModuleManager(ModuleManager m){
		this.runtime = m;
	}
	<commit_msg>update repo parameters during learning <commit_after>package be.iminds.iot.dianne.nn.learn;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.stream.Collectors;

import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Reference;
import org.osgi.service.component.annotations.ReferenceCardinality;
import org.osgi.service.component.annotations.ReferencePolicy;

import be.iminds.iot.dianne.api.dataset.Dataset;
import be.iminds.iot.dianne.api.nn.learn.Learner;
import be.iminds.iot.dianne.api.nn.learn.Processor;
import be.iminds.iot.dianne.api.nn.module.Input;
import be.iminds.iot.dianne.api.nn.module.Module;
import be.iminds.iot.dianne.api.nn.module.Output;
import be.iminds.iot.dianne.api.nn.module.Preprocessor;
import be.iminds.iot.dianne.api.nn.module.Trainable;
import be.iminds.iot.dianne.api.nn.module.dto.ModuleDTO;
import be.iminds.iot.dianne.api.nn.module.dto.ModuleInstanceDTO;
import be.iminds.iot.dianne.api.nn.module.dto.NeuralNetworkDTO;
import be.iminds.iot.dianne.api.nn.module.dto.NeuralNetworkInstanceDTO;
import be.iminds.iot.dianne.api.nn.runtime.ModuleManager;
import be.iminds.iot.dianne.api.repository.DianneRepository;
import be.iminds.iot.dianne.nn.learn.processors.MomentumProcessor;
import be.iminds.iot.dianne.nn.learn.processors.RandomBatchProcessor;
import be.iminds.iot.dianne.nn.learn.processors.RegularizationProcessor;
import be.iminds.iot.dianne.tensor.Tensor;
import be.iminds.iot.dianne.tensor.TensorFactory;

@Component
public class SimpleLearner implements Learner {

	protected String[] tag = new String[]{"test"};
	
	protected TensorFactory factory;
	protected DianneRepository repository;
	protected ModuleManager runtime;
	protected Map<String, Dataset> datasets = new HashMap<String, Dataset>();
	
	protected Thread learnerThread = null;
	protected volatile boolean learning = false;
	
	protected int updateInterval = 1000;
	
	// the network we are currently training
	protected NeuralNetworkInstanceDTO nni;
	protected Input input;
	protected Output output;
	protected Map<UUID, Trainable> toTrain;
	protected Set<Preprocessor> preprocessing;
	
	// initial parameters
	protected Map<UUID, Tensor> parameters = null;
	
	@Override
	public void learn(String nnName, String dataset,
			Map<String, String> config) throws Exception {
		if(learning){
			throw new Exception("Already running a learning session here");
		}
		
		// Fetch the dataset
		Dataset d = datasets.get(dataset);
		if(d==null){
			throw new Exception("Dataset "+dataset+" not available");
		}
		
		// Deploy an instance of nnName to train on
		NeuralNetworkDTO nn = repository.loadNeuralNetwork(nnName);
		UUID nnId = UUID.randomUUID();
		List<ModuleInstanceDTO> moduleInstances = new ArrayList<ModuleInstanceDTO>();
		for(ModuleDTO module : nn.modules){
			ModuleInstanceDTO instance = runtime.deployModule(module, nnId);
			moduleInstances.add(instance);
		}
		nni = new NeuralNetworkInstanceDTO(nnId, nnName, moduleInstances);

		input = null;
		output = null;
		toTrain = new HashMap<>();
		preprocessing = new HashSet<>();
		
		for(ModuleInstanceDTO mi : nni.modules){
			Module m = runtime.getModule(mi.moduleId, mi.nnId);
			if(m instanceof Input){
				input = (Input) m;
			} else if(m instanceof Output){
				output = (Output) m;
			} else if(m instanceof Trainable){
				toTrain.put(mi.moduleId, (Trainable)m);
			} else if(m instanceof Preprocessor){
				preprocessing.add((Preprocessor) m);
			}	
		}
		
		// load parameters
		loadParameters();
		
		
		// first get parameters for preprocessing?
		preprocessing.stream().forEach(p -> {
			if(!p.isPreprocessed())
				p.preprocess(d);
			}
		);
		
		// create a Processor from config
		// for now just fixed
		Processor p = new MomentumProcessor(new RegularizationProcessor(new RandomBatchProcessor(factory, input, output, toTrain, d, config)));
		
		
		learnerThread = new Thread(new Runnable() {
			
			@Override
			public void run() {
				learning = true;
				int i = 0;
				float runningAvg = 0;
				float error = 0;

				do {
					i++;
					error = p.processNext();
					runningAvg+= error;
					
					System.out.println(error+" - "+runningAvg/i);
					
					if(error >= 0){
						
						toTrain.entrySet().stream().forEach(e -> {
							e.getValue().updateParameters(1.0f);
							e.getValue().zeroDeltaParameters();
						});
						
						if(updateInterval>0){
							if(i % updateInterval == 0){
								// publish weights
								publishParameters();
							}
						}
					}
					

				} while(learning && error >= 0);
				System.out.println("Stopped learning");
				publishParameters();
			}
		});
		learnerThread.start();
	}
	
	@Override
	public void stop() {
		if(learning){
			learning = false;
			learnerThread.interrupt();
			learnerThread = null;
		}
	}

	protected void loadParameters(){
		try {
			parameters = repository.loadParameters(toTrain.keySet(), tag);
			parameters.entrySet().stream().forEach( e -> toTrain.get(e.getKey()).setParameters(e.getValue()));
		} catch(Exception e){
			// if no initial parameters available, publish the random initialize parameters of this instance as first parameters?
			publishParameters();
			System.out.println("No initial parameters available, publish these random values as initial parameters...");
		}
	}
	
	protected void publishParameters(){
		System.out.println("Publish parameters");
		
		// TODO collect all parameter deltas to update repository
		Map<UUID, Tensor> newParameters = toTrain.entrySet().stream().collect(
				Collectors.toMap(e -> e.getKey(), e -> e.getValue().getParameters()));
		
		if(parameters!=null){
			// compose diff
			Map<UUID, Tensor> accParameters = newParameters.entrySet().stream().collect(
					Collectors.toMap(e -> e.getKey(), 
									 e -> factory.getTensorMath().sub(null, newParameters.get(e.getKey()), parameters.get(e.getKey()))));
			repository.accParameters(accParameters, tag);
		} else {
			// just publish initial values
			repository.storeParameters(newParameters, tag);
		}
		
		// fetch update again from repo (could be merged from other learners)
		loadParameters();
	}
	
	@Reference
	void setTensorFactory(TensorFactory f){
		this.factory = f;
	}
	
	@Reference
	void setDianneRepository(DianneRepository r){
		this.repository = r;
	}
	
	@Reference
	void setModuleManager(ModuleManager m){
		this.runtime = m;
	}
	
{'fraction_non_alphanumeric': 0.121374582527685, 'fraction_numerical': 0.0020214448936544207, 'mean_word_length': 3.8941935483870966, 'pattern_counts': {'":': 0, '<': 24, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3440137', 'n_tokens_mistral': 4180, 'n_tokens_neox': 3573, 'n_words': 768}