Added a JSON path based converter for JSON input data

measurement/translator/pom.xml

@@ -12,6 +12,10 @@
 
 	<name>INOA :: Measurement :: Telemetry Translator</name>
 
+	<properties>
+		<version.com.jayway.jsonpath>2.7.0</version.com.jayway.jsonpath>
+	</properties>
+
 	<dependencyManagement>
 		<dependencies>
 			<dependency>
@@ -87,6 +91,13 @@
 			<scope>runtime</scope>
 		</dependency>
 
+		<!-- Misc -->
+		<dependency>
+			<groupId>com.jayway.jsonpath</groupId>
+			<artifactId>json-path</artifactId>
+			<version>${version.com.jayway.jsonpath}</version>
+		</dependency>
+
 		<!-- test -->
 		<dependency>
 			<groupId>io.micronaut.test</groupId>

measurement/translator/src/main/java/io/inoa/measurement/translator/ApplicationProperties.java

@@ -33,5 +33,6 @@ public static class SensorProperties {
 		private boolean ignore = false;
 		private Optional<Double> modifier = Optional.empty();
 		private Map<String, String> ext = new HashMap<>();
+		private Map<String, Object> config = new HashMap<>();
 	}
 }

measurement/translator/src/main/java/io/inoa/measurement/translator/converter/common/CommonConverter.java

@@ -1,6 +1,7 @@
 package io.inoa.measurement.translator.converter.common;
 
 import java.util.HashMap;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.regex.Pattern;
@@ -9,6 +10,8 @@
 import io.inoa.measurement.translator.ApplicationProperties;
 import io.inoa.measurement.translator.ApplicationProperties.SensorProperties;
 import io.inoa.measurement.translator.converter.AbstractConverter;
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
 import lombok.RequiredArgsConstructor;
 
 /**
@@ -18,6 +21,8 @@
 public abstract class CommonConverter extends AbstractConverter {
 
 	private final ApplicationProperties properties;
+	private final MeterRegistry meterRegistry;
+	private final Map<String, Counter> counters = new HashMap<>();
 	private final String converter;
 
 	@Override
@@ -49,4 +54,8 @@ Optional<SensorProperties> get(String type, String sensor) {
 								&& Pattern.matches(sensorProperties.getNamePattern(), sensor))
 				.findFirst();
 	}
+
+	void increment(String type, String counter) {
+		counters.computeIfAbsent(type + counter, string -> meterRegistry.counter(counter, "type", type)).increment();
+	}
 }

measurement/translator/src/main/java/io/inoa/measurement/translator/converter/common/JsonConverter.java

@@ -7,6 +7,7 @@
 import io.inoa.fleet.telemetry.TelemetryRawVO;
 import io.inoa.measurement.telemetry.TelemetryVO;
 import io.inoa.measurement.translator.ApplicationProperties;
+import io.micrometer.core.instrument.MeterRegistry;
 import jakarta.inject.Singleton;
 
 /**
@@ -15,17 +16,17 @@
 @Singleton
 public class JsonConverter extends CommonConverter {
 
-	JsonConverter(ApplicationProperties properties) {
-		super(properties, "json");
+	JsonConverter(ApplicationProperties properties, MeterRegistry meterRegistry) {
+		super(properties, meterRegistry, "json");
 	}
 
 	@Override
 	public Stream<TelemetryVO> convert(TelemetryRawVO raw, String type, String sensor) {
-		return toJsonNode(raw.getValue()).stream().flatMap(e -> Stream
-				.iterate(e.fields(), Iterator::hasNext, UnaryOperator.identity()).map(Iterator::next))
+		return toJsonNode(raw.getValue()).stream()
+				.flatMap(e -> Stream.iterate(e.fields(), Iterator::hasNext, UnaryOperator.identity())
+						.map(Iterator::next))
 				.filter(node -> node.getValue().isNumber())
 				.map(node -> convert(type, sensor, node.getValue().asDouble())
-						.setUrn(raw.getUrn() + "." + node.getKey())
-						.setSensor(sensor + "." + node.getKey()));
+						.setUrn(raw.getUrn() + "." + node.getKey()).setSensor(sensor + "." + node.getKey()));
 	}
 }

measurement/translator/src/main/java/io/inoa/measurement/translator/converter/common/JsonPathConverter.java

@@ -0,0 +1,91 @@
+package io.inoa.measurement.translator.converter.common;
+
+import java.util.ArrayList;
+import java.util.NoSuchElementException;
+import java.util.Optional;
+import java.util.stream.Stream;
+
+import com.jayway.jsonpath.DocumentContext;
+import com.jayway.jsonpath.InvalidJsonException;
+import com.jayway.jsonpath.JsonPath;
+import com.jayway.jsonpath.JsonPathException;
+
+import io.inoa.fleet.telemetry.TelemetryRawVO;
+import io.inoa.measurement.telemetry.TelemetryVO;
+import io.inoa.measurement.translator.ApplicationProperties;
+import io.micrometer.core.instrument.MeterRegistry;
+import jakarta.inject.Singleton;
+import lombok.extern.slf4j.Slf4j;
+import net.minidev.json.JSONArray;
+
+/**
+ * This is a converter for JSON input strings based on JSON paths. One need to
+ * specify the datapoint to json path correlation as config map in the sensor
+ * entry of the application.yaml. Each path entry is mapped to a datapoint, so
+ * this converter is 1 sensor to N data points!
+ *
+ * {@see https://goessner.net/articles/JsonPath/}
+ */
+@Slf4j
+@Singleton
+public class JsonPathConverter extends CommonConverter {
+
+	public static final String COUNTER_FAIL_PARSE = "translator_json_path_fail_parse";
+	public static final String COUNTER_FAIL_PATH = "translator_json_path_fail_path";
+	public static final String COUNTER_FAIL_NAN = "translator_json_path_fail_nan";
+	public static final String COUNTER_SUCCESS = "translator_modbus_success";
+
+	JsonPathConverter(ApplicationProperties properties, MeterRegistry meterRegistry) {
+		super(properties, meterRegistry, "json-path");
+	}
+
+	@Override
+	public Stream<TelemetryVO> convert(TelemetryRawVO raw, String type, String sensor) {
+		Optional<ApplicationProperties.SensorProperties> properties = get(type, sensor);
+		if (!properties.isPresent()) {
+			// No config means no data
+			return Stream.empty();
+		}
+
+		var jsonInput = new String(raw.getValue());
+		DocumentContext jsonContext;
+
+		// Parse input JSON
+		try {
+			jsonContext = JsonPath.parse(jsonInput);
+		} catch (InvalidJsonException e) {
+			increment(type, COUNTER_FAIL_PARSE);
+			log.debug("Unable to parse invalid JSON '{}': {}", jsonInput, e.getMessage());
+			return Stream.empty();
+		}
+
+		// Iterate JSON paths and apply each of them
+		var result = new ArrayList<TelemetryVO>();
+		for (String datapoint : properties.get().getConfig().keySet()) {
+			try {
+				// Get JSON path expression from config entry and apply it to the input JSON
+				var value = jsonContext.read(properties.get().getConfig().get(datapoint).toString());
+				// Result can either be an array or a single data object
+				if (value instanceof JSONArray) {
+					// Simply take the first result (by convention)
+					result.add(convert(type, datapoint,
+							Double.parseDouble(((JSONArray) value).iterator().next().toString())));
+				} else {
+					result.add(convert(type, datapoint, Double.parseDouble(value.toString())));
+				}
+			} catch (ClassCastException | JsonPathException | NoSuchElementException e) {
+				increment(type, COUNTER_FAIL_PATH);
+				log.debug("Exception while applying JSON path '{}' to JSON '{}': {}",
+						properties.get().getConfig().get(datapoint).toString(), new String(raw.getValue()),
+						e.getMessage());
+			} catch (NumberFormatException e) {
+				increment(type, COUNTER_FAIL_NAN);
+				log.debug("JSON path '{}' on JSON '{}' is not a number: {}",
+						properties.get().getConfig().get(datapoint).toString(), new String(raw.getValue()),
+						e.getMessage());
+			}
+			increment(type, COUNTER_SUCCESS);
+		}
+		return result.stream();
+	}
+}

