<commit_before>package com.ociweb.pronghorn.image;

import java.util.Arrays;

import com.ociweb.pronghorn.iot.schema.ImageSchema;
import com.ociweb.pronghorn.pipe.DataInputBlobReader;
import com.ociweb.pronghorn.pipe.Pipe;
import com.ociweb.pronghorn.pipe.RawDataSchema;
import com.ociweb.pronghorn.stage.PronghornStage;
import com.ociweb.pronghorn.stage.math.HistogramSchema;
import com.ociweb.pronghorn.stage.scheduling.GraphManager;

public class MapImageStage extends PronghornStage {

	private int[]   workspace;
	private int[][] rowLookup;// [row][(col*256)+depth] full row stored together.
	private int[]   locations;//length + location data, each -1 terminated, no 0 offsets. 	
		
	private final Pipe<ImageSchema> imgInput; 
	private final Pipe<RawDataSchema>[] newMappingData;
	private final Pipe<HistogramSchema> output;

    	
	public static MapImageStage newInstance(GraphManager graphManager, 
            Pipe<ImageSchema> monochromeInput, 
            Pipe<HistogramSchema> output,
            Pipe<RawDataSchema> ... newMappingData
            ) {
		return new MapImageStage(graphManager, monochromeInput, output, newMappingData);
	}
	
	//need outgoing schema for the map.
	protected MapImageStage(GraphManager graphManager, 
			                Pipe<ImageSchema> imgInput, 
			                Pipe<HistogramSchema> output,
			                Pipe<RawDataSchema> ... newMappingData
			                ) {
		
		super(graphManager, join(imgInput,newMappingData), output);
		
		this.imgInput = imgInput;
		this.newMappingData = newMappingData;
		this.output = output;
	}

	private int loadingNewMap = -1;	
	private boolean imageInProgress = false;
	
	@Override
	public void run() {
		
		if (!imageInProgress) {
			if (loadingNewMap >= 0) {
				loadNewMaps(loadingNewMap);
			} else {
				int n = newMappingData.length;
				while (--n>=0) {
					loadNewMaps(n);
				}
			}
		}
		
		//if we are not loading a new map check for an image to process
		if (loadingNewMap<0) {
			
			
			//read in 1 image and based on the locations look up the summary for the map
			
			//		int row = 0;
			//		int column = 0;
			//		int brightness = 0;
			//		
			//		int pos = map[row][column][brightness];
			//		int idx;
			//		while ((idx = locations[pos++])!=-1) {
			//				workspace[idx]++;
			//		}
						
			
		}		
	}
	
