Gap filler WIP

src/main/java/com/github/zeekoe/bluebird/heatpump/Gapfiller.java

@@ -0,0 +1,57 @@
+package com.github.zeekoe.bluebird.heatpump;
+
+import com.github.zeekoe.bluebird.heatpump.model.HeatpumpLog;
+import com.github.zeekoe.bluebird.influx.PointMapper;
+import com.github.zeekoe.bluebird.influx.RealInfluxConnection;
+import org.influxdb.dto.Point;
+
+import java.io.IOException;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+
+public class Gapfiller {
+ public static void main(String[] args) throws IOException, InterruptedException {
+ final HeatpumpLog[] heatpumpLogs = new Heatpump().doHeatpumpGapRequest(2*60);
+ fillTheGaps(heatpumpLogs);
+ }
+ private static void fillTheGaps(HeatpumpLog[] heatpumplogs) {
+ if (heatpumplogs.length < 2) {
+ System.out.println("Please call this method for a bigger period");
+ return;
+ }
+ List<Point> pointsTodo = new ArrayList<>();
+ int doneCount = 0;
+
+ final ZonedDateTime oldestLog = Arrays.stream(heatpumplogs).min(Comparator.comparing(HeatpumpLog::getTimestamp))
+ .orElseThrow().getTimestamp().truncatedTo(ChronoUnit.SECONDS);
+ final ZonedDateTime newestLog = Arrays.stream(heatpumplogs).max(Comparator.comparing(HeatpumpLog::getTimestamp))
+ .orElseThrow().getTimestamp().truncatedTo(ChronoUnit.SECONDS);
+
+ final RealInfluxConnection influxConnection = new RealInfluxConnection();
+
+ final List<ZonedDateTime> influxedTimes = influxConnection.retrieveInfluxedTimesBetween(oldestLog, newestLog);
+
+ for (HeatpumpLog heatpumplog : heatpumplogs) {
+ final boolean alreadyInfluxed = influxedTimes.contains(heatpumplog.getTimestamp().truncatedTo(ChronoUnit.SECONDS));
+ System.out.println(heatpumplog.getTimestamp() + " " + (alreadyInfluxed ? "Y" : "N"));
+ if (!alreadyInfluxed) {
+ final Point point = PointMapper.map(heatpumplog);
+ pointsTodo.add(point);
+
+ } else {
+ doneCount++;
+ }
+ }
+ System.out.println("Todo: " + pointsTodo.size() + ", done: " + doneCount);
+ if (doneCount > 4) { // detect errors
+ for (Point point : pointsTodo) {
+ influxConnection.writePoint(point);
+ }
+ }
+ System.out.println("Done!");
+ }
+}

src/main/java/com/github/zeekoe/bluebird/heatpump/Heatpump.java

@@ -4,15 +4,17 @@
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import com.github.zeekoe.bluebird.heatpump.model.HeatpumpLog;
 import com.github.zeekoe.bluebird.influx.InfluxConnection;
+import com.github.zeekoe.bluebird.influx.PointMapper;
 import com.github.zeekoe.bluebird.influx.RealInfluxConnection;
 import com.github.zeekoe.bluebird.infrastructure.MyHttpClient;
 import org.influxdb.dto.Point;
 
 import java.io.IOException;
-import java.util.concurrent.TimeUnit;
+import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
 
 import static com.github.zeekoe.bluebird.infrastructure.BluebirdProperties.property;
