Untitled

package compress;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import dsl.*;
import ecg.Data;

public class Compress {

	public static final int BLOCK_SIZE = 10;

	public static Query<Integer, Integer> delta() {
		return new Query<Integer, Integer>() {
			private int previous;
			private int count;
			@Override
			public void start(Sink<Integer> sink) {
				this.previous = 0;
				this.count = 0;
			}

			@Override
			public void next(Integer item, Sink<Integer> sink) {
				count++;
				if (count % BLOCK_SIZE == 1) {
					sink.next(item);
					/*System.out.println("Item");
					System.out.println(item);
					System.out.println("Prev");
					System.out.println(previous);*/
				}
				else {
					sink.next(item - previous);
				}
				previous = item;
			}

			@Override
			public void end(Sink<Integer> sink) {
				sink.end();
			}
		};
	}
	public static Query<Integer,Integer> deltaInv() {
		// TODO
		return new Query<>() {
			private int lastValue;
			private int count;
			@Override
			public void start(Sink<Integer> sink) {
				this.lastValue = 0;
				this.count = 0;
			}

			@Override
			public void next(Integer delta, Sink<Integer> sink) {
				//System.out.println("Delta");
				//System.out.println(delta);
				count++;
				if(count%BLOCK_SIZE == 1)
					lastValue = delta;
				else
					lastValue += delta;
				//System.out.println(lastValue);
				sink.next(lastValue);
			}

			@Override
			public void end(Sink<Integer> sink) {
				sink.end();
			}
		};
	}

	public static Query<Integer,Integer> zigzag() {
		// TODO
		return new Map<>(x -> (x << 1) ^ (x >> 31));
	}

	public static Query<Integer,Integer> zigzagInv() {
		// TODO
		return new Map<>(x -> (x >> 1) ^ -(x & 1));
	}

	public static Query<Integer,Integer> pack() {
		// TODO
		return new Query<>() {
			private int value;
			private int remainBits;
			private int needBits;
			private int header;
			private int count;
			List<Integer> list = new ArrayList<>();
			@Override
			public void start(Sink<Integer> sink) {
				this.value = 0;
				this.remainBits = 0;
				this.header = 0;
				this.count = 0;
			}

			@Override
			public void next(Integer item, Sink<Integer> sink) {
				count++;
				//System.out.println("Pack");
				//System.out.println(item);
				if (count%BLOCK_SIZE == 1) {
					list.add(item);
				}
				else {
					needBits = Integer.SIZE - Integer.numberOfLeadingZeros(item);
					if (remainBits < needBits || (header != 0 && needBits > header)) {
						if (value != 0) {
							list.add(value);
						}
						value = needBits << 5;
						header = needBits;
						remainBits = 5 - header;
						value |= (item << (remainBits));
					} else {
						remainBits -= header;
						value |= (item << remainBits);
					}
				}
				if (count % BLOCK_SIZE == 0) {
					//System.out.println("Pack");
					flush(sink);
				}
			}

			@Override
			public void end(Sink<Integer> sink) {
				sink.end();
			}

			private void flush(Sink<Integer> sink) {
				if (value != 0) {
					list.add(value);
					value = 0;
					remainBits = 0;
				}
				for (Integer val : list) {
					sink.next(val);
				}
				list.clear();
				count = 0;
				header = 0;
			}
		};
	}

	public static Query<Integer,Integer> unpack() {
		// TODO
		return new Query<Integer, Integer>() {
			private int header;
			private int value;
			List<Integer> list = new ArrayList<>();
			@Override
			public void start(Sink<Integer> sink) {
				this.value = 0;
				this.header = 0;
				this.list.clear();
			}

			@Override
			public void next(Integer item, Sink<Integer> sink) {
				System.out.println(item);
				//System.out.println("-----------------------");
				if (list.size() == 0) {
					list.add(item);
					//sink.next(item);
					System.out.println(item);
				}
				else {
					header = item >> 5;
					value = item & ((1 << 5) - 1); // Mask to extract only the value part
					System.out.println("value is");
					System.out.println(value);
					int totalBits;
					while (value != 0) {
						// Extract leftmost x bits
						totalBits = Integer.SIZE - Integer.numberOfLeadingZeros(value);
						//System.out.println("Totalbits is");
						//System.out.println(totalBits);
						int leftXBits = value >> (totalBits - header);
						//sink.next(leftXBits);
						list.add(leftXBits);
						//System.out.println("Unpack");
						System.out.println(leftXBits);
						value = value & ((1 << (totalBits - header)) - 1);
						System.out.println("New value is");
						System.out.println(value);
					}
				}
				//System.out.println("-----------------------");
				if (list.size() == BLOCK_SIZE) {
					for (Integer ele : list) {
						sink.next(ele);
					}
					list.clear();
					value = 0;
					header = 0;
				}
			}

			@Override
			public void end(Sink<Integer> sink) {
				sink.end();
			}
		};

	}

	public static Query<Integer,Integer> compress() {
		// TODO
		return Q.pipeline(
				delta(),
				zigzag(),
				pack());
	}

	public static Query<Integer,Integer> decompress() {
		// TODO
		return Q.pipeline(
				unpack(),
				zigzagInv(),
				deltaInv());
	}

	public static void main(String[] args) {
		System.out.println("**********************************************");
		System.out.println("***** ToyDSL & Compression/Decompression *****");
		System.out.println("**********************************************");
		System.out.println();

		System.out.println("***** Compress *****");{
			// from range [0,2048) to [0,256)
			Query<Integer,Integer> q1 = Q.map(x -> x / 8);
			Query<Integer,Integer> q2 = compress();
			Query<Integer,Integer> q = Q.pipeline(q1, q2);
			Iterator<Integer> it = Data.ecgStream("BLOCK_SIZE0-all.csv");
			Q.execute(it, q, S.lastCount());
		}
		System.out.println();

		System.out.println("***** Compress & Decompress *****");{
			// from range [0,2048) to [0,256)
			Query<Integer,Integer> q1 = Q.map(x -> x / 8);
			Query<Integer,Integer> q2 = compress();
			Query<Integer,Integer> q3 = decompress();
			Query<Integer,Integer> q = Q.pipeline(q1, q2, q3);
			Iterator<Integer> it = Data.ecgStream("BLOCK_SIZE0-all.csv");
			Q.execute(it, q, S.lastCount());
		}
		System.out.println();
	}

}