Aktueller Sand - FillAggregateTableMain2 funktioniert noch nicht richtig

src/main/java/info/peper/vz/rest/FillAggregateTableMain2.java

@@ -66,16 +66,18 @@ public class FillAggregateTableMain2 {
 				"jdbc:mariadb://mariadb.fritz.box/volkszaehler",
 				"vz",
 				getPassword())) {
-			long startTimeStamp = getTimestamp("2022-05-20T09:00:00");
+//			long startTimeStamp = getTimestamp("2025-01-01T09:00:00");
+			final long startTimeStamp = getTimestamp("2022-05-20T09:00:00");
 			final long finalEndTimeStamp = getTimestamp("2025-03-01T00:00:00");
 			for (int channelId : CHANNELS_TO_USE) {
 				processChannel(startTimeStamp, finalEndTimeStamp, 60*1000, channelId, con);
 			}
-			System.out.println("Saving...");
+			System.out.println("Saving... " + SAVELIST.size());
-			for (AggregateToSave ats : SAVELIST) {
+//			saveValues(con);
-				saveValues(con, ats.startTs, ats.endTs, ats.channelId, new long[] {ats.valuePos, ats.valueNeg});
+//			for (AggregateToSave ats : SAVELIST) {
-				System.out.println("####");
+//				saveValues(con, ats.startTs, ats.endTs, ats.channelId, new long[] {ats.valuePos, ats.valueNeg});
-			}
+//				System.out.println("####");
+//			}
 		}
 	}
 	
