wn/MeasurementCollector
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

DataParsers loaded from configuration, DataObjectQueue introduced

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2017-11-16 11:45:50 +01:00
parent 9b8df1b119
commit bc8482b03c
23 changed files with 127 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/de/hottis/MeasurementCollector/ADataParser.java
View File

@ -5,6 +5,12 @@ import java.util.List;
import org.openmuc.jmbus.DataRecord;
public interface ADataParser {
public List<ADataObject> parse(LocalDateTime timestamp, String name, List<DataRecord> dataRecords);
public abstract class ADataParser {
protected String name;
public ADataParser(String name) {
this.name = name;
}
abstract public List<ADataObject> parse(LocalDateTime timestamp, List<DataRecord> dataRecords);
}

9
src/de/hottis/MeasurementCollector/AMessageParser.java
View File

@ -1,15 +1,16 @@
package de.hottis.MeasurementCollector;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.Properties;
public abstract class AMessageParser {
private String topic;
protected ConcurrentLinkedQueue<List<ADataObject>> queue;
protected DataObjectQueue queue;
protected Properties config;
public AMessageParser(String topic, ConcurrentLinkedQueue<List<ADataObject>> queue) {
public AMessageParser(String topic, Properties config, DataObjectQueue queue) {
this.topic = topic;
this.config = config;
this.queue = queue;
}

11
src/de/hottis/MeasurementCollector/DataObjectQueue.java Normal file
View File

@ -0,0 +1,11 @@
package de.hottis.MeasurementCollector;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
@SuppressWarnings("serial")
public class DataObjectQueue extends ConcurrentLinkedQueue<List<ADataObject>> {
public DataObjectQueue() {
super();
}
}

8
src/de/hottis/MeasurementCollector/FinderOnePhasePowerMeter.java
View File

@ -6,8 +6,12 @@ import java.util.List;
import org.openmuc.jmbus.DataRecord;
public class FinderOnePhasePowerMeter implements ADataParser {
public List<ADataObject> parse(LocalDateTime timestamp, String name, List<DataRecord> dataRecords) {
public class FinderOnePhasePowerMeter extends ADataParser {
public FinderOnePhasePowerMeter(String name) {
super(name);
}
public List<ADataObject> parse(LocalDateTime timestamp, List<DataRecord> dataRecords) {
ArrayList<ADataObject> list = new ArrayList<ADataObject>();
ElectricEnergyDataObject tdo = new ElectricEnergyDataObject(timestamp, name,

8
src/de/hottis/MeasurementCollector/FinderThreePhasePowerMeter.java
View File

@ -6,8 +6,12 @@ import java.util.List;
import org.openmuc.jmbus.DataRecord;
public class FinderThreePhasePowerMeter implements ADataParser {
public List<ADataObject> parse(LocalDateTime timestamp, String name, List<DataRecord> dataRecords) {
public class FinderThreePhasePowerMeter extends ADataParser {
public FinderThreePhasePowerMeter(String name) {
super(name);
}
public List<ADataObject> parse(LocalDateTime timestamp, List<DataRecord> dataRecords) {
ArrayList<ADataObject> list = new ArrayList<ADataObject>();
ElectricEnergyDataObject tdo = new ElectricEnergyDataObject(timestamp, name,

8
src/de/hottis/MeasurementCollector/HottisFourChannelThermometer.java
View File

@ -6,8 +6,12 @@ import java.util.List;
import org.openmuc.jmbus.DataRecord;
public class HottisFourChannelThermometer implements ADataParser {
public List<ADataObject> parse(LocalDateTime timestamp, String name, List<DataRecord> dataRecords) {
public class HottisFourChannelThermometer extends ADataParser {
public HottisFourChannelThermometer(String name) {
super(name);
}
public List<ADataObject> parse(LocalDateTime timestamp, List<DataRecord> dataRecords) {
ArrayList<ADataObject> list = new ArrayList<ADataObject>();
TemperatureDataObject tdo = new TemperatureDataObject(timestamp, "Hedge", dataRecords.get(5).getScaledDataValue());

94
src/de/hottis/MeasurementCollector/MBusParser.java
View File

@ -1,9 +1,12 @@
package de.hottis.MeasurementCollector;
import java.lang.reflect.Constructor;
import java.time.LocalDateTime;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.Properties;
import org.openmuc.jmbus.DataRecord;
import org.openmuc.jmbus.MBusMessage;
@ -13,71 +16,80 @@ import com.json.parsers.JSONParser;
import com.json.parsers.JsonParserFactory;
public class MBusParser extends AMessageParser {
static final String DATA_PARSER_PROP = "mbus.dataparser";
static final String TOPIC = "IoT/Measurement/MeterbusHub";
private final JSONParser jsonParser;
private final HottisFourChannelThermometer hottisFourChannelThermometer;
private final FinderOnePhasePowerMeter finderOnePhasePowerMeter;
private final FinderThreePhasePowerMeter finderThreePhasePowerMeter;
public MBusParser(ConcurrentLinkedQueue<List<ADataObject>> queue) {
super(TOPIC, queue);
private HashMap<String, ADataParser> dataParsers;
public MBusParser(Properties config, DataObjectQueue queue) {
super(TOPIC, config, queue);
JsonParserFactory jsonParserFactory = JsonParserFactory.getInstance();
jsonParser = jsonParserFactory.newJsonParser();
hottisFourChannelThermometer = new HottisFourChannelThermometer();
finderOnePhasePowerMeter = new FinderOnePhasePowerMeter();
finderThreePhasePowerMeter = new FinderThreePhasePowerMeter();
dataParsers = new HashMap<String, ADataParser>();
}
public void registerConfiguredDataParsers() throws MeasurementCollectorException {
try {
@SuppressWarnings("unchecked")
Enumeration<String> propNames = (Enumeration<String>) config.propertyNames();
while (propNames.hasMoreElements()) {
String propName = propNames.nextElement();
if (propName.startsWith(DATA_PARSER_PROP)) {
String[] parserConfigElements = config.get(propName).toString().split(",");
String nameInMsg = parserConfigElements[0];
String nameInDatabase = parserConfigElements[1];
String dataParserClassName = parserConfigElements[2];
Class<?> klass = Class.forName(dataParserClassName);
Constructor<?> constructor = klass.getConstructor(String.class);
ADataParser dataParser = (ADataParser) constructor.newInstance(nameInDatabase);
dataParsers.put(nameInMsg, dataParser);
System.out.println(dataParserClassName + " registered for " + nameInMsg + ", " + nameInDatabase);
}
}
}
catch (Exception e) {
throw new MeasurementCollectorException("Exception when registering configured data parsers", e);
}
}
@Override
public void execute(LocalDateTime timestamp, String msgPayload) {
try {
@SuppressWarnings("rawtypes")
Map payloadMap = jsonParser.parseJson(msgPayload);
@SuppressWarnings("rawtypes")
String name = (String)(((Map)(payloadMap.get("metadata"))).get("name"));
@SuppressWarnings("rawtypes")
String mbusMsgTxt = (String)(((Map)(payloadMap.get("data"))).get("telegram"));
String [] octetsTxt = mbusMsgTxt.split(" ");
byte [] octets = new byte[octetsTxt.length];
for (int i = 0; i < octetsTxt.length; i++) {
octets[i] = (byte)(Integer.parseInt(octetsTxt[i], 16) & 0xff);
}
MBusMessage mbusMsg = MBusMessage.decode(octets, octets.length);
VariableDataStructure variableDataStructure = mbusMsg.getVariableDataResponse();
variableDataStructure.decode();
List<DataRecord> dataRecords = variableDataStructure.getDataRecords();
//for (DataRecord dataRecord : dataRecords) {
// System.out.println(dataRecord.getScaledDataValue() + " " + dataRecord.getUnit().getUnit());
//}
List<ADataObject> measurementItems;
switch (name) {
case "thermom.":
measurementItems = hottisFourChannelThermometer.parse(timestamp, name, dataRecords);
break;
case "light":
case "computer":
case "freezer":
case "dryer":
case "laundry":
case "dishwasher":
measurementItems = finderOnePhasePowerMeter.parse(timestamp, name.substring(0,1).toUpperCase() + name.substring(1).toLowerCase(), dataRecords);
break;
case "electricity":
measurementItems = finderThreePhasePowerMeter.parse(timestamp, "Total", dataRecords);
break;
default:
System.out.println("unknown name");
measurementItems = null;
if (dataParsers.containsKey(name)) {
List<ADataObject> measurementItems = dataParsers.get(name).parse(timestamp, dataRecords);
for (ADataObject ado : measurementItems) {
System.out.println(ado);
}
queue.add(measurementItems);
System.out.println("Queue size: " + queue.size());
} else {
System.out.println("unknown name: " + name);
}
for (ADataObject ado : measurementItems) {
System.out.println(ado);
}
queue.add(measurementItems);
System.out.println("Queue size: " + queue.size());
} catch (Exception e) {
System.out.println("Exception when handling mbus message: " + e);
}

11
src/de/hottis/MeasurementCollector/MeasurementCollector.java
View File

@ -1,8 +1,6 @@
package de.hottis.MeasurementCollector;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentLinkedQueue;
public class MeasurementCollector {
static final String PROPS_FILENAME = "measurementCollector.props";
@ -14,15 +12,14 @@ public class MeasurementCollector {
final Properties config = new Properties();
config.load(MeasurementCollector.class.getClassLoader().getResourceAsStream(PROPS_FILENAME));
MqttReceiver mqttReceiver = new MqttReceiver(config);
mqttReceiver.connect();
// TestParser testParser = new TestParser();
// mqttReceiver.registerParser(testParser);
DataObjectQueue queue = new DataObjectQueue();
ConcurrentLinkedQueue<List<ADataObject>> queue = new ConcurrentLinkedQueue<List<ADataObject>>();
MBusParser mbusParser = new MBusParser(queue);
MBusParser mbusParser = new MBusParser(config, queue);
mbusParser.registerConfiguredDataParsers();
mqttReceiver.registerParser(mbusParser);
}

3
src/de/hottis/MeasurementCollector/MeasurementCollectorException.java
View File

@ -1,8 +1,7 @@
package de.hottis.MeasurementCollector;
@SuppressWarnings("serial")
public class MeasurementCollectorException extends Exception {
private static final long serialVersionUID = -5819010697931904741L;
public MeasurementCollectorException(String msg, Throwable cause) {
super(msg, cause);
}

16
src/de/hottis/MeasurementCollector/TestParser.java
View File

@ -1,16 +0,0 @@
package de.hottis.MeasurementCollector;
import java.time.LocalDateTime;
public class TestParser extends AMessageParser {
public TestParser() {
super("IoT/Watchdog");
}
@Override
public void execute(LocalDateTime timestamp, String msgPayload) {
System.out.println(timestamp.toString() + " " + msgPayload);
}
}

16
src/measurementCollector.props
View File

@ -1,3 +1,13 @@
mqtt.broker = tcp://172.16.2.15:1883
; mqtt.username = tron
; mqtt.password = geheim123
; mqtt.broker = tcp://172.16.2.15:1883
mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
mqtt.username = tron
mqtt.password = geheim123
mbus.dataparser.1 = light,Light,de.hottis.MeasurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.2 = computer,Computer,de.hottis.MeasurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.3 = laundry,Laundry,de.hottis.MeasurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.4 = dryer,Dryer,de.hottis.MeasurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.5 = dishwasher,Dishwasher,de.hottis.MeasurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.6 = freezer,Freezer,de.hottis.MeasurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.7 = electricity,Total,de.hottis.MeasurementCollector.FinderThreePhasePowerMeter
mbus.dataparser.8 = thermom.,null,de.hottis.MeasurementCollector.HottisFourChannelThermometer