-import static com.github.zeekoe.bluebird.infrastructure.BluebirdProperty.INFLUXDB_MEASUREMENT;
 import static com.github.zeekoe.bluebird.infrastructure.BluebirdProperty.WEHEAT_LOG_URL;
 
 public class Heatpump implements Runnable {
@@ -24,6 +26,7 @@ public class Heatpump implements Runnable {
  private final Auth auth;
  private final InfluxConnection influxConnection;
 
+ @SuppressWarnings("unused") // instantiated by Retryer
  public Heatpump() {
  auth = new Auth();
  influxConnection = new RealInfluxConnection();
@@ -37,11 +40,7 @@ public Heatpump(InfluxConnection influxConnection) {
  @Override
  public void run() {
  try {
- final String responseBody = doHeatpumpRequest();
-
- final String heatpumpLogString = responseBody.replace("[", "").replace("]", "");
-
- final HeatpumpLog heatpumpLog = OBJECT_MAPPER.readValue(heatpumpLogString, HeatpumpLog.class);
+ final HeatpumpLog heatpumpLog = doHeatpumpRequest();
  System.out.print(heatpumpLog.gettRoom() + " ");
  influx(heatpumpLog);
  } catch (IOException | InterruptedException e) {
@@ -50,41 +49,28 @@ public void run() {
  }
 
  private void influx(HeatpumpLog heatpumpLog) {
- final Point point = Point.measurement(property(INFLUXDB_MEASUREMENT))
- .time(heatpumpLog.getTimestamp().toEpochSecond(), TimeUnit.SECONDS)
- .addField("state", heatpumpLog.getState())
- .addField("t_1", heatpumpLog.getT1())
- .addField("t_2", heatpumpLog.getT2())
- .addField("fan_power", heatpumpLog.getFanPower())
- .addField("t_compressor_in", heatpumpLog.gettCompressorIn())
- .addField("t_compressor_in_transient", heatpumpLog.gettCompressorInTransient())
- .addField("t_compressor_out", heatpumpLog.gettCompressorOut())
- .addField("t_air_in", heatpumpLog.gettAirIn())
- .addField("t_air_out", heatpumpLog.gettAirOut())
- .addField("t_water_in", heatpumpLog.gettWaterIn())
- .addField("t_water_out", heatpumpLog.gettWaterOut())
- .addField("t_compressor_out_transient", heatpumpLog.getT_compressor_out_transient())
- .addField("p_compressor_in", heatpumpLog.getP_compressor_in())
- .addField("p_compressor_out", heatpumpLog.getP_compressor_out())
- .addField("rpm", heatpumpLog.getRpm())
- .addField("fan", heatpumpLog.getFan())
- .addField("t_inverter", heatpumpLog.getT_inverter())
- .addField("compressor_power_low_accuracy", heatpumpLog.getCompressor_power_low_accuracy())
- .addField("t_room", heatpumpLog.gettRoom())
- .addField("t_room_target", heatpumpLog.gettRoomTarget())
- .addField("t_thermostat_setpoint", heatpumpLog.gettThermostatSetpoint())
- .addField("cm_mass_power_in", heatpumpLog.getCmMassPowerIn())
- .addField("cm_mass_power_out", heatpumpLog.getCmMassPowerOut())
- .addField("t_water_house_in", heatpumpLog.gettWaterHouseIn())
- .addField("cm_mass_flow", heatpumpLog.getCm_mass_flow())
- .addField("ot_boiler_feed_temperature", heatpumpLog.getOt_boiler_feed_temperature())
- .addField("ot_boiler_return_temperature", heatpumpLog.getOt_boiler_return_temperature())
- .addField("error", heatpumpLog.getError())
- .build();
+ final Point point = PointMapper.map(heatpumpLog);
  influxConnection.writePoint(point);
  }
 
- private String doHeatpumpRequest() throws IOException, InterruptedException {
- return httpClient.get(property(WEHEAT_LOG_URL), auth.getToken());
+ private HeatpumpLog doHeatpumpRequest() throws IOException, InterruptedException {
+ final String responseBody = httpClient.get(property(WEHEAT_LOG_URL), auth.getToken());
+ return OBJECT_MAPPER.readValue(responseBody, HeatpumpLog.class);
+ }
+
+ public HeatpumpLog[] doHeatpumpGapRequest(int minutes) throws IOException, InterruptedException {
+ final ZonedDateTime now = ZonedDateTime.now();
+ final String from = formatDateTime(now.minusMinutes(minutes));
+ final String to = formatDateTime(now);
+
+ String url = property(WEHEAT_LOG_URL).replace("/latest","") + "/raw?startTime=" + from + "&endTime=" + to;
+ return OBJECT_MAPPER.readValue(httpClient.get(url, auth.getToken()), HeatpumpLog[].class);
+ }
+
+ public String formatDateTime(ZonedDateTime time) {
+ // TODO There must be a better way...
+ final ZonedDateTime utcTime = time.withZoneSameInstant(ZoneId.of("UTC"));
+ return utcTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd")) + "T"
+ + utcTime.format(DateTimeFormatter.ofPattern("HH:mm:ss")) + "Z";
  }
 }

src/main/java/com/github/zeekoe/bluebird/influx/PointMapper.java

@@ -0,0 +1,47 @@
+package com.github.zeekoe.bluebird.influx;
+
+import com.github.zeekoe.bluebird.heatpump.model.HeatpumpLog;
+import org.influxdb.dto.Point;
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.TimeUnit;
+
+import static com.github.zeekoe.bluebird.infrastructure.BluebirdProperties.property;
+import static com.github.zeekoe.bluebird.infrastructure.BluebirdProperty.INFLUXDB_MEASUREMENT;
+
+public class PointMapper {
+ @NotNull
+ public static Point map(HeatpumpLog heatpumpLog) {
+ return Point.measurement(property(INFLUXDB_MEASUREMENT))
+ .time(heatpumpLog.getTimestamp().toEpochSecond(), TimeUnit.SECONDS)
+ .addField("state", heatpumpLog.getState())
+ .addField("t_1", heatpumpLog.getT1())
+ .addField("t_2", heatpumpLog.getT2())
+ .addField("fan_power", heatpumpLog.getFanPower())
+ .addField("t_compressor_in", heatpumpLog.gettCompressorIn())
+ .addField("t_compressor_in_transient", heatpumpLog.gettCompressorInTransient())
+ .addField("t_compressor_out", heatpumpLog.gettCompressorOut())
+ .addField("t_air_in", heatpumpLog.gettAirIn())
+ .addField("t_air_out", heatpumpLog.gettAirOut())
+ .addField("t_water_in", heatpumpLog.gettWaterIn())
+ .addField("t_water_out", heatpumpLog.gettWaterOut())
+ .addField("t_compressor_out_transient", heatpumpLog.getT_compressor_out_transient())
+ .addField("p_compressor_in", heatpumpLog.getP_compressor_in())
+ .addField("p_compressor_out", heatpumpLog.getP_compressor_out())
+ .addField("rpm", heatpumpLog.getRpm())
+ .addField("fan", heatpumpLog.getFan())
+ .addField("t_inverter", heatpumpLog.getT_inverter())
+ .addField("compressor_power_low_accuracy", heatpumpLog.getCompressor_power_low_accuracy())
+ .addField("t_room", heatpumpLog.gettRoom())
+ .addField("t_room_target", heatpumpLog.gettRoomTarget())
+ .addField("t_thermostat_setpoint", heatpumpLog.gettThermostatSetpoint())
+ .addField("cm_mass_power_in", heatpumpLog.getCmMassPowerIn())
+ .addField("cm_mass_power_out", heatpumpLog.getCmMassPowerOut())
+ .addField("t_water_house_in", heatpumpLog.gettWaterHouseIn())
+ .addField("cm_mass_flow", heatpumpLog.getCm_mass_flow())
+ .addField("ot_boiler_feed_temperature", heatpumpLog.getOt_boiler_feed_temperature())
+ .addField("ot_boiler_return_temperature", heatpumpLog.getOt_boiler_return_temperature())
+ .addField("error", heatpumpLog.getError())
+ .build();
+ }
+}

src/main/java/com/github/zeekoe/bluebird/influx/RealInfluxConnection.java

@@ -3,6 +3,14 @@
 import org.influxdb.InfluxDB;
 import org.influxdb.InfluxDBFactory;
 import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+import org.influxdb.dto.QueryResult;
+
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoUnit;
+import java.util.List;
+import java.util.stream.Collectors;
 
 import static com.github.zeekoe.bluebird.infrastructure.BluebirdProperties.property;
 import static com.github.zeekoe.bluebird.infrastructure.BluebirdProperty.INFLUXDB_DATABASE;
@@ -25,7 +33,18 @@ public void writePoint(Point point) {
  influxDB.write(point);
  }
 
- public InfluxDB getInfluxDB() {
- return influxDB;
+ public List<ZonedDateTime> retrieveInfluxedTimesBetween(ZonedDateTime oldestLog, ZonedDateTime newestLog) {
+ final String queryString = "SELECT \"t_room\" FROM \"blackbird\" " +
+ " WHERE time >= '" + oldestLog.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + "'" +
+ " and time <= '" + newestLog.format(DateTimeFormatter.ISO_ZONED_DATE_TIME) + "';";
+ final QueryResult result = influxDB.query(new Query(queryString));
+ final List<QueryResult.Series> series1 = result.getResults().get(0).getSeries();
+ for (List<Object> series : series1.get(0).getValues()) {
+ System.out.println(series);
+ }
+ return series1.get(0).getValues()
+ .stream().map(s -> ZonedDateTime.parse(s.get(0).toString())
+ .truncatedTo(ChronoUnit.SECONDS))
+ .collect(Collectors.toList());
  }
 }