patrick/Data
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Value + EspHomeHandler + SmartMeterHandler

Browse Source
This commit is contained in:
Patrick Haßel 2025-02-15 18:41:19 +01:00
parent 147bff7927
commit 8f727572fa
21 changed files with 407 additions and 203 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
application.properties
View File

@ -1,11 +1,11 @@
logging.level.de.ph87=DEBUG
logging.level.de.ph87.data.series=DEBUG
#-
spring.datasource.url=jdbc:h2:./database;AUTO_SERVER=TRUE
spring.datasource.driverClassName=org.h2.Driver
spring.datasource.username=sa
spring.datasource.password=password
#-
#spring.jpa.hibernate.ddl-auto=create
spring.jpa.hibernate.ddl-auto=create
#-
de.ph87.data.message.receive.mqtt.host=10.0.0.50
de.ph87.data.message.receive.mqtt.topic=#

2
src/main/java/de/ph87/data/series/Action.java → src/main/java/de/ph87/data/Action.java
View File

@ -1,4 +1,4 @@
package de.ph87.data.series;
package de.ph87.data;
public enum Action {
CREATED, CHANGED, DELETED

4
src/main/java/de/ph87/data/Backend.java
View File

@ -1,7 +1,7 @@
package de.ph87.data;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.*;
import org.springframework.boot.autoconfigure.*;
@SpringBootApplication
public class Backend {

4
src/main/java/de/ph87/data/message/IMessageHandler.java
View File

@ -1,9 +1,9 @@
package de.ph87.data.message;
import lombok.NonNull;
import lombok.*;
public interface IMessageHandler {
void handle(@NonNull final Message message);
void handle(@NonNull final Message message) throws Exception;
}

8
src/main/java/de/ph87/data/message/Message.java
View File

@ -1,18 +1,14 @@
package de.ph87.data.message;
import lombok.Getter;
import lombok.ToString;
import java.time.ZonedDateTime;
import lombok.*;
@Getter
@ToString
public class Message {
public final ZonedDateTime date = ZonedDateTime.now();
public final String topic;
@ToString.Exclude
public final String payload;
public final String payloadLoggable;

13
src/main/java/de/ph87/data/message/MessageService.java
View File

@ -1,11 +1,10 @@
package de.ph87.data.message;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import lombok.*;
import lombok.extern.slf4j.*;
import org.springframework.stereotype.*;
import java.util.List;
import java.util.*;
@Slf4j
@Service
@ -14,12 +13,12 @@ public class MessageService {
private final List<IMessageHandler> messageHandlers;
public void onMessage(@NonNull final Message message) {
public void handle(@NonNull final Message message) {
messageHandlers.forEach(handler -> {
try {
handler.handle(message);
} catch (Exception e) {
log.warn("Message handler error: message={}", e.getMessage(), e);
log.error("Failed handling message: message={}, error={}", message, e.toString());
}
});
}

92
src/main/java/de/ph87/data/message/handler/EspHomeHandler.java Normal file
View File

@ -0,0 +1,92 @@
package de.ph87.data.message.handler;
import com.fasterxml.jackson.annotation.*;
import com.fasterxml.jackson.databind.*;
import com.fasterxml.jackson.databind.annotation.*;
import de.ph87.data.message.*;
import de.ph87.data.series.*;
import de.ph87.data.unit.Value;
import de.ph87.data.unit.*;
import lombok.*;
import lombok.extern.slf4j.*;
import org.springframework.stereotype.*;
import java.time.*;
import java.util.*;
import java.util.regex.*;
import java.util.stream.*;
@Slf4j
@Service
@RequiredArgsConstructor
public class EspHomeHandler implements IMessageHandler {
private static final Pattern REGEX = Pattern.compile("^(?<area>\\w+)/sensor/(?<property>\\w+)$");
private final ObjectMapper objectMapper;
private final SeriesService seriesService;
@Override
public void handle(@NonNull final Message message) throws Exception {
final Matcher matcher = REGEX.matcher(message.topic);
if (!matcher.find()) {
return;
}
final String area = matcher.group("area");
final String property = propertyWorkaround(matcher.group("property"));
final String name = "%s/%s".formatted(area, property.replace("_", "/"));
final Unit targetUnit = switch (property) {
case "iaq" -> Unit.IAQ;
case "iaq_co2_equivalent" -> Unit.IAQ_CO2_EQUIVALENT;
case "iaq_voc_equivalent" -> Unit.IAQ_VOC_EQUIVALENT;
case "pressure" -> Unit.PRESSURE_HPA;
case "temperature" -> Unit.TEMPERATURE_C;
case "humidity_relative" -> Unit.HUMIDITY_RELATIVE_PERCENT;
case "humidity_absolute" -> Unit.HUMIDITY_ABSOLUTE_GM3;
case "sun" -> Unit.SUN_DC;
default -> null;
};
if (targetUnit == null) {
log.debug("Skipping property: {}", name);
return;
}
final Inbound inbound = objectMapper.readValue(message.payload, Inbound.class);
final Unit unitFromPayload = inbound.units.stream().filter(ufp -> ufp.base == targetUnit.base).findFirst().orElse(null);
if (unitFromPayload == null) {
log.error("Unit mismatch: fromTopic={}, fromPayload=[{}]", targetUnit, inbound.getUnits().stream().map(Enum::name).collect(Collectors.joining(",")));
return;
}
final Value value = new Value(inbound.value, unitFromPayload);
seriesService.receive(new SeriesInbound(name, inbound.date, value.as(targetUnit)));
}
private String propertyWorkaround(final String property) {
if ("iaq_equivalent_voc".equals(property)) {
return "iaq_voc_equivalent";
}
return property;
}
@Getter
@ToString
private static class Inbound {
@NonNull
public final ZonedDateTime date;
public final double value;
@NonNull
@JsonDeserialize(using = UnitListDeserializer.class)
public final List<Unit> units;
public Inbound(final long timestamp, final double value, @JsonProperty("unit") @NonNull final List<Unit> units) {
this.date = ZonedDateTime.ofInstant(Instant.ofEpochSecond(timestamp), ZoneId.systemDefault());
this.value = value;
this.units = units;
}
}
}

43
src/main/java/de/ph87/data/message/handler/SimpleJsonHandler.java
View File

@ -1,22 +1,14 @@
package de.ph87.data.message.handler;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import de.ph87.data.message.IMessageHandler;
import de.ph87.data.message.Message;
import de.ph87.data.series.SeriesInbound;
import de.ph87.data.series.SeriesService;
import de.ph87.data.unit.Unit;
import lombok.Getter;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import com.fasterxml.jackson.databind.*;
import de.ph87.data.message.*;
import de.ph87.data.series.*;
import de.ph87.data.unit.*;
import lombok.*;
import lombok.extern.slf4j.*;
import org.springframework.stereotype.*;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.*;
@Slf4j
@Service
@ -28,26 +20,31 @@ public class SimpleJsonHandler implements IMessageHandler {
private final SeriesService seriesService;
@Override
public void handle(@NonNull final Message message) {
try {
final Inbound inbound = objectMapper.readValue(message.payload, Inbound.class);
seriesService.receive(new SeriesInbound(message.topic, inbound.date, inbound.value, inbound.unit));
} catch (JsonProcessingException e) {
log.debug("Failed to parse inbound message: topic={}, message={}, error={}", message.topic, message, e.toString());
public void handle(@NonNull final Message message) throws Exception {
if (!message.topic.endsWith("/SimpleJson")) {
return;
}
final Inbound inbound = objectMapper.readValue(message.payload, Inbound.class);
seriesService.receive(new SeriesInbound(inbound.name, inbound.date, inbound.value, inbound.unit));
}
@Getter
@ToString
private static class Inbound {
@NonNull
public final String name;
@NonNull
public final ZonedDateTime date;
public final double value;
@NonNull
public final Unit unit;
public Inbound(final long timestamp, final double value, final Unit unit) {
public Inbound(@NonNull final String name, final long timestamp, final double value, @NonNull final Unit unit) {
this.name = name;
this.date = ZonedDateTime.ofInstant(Instant.ofEpochSecond(timestamp), ZoneId.systemDefault());
this.value = value;
this.unit = unit;

56
src/main/java/de/ph87/data/message/handler/SmartMeterHandler.java Normal file
View File

@ -0,0 +1,56 @@
package de.ph87.data.message.handler;
import com.fasterxml.jackson.databind.*;
import de.ph87.data.message.*;
import de.ph87.data.series.*;
import de.ph87.data.unit.Value;
import de.ph87.data.unit.*;
import lombok.*;
import lombok.extern.slf4j.*;
import org.springframework.stereotype.*;
import java.time.*;
@Slf4j
@Service
@RequiredArgsConstructor
public class SmartMeterHandler implements IMessageHandler {
private final ObjectMapper objectMapper;
private final SeriesService seriesService;
@Override
public void handle(@NonNull final Message message) throws Exception {
if (!"electricity/grid/json".equals(message.topic)) {
return;
}
final Inbound inbound = objectMapper.readValue(message.payload, Inbound.class);
seriesService.receive(new SeriesInbound("energy/purchased", inbound.date, inbound.energyPurchased));
seriesService.receive(new SeriesInbound("energy/delivered", inbound.date, inbound.energyDelivered));
seriesService.receive(new SeriesInbound("power/difference", inbound.date, inbound.powerDifference));
}
@Getter
@ToString
private static class Inbound {
@NonNull
public final ZonedDateTime date;
public final Value energyPurchased;
public final Value energyDelivered;
public final Value powerDifference;
public Inbound(final long timestamp, final double purchaseWh, final double deliveryWh, final double powerW) {
this.date = ZonedDateTime.ofInstant(Instant.ofEpochSecond(timestamp), ZoneId.systemDefault());
this.energyPurchased = new Value(purchaseWh, Unit.ENERGY_WH);
this.energyDelivered = new Value(deliveryWh, Unit.ENERGY_WH);
this.powerDifference = new Value(powerW, Unit.POWER_W);
}
}
}

8
src/main/java/de/ph87/data/message/receive/mqtt/MqttReceiver.java
View File

@ -1,5 +1,6 @@
package de.ph87.data.message.receive.mqtt;
import de.ph87.data.message.*;
import jakarta.annotation.*;
import lombok.*;
import lombok.extern.slf4j.*;
@ -19,6 +20,8 @@ public class MqttReceiver {
private final MqttConfig config;
private final MessageService messageService;
@Nullable
private MqttClient client;
@ -108,8 +111,9 @@ public class MqttReceiver {
private void _receive(final String topic, final MqttMessage mqttMessage) {
Thread.currentThread().setName("MQTT-RECEIVE");
final String payload = new String(mqttMessage.getPayload(), StandardCharsets.UTF_8);
final String payloadLoggable = payload.replace("\n", "\\n").replace("\r", "\\r");
log.debug("received: topic={}, message={}", topic, payloadLoggable);
final Message message = new Message(topic, payload);
log.debug("received: {}", message);
messageService.handle(message);
}
}

21
src/main/java/de/ph87/data/series/Series.java
View File

@ -1,11 +1,9 @@
package de.ph87.data.series;
import de.ph87.data.unit.Unit;
import de.ph87.data.unit.*;
import jakarta.persistence.*;
import lombok.*;
import java.util.UUID;
@Entity
@Getter
@ToString
@ -13,7 +11,13 @@ import java.util.UUID;
public class Series {
@Id
private String uuid = UUID.randomUUID().toString();
@GeneratedValue(strategy = GenerationType.IDENTITY)
private long id;
@Setter
@NonNull
@Column(nullable = false, unique = true)
private String name;
@Setter
@NonNull
@ -26,9 +30,10 @@ public class Series {
@Enumerated(EnumType.STRING)
private Unit unit;
@Setter
@NonNull
@Column(nullable = false, unique = true)
private String source;
public Series(@NonNull final String name, @NonNull final Unit unit) {
this.name = name;
this.title = name;
this.unit = unit;
}
}

16
src/main/java/de/ph87/data/series/SeriesDto.java
View File

@ -1,27 +1,25 @@
package de.ph87.data.series;
import de.ph87.data.unit.Unit;
import lombok.Getter;
import lombok.NonNull;
import lombok.ToString;
import de.ph87.data.unit.*;
import lombok.*;
@Getter
@ToString
public class SeriesDto {
public final String uuid;
public final long id;
public final String name;
public final String title;
public final Unit unit;
public final String source;
public SeriesDto(@NonNull final Series series) {
this.uuid = series.getUuid();
this.id = series.getId();
this.name = series.getName();
this.title = series.getTitle();
this.unit = series.getUnit();
this.source = series.getSource();
}
}

21
src/main/java/de/ph87/data/series/SeriesInbound.java
View File

@ -1,11 +1,10 @@
package de.ph87.data.series;
import de.ph87.data.unit.Unit;
import lombok.Getter;
import lombok.NonNull;
import lombok.ToString;
import de.ph87.data.unit.Value;
import de.ph87.data.unit.*;
import lombok.*;
import java.time.ZonedDateTime;
import java.time.*;
@Getter
@ToString
@ -17,16 +16,18 @@ public class SeriesInbound {
@NonNull
public final ZonedDateTime date;
public final double value;
public final Value value;
@NonNull
public final Unit unit;
public SeriesInbound(@NonNull final String name, @NonNull final ZonedDateTime date, final Value value) {
this.name = name;
this.date = date;
this.value = value;
}
public SeriesInbound(@NonNull final String name, @NonNull final ZonedDateTime date, final double value, @NonNull final Unit unit) {
this.name = name;
this.date = date;
this.value = value;
this.unit = unit;
this.value = new Value(value, unit);
}
}

9
src/main/java/de/ph87/data/series/SeriesRepository.java
View File

@ -1,7 +1,12 @@
package de.ph87.data.series;
import org.springframework.data.repository.ListCrudRepository;
import lombok.*;
import org.springframework.data.repository.*;
public interface SeriesRepository extends ListCrudRepository<Series, String> {
import java.util.*;
public interface SeriesRepository extends ListCrudRepository<Series, Long> {
Optional<Series> findByName(@NonNull String name);
}

71
src/main/java/de/ph87/data/series/SeriesService.java
View File

@ -1,16 +1,14 @@
package de.ph87.data.series;
import de.ph87.data.series.entry.EntryService;
import jakarta.annotation.PostConstruct;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import de.ph87.data.*;
import de.ph87.data.series.entry.*;
import de.ph87.data.unit.*;
import lombok.*;
import lombok.extern.slf4j.*;
import org.springframework.stereotype.*;
import org.springframework.transaction.annotation.*;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.*;
@Slf4j
@Service
@ -18,45 +16,22 @@ import java.util.function.Consumer;
@RequiredArgsConstructor
public class SeriesService {
private final Map<String, String> sources = new HashMap<>();
private final SeriesRepository seriesRepository;
private final EntryService entryService;
@PostConstruct
public void cacheInit() {
synchronized (sources) {
seriesRepository.findAll().forEach(this::cachePut);
}
}
private void cachePut(@NonNull final Series series) {
synchronized (sources) {
sources.put(series.getUuid(), series.getSource());
}
}
private void cacheRemove(final Series series) {
synchronized (sources) {
sources.remove(series.getUuid());
}
}
public SeriesDto modify(@NonNull final String uuid, @NonNull final Consumer<Series> modifier) {
final Series series = getByUuid(uuid);
public SeriesDto modify(@NonNull final long id, @NonNull final Consumer<Series> modifier) {
final Series series = getById(id);
modifier.accept(series);
cachePut(series);
return publish(series, Action.CHANGED);
}
public void delete(@NonNull final String uuid) {
final Series series = getByUuid(uuid);
cacheRemove(series);
public void delete(@NonNull final long id) {
final Series series = getById(id);
}
private Series getByUuid(@NonNull final String uuid) {
return seriesRepository.findById(uuid).orElseThrow();
private Series getById(@NonNull final long id) {
return seriesRepository.findById(id).orElseThrow();
}
private SeriesDto publish(@NonNull final Series series, @NonNull final Action action) {
@ -69,13 +44,19 @@ public class SeriesService {
return new SeriesDto(series);
}
public void receive(@NonNull final SeriesInbound measure) {
final String uuid = sources.get(measure.name);
if (uuid == null) {
return;
}
final Series series = getByUuid(uuid);
public void receive(@NonNull final SeriesInbound measure) throws Unit.NotConvertible {
final Series series = getOrCreate(measure.name, measure.value.unit);
entryService.write(series, measure);
}
private Series getOrCreate(@NonNull final String name, @NonNull final Unit unit) {
return seriesRepository
.findByName(name)
.orElseGet(() -> {
final Series series = seriesRepository.save(new Series(name, unit));
publish(series, Action.CREATED);
return series;
});
}
}

25
src/main/java/de/ph87/data/series/entry/Entry.java
View File

@ -1,19 +1,23 @@
package de.ph87.data.series.entry;
import de.ph87.data.series.Series;
import de.ph87.data.series.*;
import de.ph87.data.unit.Value;
import de.ph87.data.unit.*;
import jakarta.persistence.*;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.ToString;
import lombok.*;
import java.time.ZonedDateTime;
import java.time.temporal.ChronoUnit;
import java.time.*;
import java.time.temporal.*;
@Entity
@Getter
@ToString
@NoArgsConstructor
@Table(
uniqueConstraints = {
@UniqueConstraint(columnNames = {"series_id", "date"})
}
)
public class Entry {
@Id
@ -28,13 +32,14 @@ public class Entry {
@Column(nullable = false)
private ZonedDateTime date;
@Column(nullable = false,name = "`value`")
@Setter
@Column(nullable = false, name = "`value`")
private double value;
public Entry(@NonNull final Series series, @NonNull final ZonedDateTime date, final double value) {
public Entry(@NonNull final Series series, @NonNull final ZonedDateTime date, final Value value) throws Unit.NotConvertible {
this.series = series;
this.date = date.truncatedTo(ChronoUnit.MINUTES);
this.value = value;
this.value = value.as(series.getUnit()).value;
}
}

9
src/main/java/de/ph87/data/series/entry/EntryRepository.java
View File

@ -1,7 +1,14 @@
package de.ph87.data.series.entry;
import org.springframework.data.repository.ListCrudRepository;
import de.ph87.data.series.*;
import lombok.*;
import org.springframework.data.repository.*;
import java.time.*;
import java.util.*;
public interface EntryRepository extends ListCrudRepository<Entry, Long> {
Optional<Entry> findBySeriesAndDate(@NonNull Series series, @NonNull ZonedDateTime truncated);
}

24
src/main/java/de/ph87/data/series/entry/EntryService.java
View File

@ -1,12 +1,14 @@
package de.ph87.data.series.entry;
import de.ph87.data.series.Series;
import de.ph87.data.series.SeriesInbound;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import de.ph87.data.series.*;
import de.ph87.data.unit.*;
import lombok.*;
import lombok.extern.slf4j.*;
import org.springframework.stereotype.*;
import org.springframework.transaction.annotation.*;
import java.time.*;
import java.time.temporal.*;
@Slf4j
@Service
@ -16,8 +18,12 @@ public class EntryService {
private final EntryRepository entryRepository;
public void write(@NonNull final Series series, @NonNull final SeriesInbound measure) {
// TODO
public void write(@NonNull final Series series, @NonNull final SeriesInbound measure) throws Unit.NotConvertible {
final ZonedDateTime truncated = measure.date.truncatedTo(ChronoUnit.MINUTES);
if (entryRepository.findBySeriesAndDate(series, truncated).isEmpty()) {
final Entry created = entryRepository.save(new Entry(series, truncated, measure.value));
log.debug("Created: {}", created);
}
}
}

16
src/main/java/de/ph87/data/unit/Unit.java
View File

@ -1,6 +1,6 @@
package de.ph87.data.unit;
import lombok.NonNull;
import lombok.*;
public enum Unit {
TEMPERATURE_C("°C"),
@ -15,6 +15,10 @@ public enum Unit {
POWER_KW("kW", 1000, POWER_W),
ENERGY_WH("W"),
ENERGY_KWH("kWh", 1000, ENERGY_WH),
IAQ("IAQ"),
IAQ_CO2_EQUIVALENT("ppm"),
IAQ_VOC_EQUIVALENT("ppm"),
SUN_DC("Δ°C"),
;
public final String unit;
@ -29,10 +33,18 @@ public enum Unit {
this.base = this;
}
Unit(@NonNull final String unit, double factor, @NonNull Unit base) {
Unit(@NonNull final String unit, final double factor, @NonNull final Unit base) {
this.unit = unit;
this.factor = factor;
this.base = base;
}
public static class NotConvertible extends Exception {
public NotConvertible(final @NonNull Unit source, final Unit target) {
super("Cannot convert Units: source=%s, target=%s".formatted(source, target));
}
}
}

17
src/main/java/de/ph87/data/unit/UnitListDeserializer.java Normal file
View File

@ -0,0 +1,17 @@
package de.ph87.data.unit;
import com.fasterxml.jackson.core.*;
import com.fasterxml.jackson.databind.*;
import java.io.*;
import java.util.*;
public class UnitListDeserializer extends JsonDeserializer<List<Unit>> {
@Override
public List<Unit> deserialize(final JsonParser jsonParser, final DeserializationContext deserializationContext) throws IOException {
final String name = jsonParser.getValueAsString();
return Arrays.stream(Unit.values()).filter(unit -> unit.unit.equals(name)).toList();
}
}

23
src/main/java/de/ph87/data/unit/Value.java Normal file
View File

@ -0,0 +1,23 @@
package de.ph87.data.unit;
import lombok.*;
public class Value {
public final double value;
public final Unit unit;
public Value(final double value, @NonNull final Unit unit) {
this.value = value;
this.unit = unit;
}
public Value as(@NonNull final Unit target) throws Unit.NotConvertible {
if (this.unit.base != target.base) {
throw new Unit.NotConvertible(this.unit, target);
}
return new Value(value * this.unit.factor / target.factor, target);
}
}