measurement/translator/src/main/java/io/inoa/measurement/translator/converter/common/ModbusConverter.java

@@ -1,15 +1,12 @@
 package io.inoa.measurement.translator.converter.common;
 
-import java.util.HashMap;
-import java.util.Map;
 import java.util.Set;
 import java.util.stream.Stream;
 
 import io.inoa.fleet.telemetry.TelemetryRawVO;
 import io.inoa.lib.modbus.CRC16;
 import io.inoa.measurement.telemetry.TelemetryVO;
 import io.inoa.measurement.translator.ApplicationProperties;
-import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.MeterRegistry;
 import jakarta.inject.Singleton;
 
@@ -36,15 +33,14 @@ public class ModbusConverter extends CommonConverter {
 	 */
 	private static final Set<Integer> FUNCTION_CODE_EXCEPTION = Set.of(129, 130, 131, 132, 133, 134, 143, 144);
 
-	/** Minimum length: 1 byte functionCode + 1 byte slaveId + 1 byte byteCount + 2 byte crc */
+	/**
+	 * Minimum length: 1 byte functionCode + 1 byte slaveId + 1 byte byteCount + 2
+	 * byte crc
+	 */
 	private static final Integer MESSAGE_MIN_LENGTH = 5;
 
-	private final MeterRegistry meterRegistry;
-	private final Map<String, Counter> counters = new HashMap<>();
-
 	ModbusConverter(ApplicationProperties properties, MeterRegistry meterRegistry) {
-		super(properties, "modbus");
-		this.meterRegistry = meterRegistry;
+		super(properties, meterRegistry, "modbus");
 	}
 
 	@Override
@@ -95,8 +91,8 @@ public Stream<TelemetryVO> convert(TelemetryRawVO raw, String type, String senso
 		var dataEndIndex = 2 * byteCount + 6;
 		int hexLength = hexString.length() - 4;
 		if (hexLength < dataEndIndex) {
-			log.info("Retrieved invalid modbus message (data.length {} < byteCount {}): {}",
-					(hexLength - 6) / 2, (dataEndIndex - 6) / 2, hexString);
+			log.info("Retrieved invalid modbus message (data.length {} < byteCount {}): {}", (hexLength - 6) / 2,
+					(dataEndIndex - 6) / 2, hexString);
 			increment(type, COUNTER_FAIL_BYTE_COUNT);
 			return Stream.empty();
 		}
@@ -107,8 +103,4 @@ public Stream<TelemetryVO> convert(TelemetryRawVO raw, String type, String senso
 
 		return Stream.of(convert(type, sensor, (double) value));
 	}
-
-	private void increment(String type, String counter) {
-		counters.computeIfAbsent(type + counter, string -> meterRegistry.counter(counter, "type", type)).increment();
-	}
 }

measurement/translator/src/main/java/io/inoa/measurement/translator/converter/common/NumberConverter.java

@@ -5,6 +5,7 @@
 import io.inoa.fleet.telemetry.TelemetryRawVO;
 import io.inoa.measurement.telemetry.TelemetryVO;
 import io.inoa.measurement.translator.ApplicationProperties;
+import io.micrometer.core.instrument.MeterRegistry;
 import jakarta.inject.Singleton;
 
 /**
@@ -13,8 +14,8 @@
 @Singleton
 public class NumberConverter extends CommonConverter {
 
-	NumberConverter(ApplicationProperties properties) {
-		super(properties, "number");
+	NumberConverter(ApplicationProperties properties, MeterRegistry meterRegistry) {
+		super(properties, meterRegistry, "number");
 	}
 
 	@Override

measurement/translator/src/main/resources/application.yaml

@@ -35,6 +35,15 @@ inoa.measurement.translator:
           converter: modbus
           modifier: 10
 
+    - name: shplg-s
+      sensors:
+        - name: status
+          converter: json-path
+          config:
+            power: "$.meters[:1].power"
+            work: "$.meters[:1].total"
+            temperature: "$.temperature"
+
     # source: https://www.dzg.de/fileadmin/dzg/content/downloads/produkte-zaehler/dvh4013/DZG_Modbus_Protocol.pdf
     - name: dvh4013
       sensors:

measurement/translator/src/test/java/io/inoa/measurement/translator/converter/common/JsonPathConverterTest.java

@@ -0,0 +1,37 @@
+package io.inoa.measurement.translator.converter.common;
+
+import java.time.Instant;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import io.inoa.fleet.telemetry.TelemetryRawVO;
+import io.inoa.measurement.translator.converter.AbstractConverterTest;
+import jakarta.inject.Inject;
+
+public class JsonPathConverterTest extends AbstractConverterTest {
+
+	@Inject
+	JsonPathConverter converter;
+
+	@DisplayName("success")
+	@Test
+	void success() {
+		var raw = new TelemetryRawVO().setUrn("urn:shplg-s:0A1B2C:status").setTimestamp(Instant.now().toEpochMilli())
+				.setValue(
+						"{\"wifi_sta\":{\"connected\":true,\"ssid\":\"landskron\",\"ip\":\"192.168.178.50\",\"rssi\":-63},\"cloud\":{\"enabled\":false,\"connected\":false},\"mqtt\":{\"connected\":false},\"time\":\"10:55\",\"serial\":1,\"has_update\":false,\"mac\":\"483FDA1D3A03\",\"relays\":[{\"ison\":true,\"has_timer\":false,\"overpower\":false}],\"meters\":[{\"power\":53.79,\"is_valid\":true,\"timestamp\":1647255320,\"counters\":[46.361, 0.000, 0.000],\"total\":46}],\"temperature\":21.18,\"overtemperature\":false,\"update\":{\"status\":\"unknown\",\"has_update\":false,\"new_version\":\"\",\"old_version\":\"20190516-073020/master@ea1b23db\"},\"ram_total\":50832,\"ram_free\":40540,\"fs_size\":233681,\"fs_free\":171684,\"uptime\":72}"
+								.getBytes());
+		var messages = convert(converter, raw, "shplg-s", "status");
+		assertCount(3, messages);
+	}
+
+	@DisplayName("invalid JSON input")
+	@Test
+	void invalidInput() {
+		var raw = new TelemetryRawVO().setUrn("urn:shplg-s:0A1B2C:status").setTimestamp(Instant.now().toEpochMilli())
+				.setValue("{\"wifi_sta\"::{\"connected\":true,\"ssid\":\"landskron\",\"ip\":\"192.168.178.50\","
+						.getBytes());
+		var messages = convert(converter, raw, "shplg-s", "status");
+		assertCount(0, messages);
+	}
+}