You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@edgent.apache.org by dj...@apache.org on 2016/04/01 18:54:51 UTC
[1/5] incubator-quarks git commit: ELM327 protocol layer
Repository: incubator-quarks
Updated Branches:
refs/heads/master 35c821cb8 -> a507e342c
ELM327 protocol layer
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/eb267121
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/eb267121
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/eb267121
Branch: refs/heads/master
Commit: eb267121209412d5b716f14c414c9f83be96b045
Parents: 0b58e15
Author: Dan Debrunner <dj...@debrunners.com>
Authored: Tue Mar 29 08:27:06 2016 -0700
Committer: Dan Debrunner <dj...@debrunners.com>
Committed: Tue Mar 29 08:27:06 2016 -0700
----------------------------------------------------------------------
samples/connectors/build.xml | 1 +
.../quarks/samples/connectors/elm327/Cmd.java | 76 ++++++++++
.../connectors/elm327/ELM327Streams.java | 70 +++++++++
.../samples/connectors/elm327/Elm327Cmds.java | 63 +++++++++
.../samples/connectors/elm327/Pids01.java | 141 +++++++++++++++++++
.../samples/connectors/elm327/package-info.java | 27 ++++
.../elm327/runtime/CommandExecutor.java | 118 ++++++++++++++++
7 files changed, 496 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/eb267121/samples/connectors/build.xml
----------------------------------------------------------------------
diff --git a/samples/connectors/build.xml b/samples/connectors/build.xml
index 964a5dd..84d2d5a 100644
--- a/samples/connectors/build.xml
+++ b/samples/connectors/build.xml
@@ -13,6 +13,7 @@
<pathelement location="${quarks.connectors}/file/lib/quarks.connectors.file.jar"/>
<pathelement location="${quarks.connectors}/iotf/lib/quarks.connectors.iotf.jar"/>
<pathelement location="${quarks.connectors}/jdbc/lib/quarks.connectors.jdbc.jar"/>
+ <pathelement location="${quarks.connectors}/serial/lib/quarks.connectors.serial.jar"/>
<pathelement location="${lib}/quarks.samples.topology.jar"/>
<pathelement location="${lib}/quarks.samples.utils.jar"/>
<path refid="quarks.ext.classpath" />
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/eb267121/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Cmd.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Cmd.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Cmd.java
new file mode 100644
index 0000000..86873be
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Cmd.java
@@ -0,0 +1,76 @@
+/*
+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 quarks.samples.connectors.elm327;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import com.google.gson.JsonObject;
+
+/**
+ * ELM327 and OBD-II command interface.
+ *
+ */
+public interface Cmd {
+ /**
+ * Key ({@value}) for PID identifier in JSON result.
+ */
+ String PID = "pid";
+
+ /**
+ * Key ({@value}) for timestamp in JSON result. Timestamp value is the
+ * number of milliseconds since the 1907 epoch.
+ */
+ String TS = "ts";
+
+ /**
+ * Key ({@value}) for the returned value in JSON result.
+ * May not be present.
+ */
+ String VALUE = "value";
+
+ /**
+ * How the command is written to the serial port.
+ *
+ * @param out
+ * OutputStream to write bytes to.
+ * @throws IOException
+ * Exception writing bytes.
+ */
+ void writeCmd(OutputStream out) throws IOException;
+
+ /**
+ * Process the reply into a result.
+ *
+ * @param result
+ * JSON object to populate with the result.
+ * @param reply
+ * Bytes that were returned from the command execution.
+ *
+ * @return {@code true} result is valid, {@code false} otherwise.
+ */
+ boolean result(JsonObject result, byte[] reply);
+
+ /**
+ * Unique identifier of the command.
+ *
+ * @return Unique identifier of the command.
+ */
+ String id();
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/eb267121/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java
new file mode 100644
index 0000000..689c00d
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java
@@ -0,0 +1,70 @@
+/*
+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 quarks.samples.connectors.elm327;
+
+import java.util.concurrent.TimeUnit;
+
+import com.google.gson.JsonArray;
+
+import quarks.samples.connectors.elm327.runtime.CommandExecutor;
+import quarks.connectors.serial.SerialDevice;
+import quarks.function.Supplier;
+import quarks.topology.TStream;
+
+/**
+ * Streams fetching OBD-II data from an ELM327 through
+ * a serial device.
+ *
+ * @see <a href="https://en.wikipedia.org/wiki/ELM327">ELM327</a>
+ */
+public class ELM327Streams {
+
+ /**
+ * Periodically execute a number of ELM327 commands.
+ * Each tuple on the returned stream is a JSON array containing
+ * the result for each command.
+ * <BR>
+ * Each result is a JSON object containing the
+ * {@link Cmd#id() command identifier} with key {@link Cmd#PID pid}
+ * and any result set by the individual command, typically with
+ * the key {@link Cmd#VALUE value}.
+ *
+ * @param device Serial device the ELM327 is connected to.
+ * @param period Period to poll.
+ * @param unit Unit of {@code period}.
+ * @param cmds Commands to execute.
+ * @return Stream containing the results of the command exections.
+ */
+ public static TStream<JsonArray> poll(SerialDevice device, long period, TimeUnit unit, Cmd ... cmds) {
+
+ Supplier<JsonArray> data = device.getSource(
+ port ->
+ {
+ JsonArray array = new JsonArray();
+ for (Cmd cmd : cmds) {
+ array.add(CommandExecutor.execute(cmd, port.getOutput(), port.getInput()));
+ }
+ return array;
+
+ });
+
+ return device.topology().poll(data, period, unit);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/eb267121/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java
new file mode 100644
index 0000000..eae3231
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java
@@ -0,0 +1,63 @@
+/*
+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 quarks.samples.connectors.elm327;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+
+import com.google.gson.JsonObject;
+
+/**
+ * ELM327 commands.
+ *
+ *
+ */
+public enum Elm327Cmds implements Cmd {
+
+ INIT("ATZ"),
+ ECHO_OFF("ATE0"),
+ PROTOCOL_3("ATSP3"),
+ PROTOCOL_5("ATSP5"),
+ BYPASS_INIT("ATBI"),
+ FAST_INIT("ATFI"),
+ SLOW_INIT("ATSI"),;
+
+ private byte[] cmd;
+
+ Elm327Cmds(String code) {
+ cmd = (code + "\r").getBytes(StandardCharsets.US_ASCII);
+ }
+
+ @Override
+ public void writeCmd(OutputStream out) throws IOException {
+ out.write(cmd);
+ }
+
+ @Override
+ public boolean result(JsonObject result, byte[] data) {
+ return true;
+ }
+
+ @Override
+ public String id() {
+ return name();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/eb267121/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Pids01.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Pids01.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Pids01.java
new file mode 100644
index 0000000..0860137
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Pids01.java
@@ -0,0 +1,141 @@
+/*
+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 quarks.samples.connectors.elm327;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+
+import com.google.gson.JsonObject;
+
+import quarks.samples.connectors.elm327.runtime.CommandExecutor;
+
+/**
+ * OBD-II Standard Mode 01 Pids.
+ *
+ *
+ * @see <a href="https://en.wikipedia.org/wiki/OBD-II_PIDs#Mode_01">OBD-II Mode 01 Pids</a>
+ */
+public enum Pids01 implements Cmd {
+
+ /**
+ * Get the list of available PIDs.
+ */
+ AVAILABLE_PIDS("00"),
+
+ /**
+ * Engine coolant temperature in °C.
+ */
+ ENGINE_COOLANT_TEMP("05") {
+ @Override
+ protected boolean decode(JsonObject result, byte[] reply) {
+
+ int[] binary = CommandExecutor.binary(reply, 4, 2);
+
+ int c = binary[0] - 40;
+ result.addProperty(VALUE, c);
+
+ return true;
+ }
+ },
+
+ /**
+ * Engine speed in rpm.
+ */
+ RPM("0C") {
+ @Override
+ protected boolean decode(JsonObject result, byte[] reply) {
+
+ int[] binary = CommandExecutor.binary(reply, 4, 4);
+ int rpm = ((binary[0] * 256) + binary[1])/4;
+ result.addProperty(VALUE, rpm);
+
+ return true;
+ }
+ },
+
+ /**
+ * Vehicle speed in km/h.
+ */
+ SPEED("0D"){
+ @Override
+ protected boolean decode(JsonObject result, byte[] reply) {
+
+ int[] binary = CommandExecutor.binary(reply, 4, 2);
+
+ result.addProperty(VALUE, binary[0]);
+
+ return true;
+ }
+ },
+
+ /**
+ * Engine air intake temperature in °C.
+ */
+ AIR_INTAKE_TEMP("0F"){
+ @Override
+ protected boolean decode(JsonObject result, byte[] reply) {
+
+ int[] binary = CommandExecutor.binary(reply, 4, 2);
+
+ int c = binary[0] - 40;
+ result.addProperty(VALUE, c);
+
+ return true;
+ }
+ },
+ ;
+
+ private final String pid;
+ private final byte[] cmd;
+
+ Pids01(String pid) {
+ this.pid = pid;
+ cmd = ("01" + pid + "1\r").getBytes(StandardCharsets.US_ASCII);
+ }
+
+ public String id() {
+ return pid;
+ }
+
+ @Override
+ public void writeCmd(OutputStream out) throws IOException {
+ out.write(cmd);
+ }
+ @Override
+ public final boolean result(JsonObject result, byte[] data) {
+ return validateReply(data) && decode(result, data);
+ }
+ boolean decode(JsonObject result, byte[] data) {
+ return true;
+ }
+
+ boolean validateReply(byte[] reply) {
+ if (reply[0] != '4')
+ return false;
+ if (reply[1] != '1')
+ return false;
+ if (reply[2] != pid.charAt(0))
+ return false;
+ if (reply[3] != pid.charAt(1))
+ return false;
+
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/eb267121/samples/connectors/src/main/java/quarks/samples/connectors/elm327/package-info.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/package-info.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/package-info.java
new file mode 100644
index 0000000..e39904f
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/package-info.java
@@ -0,0 +1,27 @@
+/*
+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.
+*/
+/**
+ * OBD-II protocol sample using ELM327.
+ *
+ * ELM327 devices allow connectivity to a vehicle's OBD-II information.
+ *
+ * @see <a href="https://en.wikipedia.org/wiki/OBD-II">OBD-II</a>
+ * @see <a href="https://en.wikipedia.org/wiki/ELM327">ELM327</a>
+ */
+package quarks.samples.connectors.elm327;
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/eb267121/samples/connectors/src/main/java/quarks/samples/connectors/elm327/runtime/CommandExecutor.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/runtime/CommandExecutor.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/runtime/CommandExecutor.java
new file mode 100644
index 0000000..f390b0c
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/runtime/CommandExecutor.java
@@ -0,0 +1,118 @@
+/*
+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 quarks.samples.connectors.elm327.runtime;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import com.google.gson.JsonObject;
+
+import quarks.samples.connectors.elm327.Cmd;
+import quarks.samples.connectors.elm327.Elm327Cmds;
+
+/**
+ * Runtime execution of ELM327 & OBD-II commands.
+ *
+ */
+public class CommandExecutor {
+
+ public static int[] binary(byte[] reply, int offset, int length) {
+ int[] binary = new int[length / 2];
+ for (int i = 0; i < binary.length; i++) {
+ int h = Character.digit(reply[offset++], 16);
+ int l = Character.digit(reply[offset++], 16);
+ binary[i] = ((h * 16) + l);
+ }
+ return binary;
+ }
+
+ public static void initialize(Cmd protocol, OutputStream out, InputStream in) {
+ try {
+
+ executeUntilOK(10, Elm327Cmds.INIT, out, in);
+ Thread.sleep(1000);
+
+ executeUntilOK(1, Elm327Cmds.ECHO_OFF, out, in);
+
+ executeUntilOK(1, protocol, out, in);
+ executeUntilOK(1, Elm327Cmds.SLOW_INIT, out, in);
+ Thread.sleep(1000);
+
+ } catch (Exception ioe) {
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ private static boolean readUntilPrompt(InputStream in, ByteArrayOutputStream bytes) throws IOException {
+ bytes.reset();
+ for (;;) {
+ int b = in.read();
+ if (b == -1)
+ return false;
+ if (b == ' ')
+ continue;
+ if (b == '\r')
+ continue;
+ if (b == '>')
+ return true;
+
+ bytes.write(b);
+ }
+ }
+
+ public static JsonObject executeUntilOK(int n, Cmd cmd, OutputStream out, InputStream in) throws IOException {
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(16)) {
+ for (int i = 0; i < n; i++) {
+ cmd.writeCmd(out);
+ out.flush();
+
+ if (!readUntilPrompt(in, bytes))
+ continue;
+
+ byte[] reply = bytes.toByteArray();
+ JsonObject j = new JsonObject();
+ if (cmd.result(j, reply))
+ return j;
+ break;
+ }
+ }
+ throw new IllegalStateException("Could not execute command:" + cmd);
+ }
+
+ public static JsonObject execute(Cmd cmd, OutputStream out, InputStream in) {
+ try (ByteArrayOutputStream bytes = new ByteArrayOutputStream(16)) {
+ cmd.writeCmd(out);
+ out.flush();
+
+ JsonObject result = new JsonObject();
+ result.addProperty(Cmd.PID, cmd.id());
+ result.addProperty(Cmd.TS, System.currentTimeMillis());
+
+ readUntilPrompt(in, bytes);
+
+ cmd.result(result, bytes.toByteArray());
+
+ return result;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
[3/5] incubator-quarks git commit: Add initialization method.
Posted by dj...@apache.org.
Add initialization method.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/4da09f2f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/4da09f2f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/4da09f2f
Branch: refs/heads/master
Commit: 4da09f2fdef1bc562b32dafc5ba8eefa9143cc23
Parents: a599b33
Author: Dan Debrunner <dj...@debrunners.com>
Authored: Tue Mar 29 15:04:48 2016 -0700
Committer: Dan Debrunner <dj...@debrunners.com>
Committed: Tue Mar 29 15:04:48 2016 -0700
----------------------------------------------------------------------
.../quarks/samples/connectors/elm327/Elm327Cmds.java | 12 ++++++++++++
1 file changed, 12 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/4da09f2f/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java
index eae3231..f532982 100644
--- a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Cmds.java
@@ -24,6 +24,9 @@ import java.nio.charset.StandardCharsets;
import com.google.gson.JsonObject;
+import quarks.connectors.serial.SerialDevice;
+import quarks.samples.connectors.elm327.runtime.CommandExecutor;
+
/**
* ELM327 commands.
*
@@ -59,5 +62,14 @@ public enum Elm327Cmds implements Cmd {
public String id() {
return name();
}
+
+ /**
+ * Initialize the ELM327 to a specific protocol.
+ * @param device Serial device the ELM327 is connected to.
+ * @param protocol OBD-II protocol to initialize to.
+ */
+ public static void initializeProtocol(SerialDevice device, Elm327Cmds protocol) {
+ device.setInitializer(port -> CommandExecutor.initialize(protocol, port.getOutput(), port.getInput()));
+ }
}
[2/5] incubator-quarks git commit: Add initial Obd2Streamss
Posted by dj...@apache.org.
Add initial Obd2Streamss
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/a599b33f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/a599b33f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/a599b33f
Branch: refs/heads/master
Commit: a599b33f8e8269b42768bceaed6a41e877fa84aa
Parents: eb26712
Author: Dan Debrunner <dj...@debrunners.com>
Authored: Tue Mar 29 15:03:21 2016 -0700
Committer: Dan Debrunner <dj...@debrunners.com>
Committed: Tue Mar 29 15:03:21 2016 -0700
----------------------------------------------------------------------
.../quarks/samples/apps/obd2/Obd2Streams.java | 143 +++++++++++++++++++
1 file changed, 143 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a599b33f/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java
----------------------------------------------------------------------
diff --git a/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java b/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java
new file mode 100644
index 0000000..ea36396
--- /dev/null
+++ b/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java
@@ -0,0 +1,143 @@
+/*
+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 quarks.samples.apps.obd2;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static quarks.analytics.math3.stat.Regression.SLOPE;
+import static quarks.analytics.math3.stat.Statistic.MAX;
+import static quarks.samples.connectors.elm327.Cmd.PID;
+import static quarks.samples.connectors.elm327.Cmd.VALUE;
+import static quarks.samples.connectors.elm327.Pids01.AIR_INTAKE_TEMP;
+import static quarks.samples.connectors.elm327.Pids01.ENGINE_COOLANT_TEMP;
+import static quarks.samples.connectors.elm327.Pids01.RPM;
+import static quarks.samples.connectors.elm327.Pids01.SPEED;
+
+import java.util.concurrent.TimeUnit;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+
+import quarks.analytics.math3.json.JsonAnalytics;
+import quarks.connectors.serial.SerialDevice;
+import quarks.samples.connectors.elm327.ELM327Streams;
+import quarks.topology.TStream;
+import quarks.topology.TWindow;
+
+/**
+ * Sample OBD-II streams.
+ *
+ */
+public class Obd2Streams {
+
+ /**
+ * Get a stream of temperature readings which
+ * are increasing over the last minute.
+ *
+ * Poll temperatures every five seconds and
+ * calculate the maximum reading and rate of change
+ * (slope) over the last minute, partitioned by parameter
+ * {@link Cmd#PID pid}. Filter so that only
+ * those with a rate of increase greater than
+ * or equal to 1°C/minute is present on the returned stream.
+ *
+ * Temperatures included are {@link AIR_INTAKE_TEMP} and
+ * {@link ENGINE_COOLANT_TEMP}.
+ *
+ * @param device Serial device the ELM327 is connected to.
+ * @return Stream that will contain parameters with increasing temperatures.
+ */
+ public static TStream<JsonObject> increasingTemps(SerialDevice device) {
+
+ TStream<JsonArray> tempsA = ELM327Streams.poll(device, 5, SECONDS,
+ AIR_INTAKE_TEMP,
+ ENGINE_COOLANT_TEMP);
+
+ TStream<JsonObject> temps = tempsA.flatMap(je -> je).map(je -> je.getAsJsonObject());
+
+ TWindow<JsonObject, JsonElement> window = temps.last(1, MINUTES, j -> j.get(PID));
+
+ TStream<JsonObject> temperatureRate = JsonAnalytics.aggregate(window, PID, VALUE, MAX, SLOPE);
+
+ // Have the stream contain only tuples where
+ // the rise in temperatures >= 1°C/minute
+ temperatureRate = temperatureRate.filter(j -> {
+ JsonObject v = getObject(j, "value");
+ return v.has("SLOPE") && getDouble(v, "SLOPE") >= 1.0;
+ });
+
+ return temperatureRate;
+ }
+
+ /**
+ * Get a stream containing vehicle speed (km/h)
+ * and engine revs (rpm).
+ *
+ * {@link SPEED Speed} and {@link RPM engine revs}
+ * are polled every 200ms and returned as a stream
+ * containing JSON objects with keys {@code speed}
+ * and {@code rpm}.
+ *
+ * The two readings may not be exactly consistent with
+ * each other as there are fetched sequentially from
+ * the ELM327.
+ *
+ * @param device Serial device the ELM327 is connected to.
+ * @return Stream that will contain speed and engine revolutions.
+ */
+ public static TStream<JsonObject> tach(SerialDevice serial) {
+
+ TStream<JsonArray> rpmSpeed = ELM327Streams.poll(serial, 200, TimeUnit.MILLISECONDS,
+ SPEED, RPM);
+
+ TStream<JsonObject> tach = rpmSpeed.map(ja -> {
+ JsonObject j = new JsonObject();
+
+ double speed = getDouble(ja.get(0), VALUE);
+ double rpm = getDouble(ja.get(1), VALUE);
+ j.addProperty("speed", speed);
+ j.addProperty("rpm", rpm);
+
+ return j;
+ });
+
+ return tach;
+ }
+
+ /**
+ * Utility method to simplify accessing a JSON object.
+ * @param json JSON object containing the object to be got.
+ * @param key Key of the object to be got.
+ * @return JSON object with key {@code key} from {@code json}.
+ */
+ public static JsonObject getObject(JsonObject json, String key) {
+ return json.getAsJsonObject(key);
+ }
+
+ /**
+ * Utility method to simplify accessing a number as a double.
+ * @param json JSON object containing the number to be got.
+ * @param key Key of the number to be got.
+ * @return Number with key {@code key} from {@code json}.
+ */
+ public static double getDouble(JsonElement json, String key) {
+ return json.getAsJsonObject().get(key).getAsDouble();
+ }
+}
[5/5] incubator-quarks git commit: Merge branch 'pr-52'
Posted by dj...@apache.org.
Merge branch 'pr-52'
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/a507e342
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/a507e342
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/a507e342
Branch: refs/heads/master
Commit: a507e342c408e541b77af988e226bb95e47e6047
Parents: 35c821c 82fcaf6
Author: Daniel J. Debrunner <de...@us.ibm.com>
Authored: Fri Apr 1 09:54:50 2016 -0700
Committer: Daniel J. Debrunner <de...@us.ibm.com>
Committed: Fri Apr 1 09:54:50 2016 -0700
----------------------------------------------------------------------
samples/apps/build.xml | 7 +-
samples/connectors/build.xml | 1 +
.../quarks/samples/connectors/elm327/Cmd.java | 76 ++++++++++
.../samples/connectors/elm327/Elm327Cmds.java | 75 ++++++++++
.../connectors/elm327/Elm327Streams.java | 70 +++++++++
.../samples/connectors/elm327/Pids01.java | 141 ++++++++++++++++++
.../samples/connectors/elm327/package-info.java | 27 ++++
.../elm327/runtime/CommandExecutor.java | 118 +++++++++++++++
.../samples/connectors/obd2/Obd2Streams.java | 145 +++++++++++++++++++
9 files changed, 657 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a507e342/samples/apps/build.xml
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/a507e342/samples/connectors/build.xml
----------------------------------------------------------------------
[4/5] incubator-quarks git commit: Move OBD-II code to be under
connectors
Posted by dj...@apache.org.
Move OBD-II code to be under connectors
Project: http://git-wip-us.apache.org/repos/asf/incubator-quarks/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quarks/commit/82fcaf6e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quarks/tree/82fcaf6e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quarks/diff/82fcaf6e
Branch: refs/heads/master
Commit: 82fcaf6ecf37753eccb3d41408c52bc47f2518f5
Parents: 4da09f2
Author: Dan Debrunner <dj...@debrunners.com>
Authored: Tue Mar 29 15:22:39 2016 -0700
Committer: Dan Debrunner <dj...@debrunners.com>
Committed: Tue Mar 29 15:22:39 2016 -0700
----------------------------------------------------------------------
samples/apps/build.xml | 7 +-
.../quarks/samples/apps/obd2/Obd2Streams.java | 143 ------------------
.../connectors/elm327/ELM327Streams.java | 70 ---------
.../connectors/elm327/Elm327Streams.java | 70 +++++++++
.../samples/connectors/obd2/Obd2Streams.java | 145 +++++++++++++++++++
5 files changed, 219 insertions(+), 216 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/82fcaf6e/samples/apps/build.xml
----------------------------------------------------------------------
diff --git a/samples/apps/build.xml b/samples/apps/build.xml
index e9e5454..df08fb7 100644
--- a/samples/apps/build.xml
+++ b/samples/apps/build.xml
@@ -10,10 +10,11 @@
<path refid="quarks.samples.classpath"/>
<pathelement location="${quarks.connectors}/mqtt/lib/quarks.connectors.mqtt.jar"/>
<pathelement location="${quarks.connectors}/file/lib/quarks.connectors.file.jar"/>
+ <pathelement location="${quarks.connectors}/serial/lib/quarks.connectors.serial.jar"/>
<pathelement location="${quarks.analytics}/math3/lib/quarks.analytics.math3.jar"/>
- <pathelement location="${quarks.analytics}/math3/ext/commons-math3-3.4.1/commons-math3-3.4.1.jar"/>
- <pathelement location="${lib}/quarks.samples.topology.jar"/>
- <pathelement location="${lib}/quarks.samples.utils.jar"/>
+ <pathelement location="${quarks.samples}/quarks.samples.topology.jar"/>
+ <pathelement location="${quarks.samples}/lib/quarks.samples.utils.jar"/>
+ <pathelement location="${quarks.samples}/quarks.samples.connectors.jar"/>
<path refid="quarks.ext.classpath" />
</path>
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/82fcaf6e/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java
----------------------------------------------------------------------
diff --git a/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java b/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java
deleted file mode 100644
index ea36396..0000000
--- a/samples/apps/src/main/java/quarks/samples/apps/obd2/Obd2Streams.java
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
-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 quarks.samples.apps.obd2;
-
-import static java.util.concurrent.TimeUnit.MINUTES;
-import static java.util.concurrent.TimeUnit.SECONDS;
-import static quarks.analytics.math3.stat.Regression.SLOPE;
-import static quarks.analytics.math3.stat.Statistic.MAX;
-import static quarks.samples.connectors.elm327.Cmd.PID;
-import static quarks.samples.connectors.elm327.Cmd.VALUE;
-import static quarks.samples.connectors.elm327.Pids01.AIR_INTAKE_TEMP;
-import static quarks.samples.connectors.elm327.Pids01.ENGINE_COOLANT_TEMP;
-import static quarks.samples.connectors.elm327.Pids01.RPM;
-import static quarks.samples.connectors.elm327.Pids01.SPEED;
-
-import java.util.concurrent.TimeUnit;
-
-import com.google.gson.JsonArray;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonObject;
-
-import quarks.analytics.math3.json.JsonAnalytics;
-import quarks.connectors.serial.SerialDevice;
-import quarks.samples.connectors.elm327.ELM327Streams;
-import quarks.topology.TStream;
-import quarks.topology.TWindow;
-
-/**
- * Sample OBD-II streams.
- *
- */
-public class Obd2Streams {
-
- /**
- * Get a stream of temperature readings which
- * are increasing over the last minute.
- *
- * Poll temperatures every five seconds and
- * calculate the maximum reading and rate of change
- * (slope) over the last minute, partitioned by parameter
- * {@link Cmd#PID pid}. Filter so that only
- * those with a rate of increase greater than
- * or equal to 1°C/minute is present on the returned stream.
- *
- * Temperatures included are {@link AIR_INTAKE_TEMP} and
- * {@link ENGINE_COOLANT_TEMP}.
- *
- * @param device Serial device the ELM327 is connected to.
- * @return Stream that will contain parameters with increasing temperatures.
- */
- public static TStream<JsonObject> increasingTemps(SerialDevice device) {
-
- TStream<JsonArray> tempsA = ELM327Streams.poll(device, 5, SECONDS,
- AIR_INTAKE_TEMP,
- ENGINE_COOLANT_TEMP);
-
- TStream<JsonObject> temps = tempsA.flatMap(je -> je).map(je -> je.getAsJsonObject());
-
- TWindow<JsonObject, JsonElement> window = temps.last(1, MINUTES, j -> j.get(PID));
-
- TStream<JsonObject> temperatureRate = JsonAnalytics.aggregate(window, PID, VALUE, MAX, SLOPE);
-
- // Have the stream contain only tuples where
- // the rise in temperatures >= 1°C/minute
- temperatureRate = temperatureRate.filter(j -> {
- JsonObject v = getObject(j, "value");
- return v.has("SLOPE") && getDouble(v, "SLOPE") >= 1.0;
- });
-
- return temperatureRate;
- }
-
- /**
- * Get a stream containing vehicle speed (km/h)
- * and engine revs (rpm).
- *
- * {@link SPEED Speed} and {@link RPM engine revs}
- * are polled every 200ms and returned as a stream
- * containing JSON objects with keys {@code speed}
- * and {@code rpm}.
- *
- * The two readings may not be exactly consistent with
- * each other as there are fetched sequentially from
- * the ELM327.
- *
- * @param device Serial device the ELM327 is connected to.
- * @return Stream that will contain speed and engine revolutions.
- */
- public static TStream<JsonObject> tach(SerialDevice serial) {
-
- TStream<JsonArray> rpmSpeed = ELM327Streams.poll(serial, 200, TimeUnit.MILLISECONDS,
- SPEED, RPM);
-
- TStream<JsonObject> tach = rpmSpeed.map(ja -> {
- JsonObject j = new JsonObject();
-
- double speed = getDouble(ja.get(0), VALUE);
- double rpm = getDouble(ja.get(1), VALUE);
- j.addProperty("speed", speed);
- j.addProperty("rpm", rpm);
-
- return j;
- });
-
- return tach;
- }
-
- /**
- * Utility method to simplify accessing a JSON object.
- * @param json JSON object containing the object to be got.
- * @param key Key of the object to be got.
- * @return JSON object with key {@code key} from {@code json}.
- */
- public static JsonObject getObject(JsonObject json, String key) {
- return json.getAsJsonObject(key);
- }
-
- /**
- * Utility method to simplify accessing a number as a double.
- * @param json JSON object containing the number to be got.
- * @param key Key of the number to be got.
- * @return Number with key {@code key} from {@code json}.
- */
- public static double getDouble(JsonElement json, String key) {
- return json.getAsJsonObject().get(key).getAsDouble();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/82fcaf6e/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java
deleted file mode 100644
index 689c00d..0000000
--- a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/ELM327Streams.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
-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 quarks.samples.connectors.elm327;
-
-import java.util.concurrent.TimeUnit;
-
-import com.google.gson.JsonArray;
-
-import quarks.samples.connectors.elm327.runtime.CommandExecutor;
-import quarks.connectors.serial.SerialDevice;
-import quarks.function.Supplier;
-import quarks.topology.TStream;
-
-/**
- * Streams fetching OBD-II data from an ELM327 through
- * a serial device.
- *
- * @see <a href="https://en.wikipedia.org/wiki/ELM327">ELM327</a>
- */
-public class ELM327Streams {
-
- /**
- * Periodically execute a number of ELM327 commands.
- * Each tuple on the returned stream is a JSON array containing
- * the result for each command.
- * <BR>
- * Each result is a JSON object containing the
- * {@link Cmd#id() command identifier} with key {@link Cmd#PID pid}
- * and any result set by the individual command, typically with
- * the key {@link Cmd#VALUE value}.
- *
- * @param device Serial device the ELM327 is connected to.
- * @param period Period to poll.
- * @param unit Unit of {@code period}.
- * @param cmds Commands to execute.
- * @return Stream containing the results of the command exections.
- */
- public static TStream<JsonArray> poll(SerialDevice device, long period, TimeUnit unit, Cmd ... cmds) {
-
- Supplier<JsonArray> data = device.getSource(
- port ->
- {
- JsonArray array = new JsonArray();
- for (Cmd cmd : cmds) {
- array.add(CommandExecutor.execute(cmd, port.getOutput(), port.getInput()));
- }
- return array;
-
- });
-
- return device.topology().poll(data, period, unit);
-
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/82fcaf6e/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Streams.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Streams.java b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Streams.java
new file mode 100644
index 0000000..94ec545
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/elm327/Elm327Streams.java
@@ -0,0 +1,70 @@
+/*
+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 quarks.samples.connectors.elm327;
+
+import java.util.concurrent.TimeUnit;
+
+import com.google.gson.JsonArray;
+
+import quarks.samples.connectors.elm327.runtime.CommandExecutor;
+import quarks.connectors.serial.SerialDevice;
+import quarks.function.Supplier;
+import quarks.topology.TStream;
+
+/**
+ * Streams fetching OBD-II data from an ELM327 through
+ * a serial device.
+ *
+ * @see <a href="https://en.wikipedia.org/wiki/ELM327">ELM327</a>
+ */
+public class Elm327Streams {
+
+ /**
+ * Periodically execute a number of ELM327 commands.
+ * Each tuple on the returned stream is a JSON array containing
+ * the result for each command.
+ * <BR>
+ * Each result is a JSON object containing the
+ * {@link Cmd#id() command identifier} with key {@link Cmd#PID pid}
+ * and any result set by the individual command, typically with
+ * the key {@link Cmd#VALUE value}.
+ *
+ * @param device Serial device the ELM327 is connected to.
+ * @param period Period to poll.
+ * @param unit Unit of {@code period}.
+ * @param cmds Commands to execute.
+ * @return Stream containing the results of the command exections.
+ */
+ public static TStream<JsonArray> poll(SerialDevice device, long period, TimeUnit unit, Cmd ... cmds) {
+
+ Supplier<JsonArray> data = device.getSource(
+ port ->
+ {
+ JsonArray array = new JsonArray();
+ for (Cmd cmd : cmds) {
+ array.add(CommandExecutor.execute(cmd, port.getOutput(), port.getInput()));
+ }
+ return array;
+
+ });
+
+ return device.topology().poll(data, period, unit);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-quarks/blob/82fcaf6e/samples/connectors/src/main/java/quarks/samples/connectors/obd2/Obd2Streams.java
----------------------------------------------------------------------
diff --git a/samples/connectors/src/main/java/quarks/samples/connectors/obd2/Obd2Streams.java b/samples/connectors/src/main/java/quarks/samples/connectors/obd2/Obd2Streams.java
new file mode 100644
index 0000000..50e1502
--- /dev/null
+++ b/samples/connectors/src/main/java/quarks/samples/connectors/obd2/Obd2Streams.java
@@ -0,0 +1,145 @@
+/*
+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 quarks.samples.connectors.obd2;
+
+import static java.util.concurrent.TimeUnit.MINUTES;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static quarks.analytics.math3.stat.Regression.SLOPE;
+import static quarks.analytics.math3.stat.Statistic.MAX;
+import static quarks.samples.connectors.elm327.Cmd.PID;
+import static quarks.samples.connectors.elm327.Cmd.VALUE;
+import static quarks.samples.connectors.elm327.Pids01.AIR_INTAKE_TEMP;
+import static quarks.samples.connectors.elm327.Pids01.ENGINE_COOLANT_TEMP;
+import static quarks.samples.connectors.elm327.Pids01.RPM;
+import static quarks.samples.connectors.elm327.Pids01.SPEED;
+
+import java.util.concurrent.TimeUnit;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+
+import quarks.analytics.math3.json.JsonAnalytics;
+import quarks.connectors.serial.SerialDevice;
+import quarks.samples.connectors.elm327.Elm327Streams;
+import quarks.topology.TStream;
+import quarks.topology.TWindow;
+
+/**
+ * Sample OBD-II streams.
+ *
+ */
+public class Obd2Streams {
+
+ /**
+ * Get a stream of temperature readings which
+ * are increasing over the last minute.
+ *
+ * Poll temperatures every five seconds and
+ * calculate the maximum reading and rate of change
+ * (slope) over the last minute, partitioned by parameter
+ * {@link quarks.samples.connectors.elm327.Cmd#PID pid}. Filter so that only
+ * those with a rate of increase greater than
+ * or equal to 1°C/minute is present on the returned stream.
+ *
+ * Temperatures included are
+ * {@link quarks.samples.connectors.elm327.Pids01#AIR_INTAKE_TEMP AIR_INTAKE_TEMP} and
+ * {@link quarks.samples.connectors.elm327.Pids01#ENGINE_COOLANT_TEMP ENGINE_COOLANT_TEMP}.
+ *
+ * @param device Serial device the ELM327 is connected to.
+ * @return Stream that will contain parameters with increasing temperatures.
+ */
+ public static TStream<JsonObject> increasingTemps(SerialDevice device) {
+
+ TStream<JsonArray> tempsA = Elm327Streams.poll(device, 5, SECONDS,
+ AIR_INTAKE_TEMP,
+ ENGINE_COOLANT_TEMP);
+
+ TStream<JsonObject> temps = tempsA.flatMap(je -> je).map(je -> je.getAsJsonObject());
+
+ TWindow<JsonObject, JsonElement> window = temps.last(1, MINUTES, j -> j.get(PID));
+
+ TStream<JsonObject> temperatureRate = JsonAnalytics.aggregate(window, PID, VALUE, MAX, SLOPE);
+
+ // Have the stream contain only tuples where
+ // the rise in temperatures >= 1°C/minute
+ temperatureRate = temperatureRate.filter(j -> {
+ JsonObject v = getObject(j, "value");
+ return v.has("SLOPE") && getDouble(v, "SLOPE") >= 1.0;
+ });
+
+ return temperatureRate;
+ }
+
+ /**
+ * Get a stream containing vehicle speed (km/h)
+ * and engine revs (rpm).
+ *
+ * {@link quarks.samples.connectors.elm327.Pids01#SPEED Speed}
+ * and {@link quarks.samples.connectors.elm327.Pids01#RPM engine revs}
+ * are polled every 200ms and returned as a stream
+ * containing JSON objects with keys {@code speed}
+ * and {@code rpm}.
+ *
+ * The two readings may not be exactly consistent with
+ * each other as there are fetched sequentially from
+ * the ELM327.
+ *
+ * @param device Serial device the ELM327 is connected to.
+ * @return Stream that will contain speed and engine revolutions.
+ */
+ public static TStream<JsonObject> tach(SerialDevice device) {
+
+ TStream<JsonArray> rpmSpeed = Elm327Streams.poll(device, 200, TimeUnit.MILLISECONDS,
+ SPEED, RPM);
+
+ TStream<JsonObject> tach = rpmSpeed.map(ja -> {
+ JsonObject j = new JsonObject();
+
+ double speed = getDouble(ja.get(0), VALUE);
+ double rpm = getDouble(ja.get(1), VALUE);
+ j.addProperty("speed", speed);
+ j.addProperty("rpm", rpm);
+
+ return j;
+ });
+
+ return tach;
+ }
+
+ /**
+ * Utility method to simplify accessing a JSON object.
+ * @param json JSON object containing the object to be got.
+ * @param key Key of the object to be got.
+ * @return JSON object with key {@code key} from {@code json}.
+ */
+ public static JsonObject getObject(JsonObject json, String key) {
+ return json.getAsJsonObject(key);
+ }
+
+ /**
+ * Utility method to simplify accessing a number as a double.
+ * @param json JSON object containing the number to be got.
+ * @param key Key of the number to be got.
+ * @return Number with key {@code key} from {@code json}.
+ */
+ public static double getDouble(JsonElement json, String key) {
+ return json.getAsJsonObject().get(key).getAsDouble();
+ }
+}