[Backend/Edge] Make resend (backfill) work with a plain Timedata.InfluxDB

io.openems.backend.timedata.influx/bnd.bnd

@@ -10,6 +10,7 @@ Bundle-Version: 1.0.0.${tstamp}
 	io.openems.shared.influxdb,\
 	io.openems.wrapper.influxdb-client-core,\
 	io.openems.wrapper.influxdb-client-java,\
+	io.openems.wrapper.influxdb-client-utils,\
 
 -testpath: \
 	${testpath}

io.openems.backend.timedata.influx/src/io/openems/backend/timedata/influx/TimedataInfluxDb.java

@@ -159,7 +159,19 @@ public void write(String edgeId, AggregatedDataNotification notification) {
 
 	@Override
 	public void write(String edgeId, ResendDataNotification data) {
-		// TODO Auto-generated method stub
+		if (this.config.isReadOnly()) {
+			return;
+		}
+
+		// Write resent (backfilled) data to the default location, like timestamped data,
+		// so gaps from a reconnect are filled instead of silently dropped.
+		this.writeData(//
+				edgeId, //
+				data, //
+				(influxEdgeId, channel) -> {
+					this.timestampedChannelsForEdge.put(influxEdgeId, channel);
+					return true;
+				});
 	}
 
 	private boolean isTimestampedChannel(int edgeId, String channel) {

io.openems.backend.timedata.influx/test/io/openems/backend/timedata/influx/TimedataInfluxDbTest.java

@@ -0,0 +1,92 @@
+package io.openems.backend.timedata.influx;
+
+import static io.openems.common.utils.ReflectionUtils.setAttributeViaReflection;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import com.google.common.collect.TreeBasedTable;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonPrimitive;
+import com.influxdb.client.write.Point;
+
+import io.openems.common.jsonrpc.notification.AggregatedDataNotification;
+import io.openems.common.jsonrpc.notification.ResendDataNotification;
+import io.openems.common.oem.DummyOpenemsBackendOem;
+import io.openems.shared.influxdb.InfluxConnector;
+
+public class TimedataInfluxDbTest {
+
+	private static final String EDGE_ID = "edge0";
+	private static final String CHANNEL = "ess0/ActivePower";
+	private static final long TIMESTAMP = 1577836800000L;
+
+	@Test
+	public void testWriteResendDataNotification() throws Exception {
+		final var influxConnector = mock(InfluxConnector.class);
+		final var sut = createTimedataInfluxDb(false, influxConnector);
+
+		sut.write(EDGE_ID, createResendDataNotification());
+		sut.write(EDGE_ID, createAggregatedDataNotification());
+
+		final var captor = ArgumentCaptor.forClass(Point.class);
+		verify(influxConnector, times(2)).write(captor.capture());
+
+		final var resendPoint = captor.getAllValues().get(0);
+		assertTrue(resendPoint.hasFields());
+		assertEquals("0", resendPoint.getTags().get("edge"));
+		assertEquals(123L, resendPoint.getFields().get(CHANNEL));
+		assertFalse(captor.getAllValues().get(1).hasFields());
+	}
+
+	@Test
+	public void testWriteResendDataNotificationInReadOnlyMode() throws Exception {
+		final var influxConnector = mock(InfluxConnector.class);
+		final var sut = createTimedataInfluxDb(true, influxConnector);
+
+		sut.write(EDGE_ID, createResendDataNotification());
+
+		verify(influxConnector, never()).write(any(Point.class));
+	}
+
+	private static TimedataInfluxDb createTimedataInfluxDb(boolean isReadOnly, InfluxConnector influxConnector)
+			throws Exception {
+		final var sut = new TimedataInfluxDb();
+		setAttributeViaReflection(sut, "config", createConfig(isReadOnly));
+		setAttributeViaReflection(sut, "timeFilter", TimeFilter.from("", ""));
+		setAttributeViaReflection(sut, "channelFilter", ChannelFilter.from(new String[0], new String[0]));
+		setAttributeViaReflection(sut, "influxConnector", influxConnector);
+		setAttributeViaReflection(sut, "oem", new DummyOpenemsBackendOem());
+		return sut;
+	}
+
+	private static Config createConfig(boolean isReadOnly) {
+		final var config = mock(Config.class);
+		when(config.isReadOnly()).thenReturn(isReadOnly);
+		when(config.measurement()).thenReturn("data");
+		when(config.blacklistedChannels()).thenReturn(new String[0]);
+		when(config.blacklistedChannelIds()).thenReturn(new String[0]);
+		return config;
+	}
+
+	private static ResendDataNotification createResendDataNotification() {
+		final var data = TreeBasedTable.<Long, String, JsonElement>create();
+		data.put(TIMESTAMP, CHANNEL, new JsonPrimitive(123));
+		return new ResendDataNotification(data);
+	}
+
+	private static AggregatedDataNotification createAggregatedDataNotification() {
+		final var data = TreeBasedTable.<Long, String, JsonElement>create();
+		data.put(TIMESTAMP, CHANNEL, new JsonPrimitive(456));
+		return new AggregatedDataNotification(data);
+	}
+}

io.openems.edge.timedata.rrd4j/src/io/openems/edge/timedata/rrd4j/Rrd4jReadHandler.java

@@ -34,6 +34,7 @@
 import io.openems.common.exceptions.OpenemsException;
 import io.openems.common.timedata.Resolution;
 import io.openems.common.types.ChannelAddress;
+import io.openems.common.types.OpenemsType;
 import io.openems.edge.common.channel.Channel;
 import io.openems.edge.common.component.ComponentManager;
 import io.openems.edge.timedata.api.Timeranges;
@@ -223,42 +224,48 @@ public SortedMap<Long, SortedMap<ChannelAddress, JsonElement>> queryResendData(/
 					continue;
 				}
 
-				for (int i = 0; i < database.getArcCount(); i++) {
-					final var archive = database.getArchive(i);
-					final var arcStep = archive.getArcStep();
+				// Read at the channel's native storage resolution, mirroring
+				// queryHistoricData. The previous per-archive 'adjustSeconds = arcStep - 300'
+				// logic silently dropped AVERAGE channels (e.g. Voltage, Current, Frequency):
+				// for a 300s AVERAGE archive the shift was 0, so the narrow (<=300s) resend
+				// window left rrd4j slots either NaN or just outside the window and only
+				// cumulated (3600s) channels survived. Selecting the archive that matches the
+				// channel's consolidation and reading it at its own step fixes backfill for
+				// instantaneous channels while keeping cumulated channels hourly.
+				final var chDef = Rrd4jSupplier.getDsDefForChannel(channel.channelDoc().getUnit());
 
-					final var adjustSeconds = arcStep - Rrd4jConstants.DEFAULT_STEP_SECONDS;
+				long resolution = Rrd4jConstants.DEFAULT_STEP_SECONDS;
+				for (int i = 0; i < database.getArcCount(); i++) {
+					if (database.getArchive(i).getConsolFun() == chDef.consolFun()) {
+						resolution = database.getArchive(i).getArcStep();
+						break;
+					}
+				}
 
-					final var start = Math.max(fromTime - adjustSeconds, archive.getStartTime());
-					final var stop = Math.min(toTime - adjustSeconds, archive.getEndTime());
-					if (start > archive.getEndTime()) {
+				final var fetchData = database
+					.createFetchRequest(chDef.consolFun(), fromTime, toTime, resolution) //
+					.fetchData();
+				final var result = Rrd4jSupplier.postProcessData(fetchData, resolution);
+				final var firstTimestamp = fetchData.getFirstTimestamp();
+				for (int i = 0; i < result.length; i++) {
+					final var value = result[i];
+					if (Double.isNaN(value)) {
 						continue;
 					}
-					if (stop < archive.getStartTime()) {
+					final var timestamp = firstTimestamp + (i * resolution);
+					if (timestamp < fromTime || timestamp > toTime) {
 						continue;
 					}
-
-					final var fetchData = database.createFetchRequest(archive.getConsolFun(), start, stop, arcStep) //
-							.fetchData();
-
-					final var timestamps = fetchData.getTimestamps();
-					final var values = fetchData.getValues()[0];
-					for (int j = 0; j < values.length; j++) {
-						final var value = values[j];
-						if (Double.isNaN(value)) {
-							continue;
-						}
-						final var timestamp = timestamps[j] + adjustSeconds;
-
-						if (timestamp < fromTime //
-								|| timestamp > toTime) {
-							continue;
-						}
-
-						// return timestamps in milliseconds
-						resultMap.computeIfAbsent(timestamp * 1000, t -> new TreeMap<>()) //
-								.put(channelAddress, new JsonPrimitive(value));
-					}
+					// coerce the rrd4j double back to the channel's type so integer/boolean
+					// channels are written with the same InfluxDB field type as live data;
+					// otherwise a float-vs-integer field conflict makes InfluxDB drop the point.
+					final JsonPrimitive jsonValue = switch (channel.getType()) {
+						case BOOLEAN, SHORT, INTEGER, LONG -> new JsonPrimitive(Long.valueOf(Math.round(value)));
+						default -> new JsonPrimitive(Double.valueOf(value));
+					};
+					// return timestamps in milliseconds
+					resultMap.computeIfAbsent(timestamp * 1000, t -> new TreeMap<>()) //
+						.put(channelAddress, jsonValue);
 				}
 
 			} catch (Exception e) {

io.openems.edge.timedata.rrd4j/test/io/openems/edge/timedata/rrd4j/Rrd4jReadHandlerTest.java

@@ -163,6 +163,44 @@ public void testQueryHistoricDataWithResolution15minutes() throws Exception {
 		), this.query(new Resolution(15, ChronoUnit.MINUTES)));
 	}
 
+	@Test
+	public void testQueryResendDataReturnsAverageChannel() throws Exception {
+		// AVERAGE-consolidated channel (Unit.WATT maps to ConsolFun.AVERAGE; see
+		// Rrd4jSupplier.getDsDefForChannel). The previous adjustSeconds = arcStep - 300
+		// logic silently dropped these from the resend. This test asserts that the
+		// WATT samples written by setUp (0, 100, 200, ... every 5 min) are now returned
+		// by queryResendData over a window that spans several of those samples.
+		final var channelAddress = this.dummyComponent.channel(DummyComponent.ChannelId.DUMMY_CHANNEL).address();
+
+		final var result = this.readHandler.queryResendData(this.rrdbId, //
+				START.atZone(ZoneId.of("UTC")), //
+				START.plus(30, ChronoUnit.MINUTES).atZone(ZoneId.of("UTC")), //
+				Set.of(channelAddress), //
+				false);
+
+		// KEY assertion: the resend must NOT be empty (it was empty under the old code
+		// for AVERAGE channels).
+		assertTrue("queryResendData must return AVERAGE-consolidated WATT data", !result.isEmpty());
+
+		// Each AVERAGE 5-minute sample: 0, 100, 200, 300, 400, 500 at START + n*5min.
+		// Keys are timestamps in MILLISECONDS (see Rrd4jReadHandler.queryResendData).
+		for (int i = 0; i < 6; i++) {
+			final var expectedTimestampMillis = START.plus(5 * i, ChronoUnit.MINUTES).toEpochMilli();
+			final var row = result.get(expectedTimestampMillis);
+			assertTrue("Expected a resend row at " + expectedTimestampMillis + "ms", row != null);
+			assertEquals("Expected WATT value " + (i * 100) + " at sample " + i, //
+					new JsonPrimitive((double) (i * 100)), row.get(channelAddress));
+
+			// DUMMY_CHANNEL is OpenemsType.INTEGER, so the resend MUST emit an integer
+			// JSON (no decimal) to match the field type live data writes; otherwise the
+			// timeseries backend drops the whole point on a float-vs-integer conflict.
+			// gson's JsonPrimitive#equals treats 100 and 100.0 as equal, so the check
+			// above passes either way - assert the serialized form explicitly.
+			assertEquals("INTEGER channel must resend as an integer, not a float", //
+					String.valueOf(i * 100), row.get(channelAddress).toString());
+		}
+	}
+
 	@Test
 	public void testStreamRanges() throws Exception {
 		final var utc = ZoneId.of("UTC");