This is an automated email from the ASF dual-hosted git repository. zehnder pushed a commit to branch SP-1592 in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit b04ae11e9bfa19d2548d0288aa41e6b331bb44ff Author: Philipp Zehnder <[email protected]> AuthorDate: Wed May 31 14:41:05 2023 +0200 [#1592] Add first version for an io link adapter --- .../connect/iiot/ConnectAdapterIiotInit.java | 2 + .../iiot/adapters/iolink/IfmAlMqttAdapter.java | 159 +++++++++++++++++++++ .../iiot/adapters/iolink/sensor/IoLinkSensor.java | 29 ++++ .../iiot/adapters/iolink/sensor/SensorVVB001.java | 152 ++++++++++++++++++++ .../documentation.md | 44 ++++++ .../icon.png | Bin 0 -> 9219 bytes .../strings.en | 50 +++++++ .../streampipes/sensor/SensorVVB001Test.java | 52 +++++++ 8 files changed, 488 insertions(+) diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java index 03685343b..dcd646eac 100644 --- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java +++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/ConnectAdapterIiotInit.java @@ -18,6 +18,7 @@ package org.apache.streampipes.connect.iiot; +import org.apache.streampipes.connect.iiot.adapters.iolink.IfmAlMqttAdapter; import org.apache.streampipes.connect.iiot.adapters.opcua.OpcUaAdapter; import org.apache.streampipes.connect.iiot.adapters.plc4x.modbus.Plc4xModbusAdapter; import org.apache.streampipes.connect.iiot.adapters.plc4x.s7.Plc4xS7Adapter; @@ -49,6 +50,7 @@ public class ConnectAdapterIiotInit extends ExtensionsModelSubmitter { 8001) .registerAdapter(new MachineDataSimulatorAdapter()) .registerAdapter(new FileReplayAdapter()) + .registerAdapter(new IfmAlMqttAdapter()) .registerAdapter(new RosBridgeAdapter()) .registerAdapter(new OpcUaAdapter()) .registerAdapter(new Plc4xS7Adapter()) diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/IfmAlMqttAdapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/IfmAlMqttAdapter.java new file mode 100644 index 000000000..548d1c3bb --- /dev/null +++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/IfmAlMqttAdapter.java @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.streampipes.connect.iiot.adapters.iolink; + +import org.apache.streampipes.commons.exceptions.connect.AdapterException; +import org.apache.streampipes.connect.iiot.adapters.iolink.sensor.SensorVVB001; +import org.apache.streampipes.extensions.management.connect.AdapterInterface; +import org.apache.streampipes.extensions.management.connect.adapter.parser.JsonParsers; +import org.apache.streampipes.extensions.management.connect.adapter.parser.json.JsonObjectParser; +import org.apache.streampipes.extensions.management.context.IAdapterGuessSchemaContext; +import org.apache.streampipes.extensions.management.context.IAdapterRuntimeContext; +import org.apache.streampipes.model.AdapterType; +import org.apache.streampipes.model.connect.adapter.AdapterConfiguration; +import org.apache.streampipes.model.connect.adapter.IEventCollector; +import org.apache.streampipes.model.connect.adapter.IParser; +import org.apache.streampipes.model.connect.guess.GuessSchema; +import org.apache.streampipes.pe.shared.config.mqtt.MqttConfig; +import org.apache.streampipes.pe.shared.config.mqtt.MqttConnectUtils; +import org.apache.streampipes.pe.shared.config.mqtt.MqttConsumer; +import org.apache.streampipes.sdk.builder.adapter.AdapterConfigurationBuilder; +import org.apache.streampipes.sdk.extractor.IAdapterParameterExtractor; +import org.apache.streampipes.sdk.extractor.StaticPropertyExtractor; +import org.apache.streampipes.sdk.helpers.Labels; +import org.apache.streampipes.sdk.helpers.Locales; +import org.apache.streampipes.sdk.helpers.Options; +import org.apache.streampipes.sdk.utils.Assets; + +import org.apache.commons.io.IOUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.InputStream; +import java.util.List; +import java.util.Map; + +public class IfmAlMqttAdapter implements AdapterInterface { + + public static final String ID = "org.apache.streampipes.connect.iiot.adapters.iolink"; + + private static final Logger LOG = LoggerFactory.getLogger(IfmAlMqttAdapter.class); + + private static final String PORTS = "ports"; + private static final String SENSOR_TYPE = "sensor_type"; + + private Thread thread; + private MqttConsumer mqttConsumer; + private MqttConfig mqttConfig; + + private IParser parser; + + private String sensorType = ""; + + private List<String> ports; + + public IfmAlMqttAdapter() { + parser = new JsonParsers(new JsonObjectParser()); + } + + @Override + public AdapterConfiguration declareConfig() { + return AdapterConfigurationBuilder + .create(ID) + .withLocales(Locales.EN) + .withAssets(Assets.DOCUMENTATION, Assets.ICON) + .withCategory(AdapterType.Generic, AdapterType.Manufacturing) + .requiredTextParameter(MqttConnectUtils.getBrokerUrlLabel()) + .requiredAlternatives(MqttConnectUtils.getAccessModeLabel(), MqttConnectUtils.getAlternativesOne(), + MqttConnectUtils.getAlternativesTwo()) + .requiredMultiValueSelection(Labels.withId(PORTS), + Options.from("Port 1", "Port 2", "Port 3", "Port 4")) + .requiredSingleValueSelection(Labels.withId(SENSOR_TYPE), + Options.from("VVB001")) + .requiredTextParameter(MqttConnectUtils.getTopicLabel()) + .buildConfiguration(); + } + + @Override + public void onAdapterStarted(IAdapterParameterExtractor extractor, + IEventCollector collector, + IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException { + var sensor = new SensorVVB001(); + + this.applyConfiguration(extractor.getStaticPropertyExtractor()); + this.mqttConsumer = new MqttConsumer( + this.mqttConfig, + (mqttEvent) -> { + try { + InputStream in = convertByte(mqttEvent); + parser.parse(in, (event) -> { + var data = (Map<String, Object>) event.get("data"); + var payload = (Map<String, Object>) data.get("payload"); + + for (int i = 0; i < ports.size(); i++) { + + var portResult = ((Map<String, Object>) + payload.get("/iolinkmaster/port[" + ports.get(i) + "]/iolinkdevice/pdin")); + var eventData = (String) portResult.get("data"); + + var parsedEvent = sensor.parseEvent(eventData); + parsedEvent.put("timestamp", System.currentTimeMillis() + i); + parsedEvent.put("port", "port" + ports.get(i)); + + collector.collect(parsedEvent); + } + }); + } catch (Exception e) { + // TODO add error message +// adapterRuntimeContext.getLogger().addErrorMessage(); + LOG.error("Could not parse event", e); + } + }); + + thread = new Thread(this.mqttConsumer); + thread.start(); + } + + @Override + public void onAdapterStopped(IAdapterParameterExtractor extractor, + IAdapterRuntimeContext adapterRuntimeContext) throws AdapterException { + this.mqttConsumer.close(); + } + + @Override + public GuessSchema onSchemaRequested(IAdapterParameterExtractor extractor, + IAdapterGuessSchemaContext adapterGuessSchemaContext) throws AdapterException { + this.applyConfiguration(extractor.getStaticPropertyExtractor()); + + return new SensorVVB001().getEventSchema(); + } + + private void applyConfiguration(StaticPropertyExtractor extractor) { + mqttConfig = MqttConnectUtils.getMqttConfig(extractor); + sensorType = extractor.selectedSingleValue(SENSOR_TYPE, String.class); + var selectedPorts = extractor.selectedMultiValues(PORTS, String.class); + ports = selectedPorts.stream() + .map(port -> port.substring(5)) + .toList(); + } + + private InputStream convertByte(byte[] event) { + return IOUtils.toInputStream(new String(event), "UTF-8"); + } +} diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/sensor/IoLinkSensor.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/sensor/IoLinkSensor.java new file mode 100644 index 000000000..f371aa1e3 --- /dev/null +++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/sensor/IoLinkSensor.java @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.streampipes.connect.iiot.adapters.iolink.sensor; + +import org.apache.streampipes.model.connect.guess.GuessSchema; + +import java.util.Map; + +public interface IoLinkSensor { + GuessSchema getEventSchema(); + + Map<String, Object> parseEvent(String encodedEvent); +} diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/sensor/SensorVVB001.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/sensor/SensorVVB001.java new file mode 100644 index 000000000..5a61b869c --- /dev/null +++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/adapters/iolink/sensor/SensorVVB001.java @@ -0,0 +1,152 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.streampipes.connect.iiot.adapters.iolink.sensor; + +import org.apache.streampipes.model.connect.guess.GuessSchema; +import org.apache.streampipes.model.schema.PropertyScope; +import org.apache.streampipes.sdk.builder.PrimitivePropertyBuilder; +import org.apache.streampipes.sdk.builder.adapter.GuessSchemaBuilder; +import org.apache.streampipes.sdk.utils.Datatypes; + +import java.util.HashMap; +import java.util.Map; + +import static org.apache.streampipes.sdk.helpers.EpProperties.timestampProperty; + +public class SensorVVB001 implements IoLinkSensor { + + public static final String V_RMS_NAME = "vRms"; + public static final String A_PEAK_NAME = "aPeak"; + public static final String A_RMS_NAME = "aRms"; + public static final String TEMPERATURE_NAME = "temperature"; + public static final String CREST_NAME = "crest"; + public static final String STATUS_NAME = "status"; + public static final String OUT_1_NAME = "out1"; + public static final String OUT_2_NAME = "out2"; + public static final String PORT_NAME = "port"; + + + @Override + public GuessSchema getEventSchema() { + + return GuessSchemaBuilder.create() + .property(timestampProperty("timestamp")) + .sample("timestamp", 1685525380729L) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Float, V_RMS_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Speed RMS value") + .build() + ) + .sample(V_RMS_NAME, 0.0023) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Float, A_PEAK_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Acceleration peak value") + .build() + ) + .sample(A_PEAK_NAME, 6.6) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Float, A_RMS_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Acceleration RMS value") + .build() + ) + .sample(A_RMS_NAME, 1.8) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Float, TEMPERATURE_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Current temperature") + .build() + ) + .sample(TEMPERATURE_NAME, 22.0) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Float, CREST_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Acceleration crest factor") + .build() + ) + .sample(CREST_NAME, 3.7) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Integer, STATUS_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Device status (0: OK, 1: Maintenance required, " + + "2: Out of specification, 3: Function Test, 4: Error)") + .build() + ) + .sample(STATUS_NAME, 0) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Boolean, OUT_1_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Current state of the digital signal") + .build() + ) + .sample(OUT_1_NAME, true) + .property( + PrimitivePropertyBuilder + .create(Datatypes.Boolean, OUT_2_NAME) + .scope(PropertyScope.MEASUREMENT_PROPERTY) + .description("Current state of the digital signal") + .build() + ) + .sample(OUT_2_NAME, true) + .property( + PrimitivePropertyBuilder + .create(Datatypes.String, PORT_NAME) + .scope(PropertyScope.DIMENSION_PROPERTY) + .description("Port the sensor is connected to at IOLink master") + .build() + ) + .sample(PORT_NAME, "port1") + .build(); + } + + @Override + public Map<String, Object> parseEvent(String encodedEvent) { + var event = new HashMap<String, Object>(); + event.put(V_RMS_NAME, getValue(0, 4, 0.0001, encodedEvent)); + event.put(A_PEAK_NAME, getValue(8, 12, 0.1, encodedEvent)); + event.put(A_RMS_NAME, getValue(16, 20, 0.1, encodedEvent)); + event.put(TEMPERATURE_NAME, getValue(24, 28, 0.1, encodedEvent)); + event.put(CREST_NAME, getValue(32, 36, 0.1, encodedEvent)); + event.put(STATUS_NAME, (int) getValue(38, 39, 1.0, encodedEvent)); + + // The last two bits represent the status + var outValues = (int) getValue(39, 40, 1.0, encodedEvent); + event.put(OUT_1_NAME, (outValues & 1) != 0); + event.put(OUT_2_NAME, (outValues & 2) != 0); + + return event; + } + + private double getValue(int start, int end, double scaleFactor, String encodedEvent) { + var word = encodedEvent.substring(start, end); + + var rawValue = (double) (Integer.parseInt(word, 16)); + + return rawValue * scaleFactor; + } +} diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/documentation.md b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/documentation.md new file mode 100644 index 000000000..1d68c9e04 --- /dev/null +++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/documentation.md @@ -0,0 +1,44 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one or more + ~ contributor license agreements. See the NOTICE file distributed with + ~ this work for additional information regarding copyright ownership. + ~ The ASF licenses this file to You under the Apache License, Version 2.0 + ~ (the "License"); you may not use this file except in compliance with + ~ the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + ~ + --> + +## MQTT + +<p align="center"> + <img src="icon.png" width="150px;" class="pe-image-documentation"/> +</p> + +*** + +## Description + +TBD + + +*** + +## Configuration + +Describe the configuration parameters here + +### TBD + +TBD + +## Output + +TBD diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/icon.png b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/icon.png new file mode 100644 index 000000000..39c33a25e Binary files /dev/null and b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/icon.png differ diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/strings.en b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/strings.en new file mode 100644 index 000000000..efe91e3d1 --- /dev/null +++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.adapters.iolink/strings.en @@ -0,0 +1,50 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + + +org.apache.streampipes.connect.iiot.adapters.iolink.title=IOLink +org.apache.streampipes.connect.iiot.adapters.iolink.description=Consumes MQTT data representing IOLink data + +access-mode.title=Access Mode +access-mode.description=Provide user credentials for the MQTT broker + +anonymous-alternative.title=Unauthenticated +anonymous-alternative.description= + +username-alternative.title=Username/Password +username-alternative.description= + +username-group.title=User Group +username-group.description= + +username.title=Username +username.description= + +password.title=Password +password.description= + +broker_url.title=Broker URL +broker_url.description=Example: tcp://test-server.com:1883 (Protocol required. Port required)" + +topic.title=Topic +topic.description=Example: test/topic + +ports.title=Ports +ports.description=Select the ports with the selected sensor type + +sensor_type.title=Sensor Type +sensor_type.description=Select the sensor type that you want to connect diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/sensor/SensorVVB001Test.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/sensor/SensorVVB001Test.java new file mode 100644 index 000000000..199818d51 --- /dev/null +++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/test/java/org/apache/streampipes/sensor/SensorVVB001Test.java @@ -0,0 +1,52 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.streampipes.sensor; + +import org.apache.streampipes.connect.iiot.adapters.iolink.sensor.SensorVVB001; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +public class SensorVVB001Test { + + private SensorVVB001 sensor = new SensorVVB001(); + + private String sampleInput = "0017FC000042FF000012FF00015CFF000025FF03"; + + @Test + public void checkEventSize() { + var result = sensor.parseEvent(sampleInput); + assertEquals(8, result.keySet().size()); + } + + @Test + public void parseEvent() { + var result = sensor.parseEvent(sampleInput); + + assertEquals(0.0023, (double) result.get(SensorVVB001.V_RMS_NAME), 0.001); + assertEquals(6.600, (double) result.get(SensorVVB001.A_PEAK_NAME), 0.001); + assertEquals(1.8, (double) result.get(SensorVVB001.A_RMS_NAME), 0.001); + assertEquals(34.8, (double) result.get(SensorVVB001.TEMPERATURE_NAME), 0.001); + assertEquals(3.7, (double) result.get(SensorVVB001.CREST_NAME), 0.001); + assertEquals(0, result.get(SensorVVB001.STATUS_NAME)); + assertEquals(true, result.get(SensorVVB001.OUT_1_NAME)); + assertEquals(true, result.get(SensorVVB001.OUT_2_NAME)); + } +} \ No newline at end of file