	private void loadNewMaps(int n) {
		 Pipe<RawDataSchema> pipe = newMappingData[n];
		if (Pipe.hasContentToRead(pipe)) {
			loadingNewMap = n;
		}
		
		while (Pipe.hasContentToRead(pipe)) {
		
			int msgIdx = Pipe.takeMsgIdx(pipe);		
			if (msgIdx == RawDataSchema.MSG_CHUNKEDSTREAM_1) {
				DataInputBlobReader.accumLowLevelAPIField(Pipe.inputStream(pipe));
				Pipe.confirmLowLevelRead(pipe, Pipe.sizeOf(pipe, RawDataSchema.MSG_CHUNKEDSTREAM_1));
				Pipe.readNextWithoutReleasingReadLock(pipe);
					
				DataInputBlobReader<RawDataSchema> inputStream = Pipe.inputStream(pipe);
				
				//consume data if possible, note full pipe of data must be large enough for a block.
				if (inputStream.available()>=8) {
					int tempPos = inputStream.absolutePosition();//in case we need to roll back
					int blockType = inputStream.readInt();//4 bytes
					int blockSize = inputStream.readInt();//4 bytes
					if (inputStream.available()>=blockSize) {
						//consume we have all the data for this block
					    if (blockType>=0) {
					    	//row of data<commit_msg>Map image to location stage complete<commit_after>package com.ociweb.pronghorn.image;

import java.util.Arrays;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ociweb.pronghorn.iot.schema.ImageSchema;
import com.ociweb.pronghorn.pipe.DataInputBlobReader;
import com.ociweb.pronghorn.pipe.DataOutputBlobWriter;
import com.ociweb.pronghorn.pipe.Pipe;
import com.ociweb.pronghorn.pipe.RawDataSchema;
import com.ociweb.pronghorn.stage.PronghornStage;
import com.ociweb.pronghorn.stage.math.HistogramSchema;
import com.ociweb.pronghorn.stage.scheduling.GraphManager;

public class MapImageStage extends PronghornStage {

	private int[]   workspace;
	private int[][] rowLookup;// [row][(col*256)+depth] full row stored together.
	private int[]   locations;//length + location data, each -1 terminated, no 0 offsets. 	
		
	private final Pipe<ImageSchema> imgInput; 
	private final Pipe<RawDataSchema>[] newMappingData;
	private final Pipe<HistogramSchema> output;
	private boolean isShuttingDown = false;
	private static final Logger logger = LoggerFactory.getLogger(MapImageStage.class);
    	
	public static MapImageStage newInstance(GraphManager graphManager, 
            Pipe<ImageSchema> monochromeInput, 
            Pipe<HistogramSchema> output,
            Pipe<RawDataSchema> ... newMappingData
            ) {
		return new MapImageStage(graphManager, monochromeInput, output, newMappingData);
	}
	
	//need outgoing schema for the map.
	protected MapImageStage(GraphManager graphManager, 
			                Pipe<ImageSchema> imgInput, 
			                Pipe<HistogramSchema> output,
			                Pipe<RawDataSchema> ... newMappingData
			                ) {
		
		super(graphManager, join(imgInput,newMappingData), output);
		
		this.imgInput = imgInput;
		this.newMappingData = newMappingData;
		this.output = output;
	}

	private int loadingNewMap = -1;	
	private boolean imageInProgress = false;
	private int totalRows;
	private int totalWidth;
	private long time;
	private int activeRow;
	
	
	@Override
	public void run() {
		
		if (!isShuttingDown) {
			if (!imageInProgress) {
				if (loadingNewMap >= 0) {
					loadNewMaps(loadingNewMap);
				} else {
					int n = newMappingData.length;
					while (--n>=0) {
						loadNewMaps(n);
					}
				}
			} else {
				//we are image in progress
				if (activeRow == totalRows) {
					//we have a complete message to send
					publishHistogram();
				}				
			}
			
			//if we are not loading a new map check for an image to process
			if (loadingNewMap<0) {
				
				while (Pipe.hasContentToRead(imgInput)) {
					
					int msgIdx = Pipe.takeMsgIdx(imgInput);
					
					if (ImageSchema.MSG_FRAMECHUNK_2 == msgIdx) {
						
						if (activeRow < totalRows) {
							
							DataInputBlobReader<ImageSchema> rowData = Pipe.openInputStream(imgInput);
							
							for(int activeColumn = 0; activeColumn<totalWidth; activeColumn++) {
								int color = rowData.readByte();
								
								int pos = rowLookup[activeRow][(activeColumn*256)+color];
								if (pos >= 0) {
									int idx;
									while ((idx = locations[pos++])!=-1) {
											workspace[idx]++;
									}
								}
							}
							activeRow++;
							if (activeRow == totalRows) {
								publishHistogram();
								break;
							}							
						} else {
							//error too many rows.
							logger.error("too many rows only expected {}",totalRows);
							Pipe.skipNextFragment(imgInput, msgIdx);
						}
						
					} else if (ImageSchema.MSG_FRAMESTART_1 == msgIdx) {
						imageInProgress = true;
						totalWidth = Pipe.takeInt(imgInput);
						totalRows = Pipe.takeInt(imgInput);
						time = Pipe.takeLong(imgInput); 
						Arrays.fill(workspace, 0);
						activeRow = 0;
					} else {
						//shutdown...
						isShuttingDown = true;
						Pipe.confirmLowLevelWrite(imgInput, Pipe.EOF_SIZE);
						Pipe.releaseReadLock(imgInput);
						break;
					}
				}
			}		
		} else {
			if (Pipe.hasRoomForWrite(output)) {
				Pipe.publishEOF(output);
				requestShutdown();
			}
		}
	}

	private void publishHistogram() {
		if (Pipe.hasRoomForWrite(output)) {
			int size = Pipe.addMsgIdx(output, HistogramSchema.MSG_HISTOGRAM_1);
			
			Pipe.addIntValue(workspace.length, output);
			
			DataOutputBlobWriter<HistogramSchema> outputStream = Pipe.openOutputStream(output);								
			int i = workspace.length;
			while (--i>=0) {
				outputStream.writePackedInt(workspace[i]);
			}
			DataOutputBlobWriter.closeLowLevelField(outputStream);
			
			Pipe.confirmLowLevelWrite(output, size);
			Pipe.releaseReadLock(output);
			
			totalRows = 0;//clear we have sent the value.
			imageInProgress = false;
		}
	}
	
	private void loadNewMaps(int n) {
		 Pipe<RawDataSchema> pipe = newMappingData[n];
		if (Pipe.hasContentToRead(pipe)) {
			loadingNewMap = n;
		}
		
		while (Pipe.hasContentToRead(pipe)) {
		
			int msgIdx = Pipe.takeMsgIdx(pipe);		
			if (msgIdx == RawDataSchema.MSG_CHUNKEDSTREAM_1) {
				DataInputBlobReader.accumLowLevelAPIField(Pipe.inputStream(pipe));
				Pipe.confirmLowLevelRead(pipe, Pipe.sizeOf(pipe, RawDataSchema.MSG_CHUNKEDSTREAM_1));
				Pipe.readNextWithoutReleasingReadLock(pipe);
					
				DataInputBlobReader<RawDataSchema> inputStream = Pipe.inputStream(pipe);
				
				//consume data if possible, note full pipe of data must be large enough for a block.
				if (inputStream.available()>=8) {
					int tempPos = inputStream.absolutePosition();//in case we need to roll back
					int blockType = inputStream.readInt();//4 bytes
					int blockSize = inputStream.readInt();//4 bytes
					if (inputStream.available()>=blockSize) {
						//consume we have all the data for this block
					    if (blockType>=0) {
					    	//row of data
{'fraction_non_alphanumeric': 0.10550609957138148, 'fraction_numerical': 0.005495109352676118, 'mean_word_length': 2.991228070175439, 'pattern_counts': {'":': 0, '<': 31, '<?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': '3017537', 'n_tokens_mistral': 3569, 'n_tokens_neox': 2830, 'n_words': 634}