@@ -95,11 +97,18 @@ public class FillAggregateTableMain2 {
 			stmt.setFetchSize(1000);
 			boolean noNextRecord = false;
 			try (final ResultSet rs = stmt.executeQuery()) {
-				while (noNextRecord || (rs.next() && currentTimestamp <= endTimestamp)) {
+				long sumDiff = 0;
+				while (noNextRecord || (currentTimestamp <= endTimestamp && rs.next())) {
 					noNextRecord = false;
 					final long timestamp = rs.getLong("timestamp");
-					final long value = rs.getLong("value");
+					final long value;
+					if (rs.getDouble("value") >= 0 && rs.getDouble("value") < 1) {
+						value = 0;
+					} else {
+						value = rs.getLong("value");
+					}
 					final long tsDiff = Math.min(timestamp - currentTimestamp, intervalEndTimestamp - currentTimestamp);
+					sumDiff += tsDiff;
 					currentTimestamp = timestamp;
 					if (value > 0) {
 						intervalPos += (value * tsDiff);
@@ -107,11 +116,17 @@ public class FillAggregateTableMain2 {
 						intervalNeg += (-value * tsDiff);
 					}
 					if (timestamp >= intervalEndTimestamp) {
-//						System.out.println(
+						if (sumDiff != interval) {
-//								DTF.format(Instant.ofEpochMilli(intervalStartTimestamp)) + " - " + 
+							System.err.println("sumDiff: " + sumDiff + " / interval: " + interval);
-//								DTF.format(Instant.ofEpochMilli(intervalEndTimestamp)) + " / " +
+						}
-//								(intervalPos/3600) + " / " +
+						sumDiff = 0;
-//								(intervalNeg/3600));
+						if (intervalStartTimestamp % (86400000*10) == 0) {
+							System.out.println(
+									DTF.format(Instant.ofEpochMilli(intervalStartTimestamp)) + " - " + 
+									DTF.format(Instant.ofEpochMilli(intervalEndTimestamp)) + " / " +
+									(intervalPos/3600) + " / " +
+									(intervalNeg/3600));
+						}
 						SAVELIST.add(new AggregateToSave(intervalStartTimestamp, intervalEndTimestamp, intervalPos/3600, intervalNeg/3600, channelId));
 						intervalStartTimestamp += interval;
 						intervalEndTimestamp += interval;
@@ -155,9 +170,21 @@ public class FillAggregateTableMain2 {
 	
 	private static void saveValues(final Connection con) throws SQLException {
 		try (final PreparedStatement stmt = con.prepareStatement("INSERT INTO tobias_aggregate (channel_id, timestamp_start, timestamp_end, sum_positive, sum_negative) VALUES (?, ?, ?, ?, ?)")) {
+			int i = 0;
 			for (AggregateToSave ats : SAVELIST) {
-				
+				stmt.setInt(1, ats.channelId);
+				stmt.setLong(2, ats.startTs);
+				stmt.setLong(3, ats.endTs);
+				stmt.setLong(4, ats.valuePos);
+				stmt.setLong(5, ats.valueNeg);
+				stmt.addBatch();
+				i++;
+				if (i % 100000 == 0) {
+					System.out.println(i);
+					stmt.executeBatch();
 				}
 			}
+			stmt.executeBatch();
+		}
 	}
 }

src/main/java/info/peper/vz/rest/ReadDbMain2.java

@@ -0,0 +1,107 @@
+package info.peper.vz.rest;
+
+import java.io.BufferedReader;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.TemporalAccessor;
+
+public class ReadDbMain2 {
+
+	private static final DateTimeFormatter DTF = DateTimeFormatter.ISO_LOCAL_DATE_TIME.withZone(ZoneId.systemDefault());
+	private static final int[] CHANNELS_TO_USE = new int[] {4};
+	
+	public static void main(String[] args) throws Exception {
+		try (final Connection con = DriverManager.getConnection(
+				"jdbc:mariadb://mariadb.fritz.box/volkszaehler",
+				"vz",
+				getPassword())) {
+			try (final PreparedStatement stmt = con.prepareStatement("SELECT timestamp FROM volkszaehler.data WHERE channel_id=4 AND timestamp >= 1740697200000 ORDER BY timestamp"); final ResultSet rs = stmt.executeQuery()) {
+				long lastTime = 0;
+				long longestDiff = 0;
+				while (rs.next()) {
+					if (lastTime > 0) {
+						long diff = rs.getLong("timestamp") - lastTime;
+						if (diff > longestDiff) {
+							longestDiff = diff;
+						}
+					}
+					lastTime = rs.getLong("timestamp");
+				}
+				System.out.println(longestDiff);
+			}
+		}
+	}
+
+	private static String getPassword() throws IOException {
+		try (final InputStream is = new FileInputStream("db-password.txt")) {
+			final BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(is));
+			return bufferedReader.readLine();
+		}
+	}
+
+	private static long getTimestamp(final String dateTime) {
+		final TemporalAccessor tempAccessor = DTF.parse(dateTime);
+		final Instant instant = Instant.from(tempAccessor);
+		return Instant.EPOCH.until(instant, ChronoUnit.MILLIS);
+	}
+	
+	private static void saveValues(final Connection con,
+			final long startTimestamp,
+			final long endTimestamp,
+			final int channelId,
+			final long[] values) throws SQLException {
+		try (final PreparedStatement stmt = con.prepareStatement("INSERT INTO tobias_aggregate (channel_id, timestamp_start, timestamp_end, sum_positive, sum_negative) VALUES (?, ?, ?, ?, ?)")) {
+			stmt.setInt(1, channelId);
+			stmt.setLong(2, startTimestamp);
+			stmt.setLong(3, endTimestamp);
+			stmt.setLong(4, values[0]);
+			stmt.setLong(5, values[1]);
+			stmt.execute();
+		}
+	}
+	
+	private static long[] getValues(final Connection con,
+			final long startTimestamp,
+			final long endTimestamp,
+			final int channelId) throws SQLException {
+		long currentTimestamp = startTimestamp;
+		long wattMillisecondsPos = 0;
+		long wattMillisecondsNeg = 0;
+		
+		while (currentTimestamp < endTimestamp) {
+			try (final PreparedStatement stmt = con.prepareStatement("SELECT * FROM volkszaehler.data WHERE channel_id=? AND timestamp>? ORDER BY timestamp;")) {
+				stmt.setMaxRows(1000);
+				stmt.setInt(1, channelId);
+				stmt.setLong(2, currentTimestamp);
+				try (final ResultSet rs = stmt.executeQuery()) {
+					while (rs.next() && currentTimestamp <= endTimestamp) {
+						final long rsTimestamp = rs.getLong("timestamp");
+						final long rsValue = rs.getLong("value");
+						final long tsDiff = rsTimestamp - Math.min(endTimestamp, currentTimestamp);
+						currentTimestamp = rsTimestamp;
+						if (rsValue > 0) {
+							wattMillisecondsPos += (rsValue * tsDiff);
+						} else if (rsValue < 0) {
+							wattMillisecondsNeg += (-rsValue * tsDiff);
+						}
+					}
+				}
+			}
+		}
+		return new long[] {wattMillisecondsPos/3600, wattMillisecondsNeg/3600};
+
+	}
+	
+}

src/main/java/info/peper/vz/rest/VzRestController.java

@@ -39,7 +39,7 @@ class VzRestController {
 	Sums getData(@RequestParam("timestampStart")final long timestampStart,
 			@RequestParam("timestampEnd")final long timestampEnd,
 			@RequestParam("channelId")final int channelId) {
-		this.jdbcTemplateWithLimitedRows.setMaxRows(1000);
+//		this.jdbcTemplateWithLimitedRows.setMaxRows(1000);
 		final List<Aggregate> aggregates = jdbcTemplate.query(
 				"SELECT * FROM volkszaehler.tobias_aggregate WHERE channel_id=? AND timestamp_start>=? AND timestamp_end<=? ORDER BY timestamp_start;", 
 				(rs, rowNum) -> new Aggregate(rs.getInt("channel_id"), rs.getLong("timestamp_start"), rs.getLong("timestamp_end"),
@@ -49,6 +49,7 @@ class VzRestController {
 		long lastTimestamp = Long.MIN_VALUE;
 		long sumPos = 0;
 		long sumNeg = 0;
+		log.info("Number of aggregates: " + aggregates.size());
 		for (Aggregate ag : aggregates) {
 			if (ag.getTimestampStart() < firstTimestamp) {
 				firstTimestamp = ag.getTimestampStart();
@@ -59,22 +60,22 @@ class VzRestController {
 			sumPos += ag.getSumPositive();
 			sumNeg += ag.getSumNegative();
 		}
-		log.debug("sumPos: " + sumPos + " / sumNeg: " + sumNeg);
+		log.info("sumPos: " + sumPos + " / sumNeg: " + sumNeg);
-		log.debug("timestampStart: " + timestampStart);
+		log.info("timestampStart: " + timestampStart);
-		log.debug("timestampEnd:   " + timestampEnd);
+		log.info("timestampEnd:   " + timestampEnd);
-		log.debug("firstTimestamp: " + firstTimestamp);
+		log.info("firstTimestamp: " + firstTimestamp);
-		log.debug("lastTimestamp:  " + lastTimestamp);
+		log.info("lastTimestamp:  " + lastTimestamp);
 		if (timestampStart < firstTimestamp ) {
 			final Sums startSums = getSums(timestampStart, firstTimestamp, channelId);
 			sumPos += startSums.getSumPositive();
 			sumNeg += startSums.getSumNegative();
-			log.debug("Start: " + startSums.toString());
+			log.info("Start: " + startSums.toString());
 		}
 		if (timestampEnd > lastTimestamp) {
 			final Sums endSums = getSums(lastTimestamp, timestampEnd, channelId);
 			sumPos += endSums.getSumPositive();
 			sumNeg += endSums.getSumNegative();
-			log.debug("End: " + endSums.toString());
+			log.info("End: " + endSums.toString());
 			
 		}
 		return new Sums(sumPos, sumNeg);