GridReceiver

src/main/java/de/ph87/data/demo/DemoService.java

@@ -18,6 +18,8 @@ import org.springframework.transaction.annotation.Transactional;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 
+import static de.ph87.data.grid.GridReceiver.GRID_DELIVERY_SERIES_NAME;
+import static de.ph87.data.grid.GridReceiver.GRID_PURCHASE_SERIES_NAME;
 import static de.ph87.data.oil.OilController.OIL_SERIES_NAME;
 import static de.ph87.data.photovoltaic.PhotovoltaicMqttReceiver.PHOTOVOLTAIC_ENERGY_SERIES_NAME;
 
@@ -39,6 +41,8 @@ public class DemoService {
   @EventListener(ApplicationStartedEvent.class)
   public void startup() {
     series(PHOTOVOLTAIC_ENERGY_SERIES_NAME, SeriesMode.INCREASING);
+    series(GRID_PURCHASE_SERIES_NAME, SeriesMode.INCREASING);
+    series(GRID_DELIVERY_SERIES_NAME, SeriesMode.INCREASING);
 
     final Series oil = series(OIL_SERIES_NAME, SeriesMode.DECREASING);
     oil.setPeriod(null);

src/main/java/de/ph87/data/grid/GridInbound.java

@@ -0,0 +1,31 @@
+package de.ph87.data.grid;
+
+import lombok.Getter;
+import lombok.ToString;
+
+import java.time.ZonedDateTime;
+
+import static de.ph87.data.series.period.consumption.ConsumptionController.ZDT;
+
+@Getter
+@ToString
+public class GridInbound {
+
+  public final String meter = "1ZPA0020300305"; // TODO implement into SmartMeter firmware
+
+  public final ZonedDateTime date;
+
+  public final double purchaseWh;
+
+  public final double deliveryWh;
+
+  public final double powerW;
+
+  public GridInbound(final long timestamp, final double purchaseWh, final double deliveryWh, final double powerW) {
+    this.date = ZDT(timestamp);
+    this.purchaseWh = purchaseWh;
+    this.deliveryWh = deliveryWh;
+    this.powerW = powerW;
+  }
+
+}

src/main/java/de/ph87/data/grid/GridReceiver.java

@@ -0,0 +1,71 @@
+package de.ph87.data.grid;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import de.ph87.data.mqtt.MqttEvent;
+import de.ph87.data.series.measure.MeasureEvent;
+import lombok.NonNull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Service;
+
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class GridReceiver {
+
+  public static final String GRID_PURCHASE_SERIES_NAME = "grid.purchaseKWh";
+
+  public static final String GRID_DELIVERY_SERIES_NAME = "grid.deliveryKWh";
+
+  private final ApplicationEventPublisher applicationEventPublisher;
+
+  private final ObjectMapper objectMapper;
+
+  @EventListener(MqttEvent.class)
+  public void onEvent(@NonNull final MqttEvent event) {
+    if (!event.topic.equals("electricity/grid/json")) {
+      return;
+    }
+
+    final GridInbound inbound;
+    try {
+      inbound = objectMapper.readValue(event.payload, GridInbound.class);
+    } catch (JsonProcessingException e) {
+      log.error("Failed to parse GridInbound json: event={}, error={}", event, e.toString());
+      return;
+    }
+
+    final double purchaseKWh = energyToKWh(inbound.purchaseWh, "Wh");
+    applicationEventPublisher.publishEvent(new MeasureEvent(GRID_PURCHASE_SERIES_NAME, inbound.meter, inbound.date, purchaseKWh));
+
+    final double deliveryKWh = energyToKWh(inbound.deliveryWh, "Wh");
+    applicationEventPublisher.publishEvent(new MeasureEvent(GRID_DELIVERY_SERIES_NAME, inbound.meter, inbound.date, deliveryKWh));
+  }
+
+  public static double energyToKWh(final double energy, final String energyUnit) {
+    return switch (energyUnit) {
+      case "mWh" -> energy / 1000000;
+      case "Wh" -> energy / 1000;
+      case "kWh" -> energy;
+      case "MWh" -> energy * 1000;
+      case "GWh" -> energy * 1000000;
+      default -> throw new RuntimeException();
+    };
+  }
+
+  @SuppressWarnings("unused")
+  public static double powerToW(final double power, final String powerUnit) {
+    return switch (powerUnit) {
+      case "mW" -> power / 1000;
+      case "W" -> power;
+      case "kW" -> power * 1000;
+      case "MW" -> power * 1000000;
+      case "GW" -> power * 1000000000;
+      default -> throw new RuntimeException();
+    };
+  }
+
+}

src/main/java/de/ph87/data/mqtt/MqttService.java

@@ -32,7 +32,7 @@ public class MqttService {
     options.setCleanSession(config.isCleanSession());
     options.setConnectionTimeout(config.getConnectTimeoutSec());
     client.connect(options);
-    client.subscribe("#", this::safeReceive);
+    client.subscribe(config.getTopic(), this::safeReceive);
     client.setCallback(new Callback());
   }
 

src/main/java/de/ph87/data/series/SeriesService.java

@@ -27,7 +27,7 @@ public class SeriesService {
     log.debug("Handling MeasureEvent: {}", event);
     final Optional<Series> seriesOptional = seriesRepository.findByNameOrAliasesContains(event.getName(), event.getName());
     if (seriesOptional.isEmpty()) {
-      log.debug("No series found with name or alias: \"{}\"", event.getName());
+      log.warn("No series found with name or alias: \"{}\"", event.getName());
       return;
     }