GPS Event [16] – Coral8 [4]

Time for an experiment. Let’s look at a way to detect that a sensor stopped streaming data. For the example I will use the same code as in the last test.

The idea is to monitor StreamIn and when it stops, insert a message into a new output stream. The output stream can then be connected to an output adapter and send the message "out there". So, here is the output stream named sAlarm, with three fields:

  • Entity – name of the device which stopped streaming data;
  • Msg – a message, like "Stream stopped";
  • AlmTime – time of detection.
-- Alarm stream
CREATE OUTPUT STREAM sAlarm
SCHEMA (
    Entity STRING,
    Msg STRING,
    AlmTime STRING
);

The AlmTime is formatted as a STRING, because I intend to send this message later via an email or similar adapter.

The detection technique is straight from the Coral8 CCL Cookbook; let’s see how it works.

-- Detect no data from StreamIn
INSERT INTO sAlarm
SELECT E1.Entity
	,'Stream stopped'
	,TO_STRING(GETTIMESTAMP(E1))
FROM StreamIn AS E1, StreamIn AS E2
	MATCHING [2 SECONDS: E1, !E2]
	ON E1.Entity = E2.Entity
;

The MATCHING clause needs at least two streams (think tables), hence  StreamIn is used twice with different aliases: E1 and E2. It is waiting for an event in E1, followed by a non-event from E2, within a time-window of two seconds. The GETTIMESTAMP(E1) function returns the timestamp of the last row from the stream, so the AlmTime is actually time of the last record received.

As in the previous example, I have sent ten numbers (0-9) and then the stream stopped. Notice that AlmTime matches the time-stamp of the last StreamIn entry.

To be continued.

GPS Event [15] – Coral8 [3]

In the previous post I have managed to send GPS data to Coral8. This time I would like to use Coral8 as a database-loading front end, with minimum changes to the already existing program. I already have MS SQL Server running on the same machine as Coral8, so this should not be too complicated.

Here is the to do list:

  1. Create a database and a table for GPS data;
  2. Create an ODBC connection to the database;
  3. Register the database as a service with Coral8, using the ODBC connection;
  4. Restart Coral8 Server;
  5. Attach a write-to-database output adapter to the output stream;
  6. Test.

Step 1 – Create a database

Using MS SQL Server Manager Studio I have created a new database named DamirTest and and a table GpsData.

USE DamirTest
CREATE TABLE dbo.GpsData(
	Entity varchar(25) NOT NULL
	,Latitude float
	,Longitude float
	,Course float
	,Speed float
	,TimeUTC datetime NOT NULL
)

As you can see, the idea is to match the record of the output stream (see the previous post).

Step 2 – Create an ODBC connection

Coral8 communicates with MS SQL Server using an ODBC connection, so one has to be added for the database. The dialog can be accessed via Start >>> Administrative Tools >>> Data Sources (ODBC).  For SQL Server 2005 — and latter — specify SQL Native Client for the driver.



Step 3 – Register the database as a service with Coral8

Coral8 remote services are listed in the coral8-services.xml file. The file can be found in /Coral8/Server/conf/ directory. I had to add the following segment to the file:

<service name="DamirTest" type="DATABASE">
    <description>Connection to DamirTest</description>
    <param name="DBDriverType">DBDriverODBC</param>
    <!-- DBDriverConnectString = DSN name -->
    <param name="DBDriverConnectString">DamirTest</param>
    <param name="DBReadNoCommit">True</param>
    <!-- Username and Password -->
    <param name="Username">my_username_here</param>
    <param name="Password">my_password_here</param>
</service>

Step 4 – Restart Coral8 Server

Coral8 reads configuration files on start-up, so the server has to be restarted. There is no need to re-boot the machine, stop and start the program only.

Step 5 – Attach a write-to-database output adapter

No need to remember this syntax. I have simply used point and click to attach the adapter to the output stream — I did have to enter the query in the adapter form.

ATTACH OUTPUT ADAPTER ToDB 
    TYPE WriteToDBAdapterType
    TO STREAM StreamOut
PROPERTIES
    DBNAME = "DamirTest",
    QUERY  = [[
INSERT INTO [DamirTest].[dbo].[GpsData]
    (Entity ,Latitude ,Longitude
    ,Course ,Speed ,TimeUTC)
VALUES
    (?Entity ,?Latitude ,?Longitude
    ,?Course ,?Speed ,?TimeUTC);
 ]],
    COMMITFREQUENCY = "0"
;

Note how all stream field names start with a question mark. They are positionally mapped to the list of database fields; so it is not really necessary to use same names for database and stream fields.

Step 6 – Test

This is all it takes, the rest of the code from the previous post remains unchanged. As in the last example, I have sent ten GPS readings to Coral8. Here they are in the output stream,

and in the database.

To be continued.

GPS Event [14] – Coral8 [2]

Ever heard about that hotel that didn’t have the 13th floor? Empty superstition, I don’t believe in things like that. Now where was I? Yes, GPS data to Coral8 — here we go.

Having done my homework in the previous post, this should not be too complicated. For the template I have chosen: entity, latitude, longitude, course, speed and the time stamp. The first thing I noticed is that my time format does not match the format used by Coral8. After few trail and error attempts, I have decided to transfer the time stamp to the server as a string, and later convert it to the time format. For this I needed an additional local stream.

-- Input from GPS
CREATE INPUT STREAM GpsIn
SCHEMA (
    Entity    STRING
    ,Latitude  FLOAT
    ,Longitude FLOAT
    ,Course FLOAT
    ,Speed FLOAT
    ,TimeUTC STRING
);
 
-- Local formatted stream with
-- TimeUTC as TIMESTAMP
CREATE LOCAL STREAM GpsFmtd
SCHEMA (
    Entity    STRING
    ,Latitude  FLOAT
    ,Longitude FLOAT
    ,Course FLOAT
    ,Speed FLOAT
    ,TimeUTC TIMESTAMP
);
 
-- Output stream
CREATE OUTPUT STREAM StreamOut
SCHEMA (
    Entity    STRING
    ,Latitude  FLOAT
    ,Longitude FLOAT
    ,Course FLOAT
    ,Speed FLOAT
    ,TimeUTC TIMESTAMP
);

The only purpose of the local GpsFmtd stream is to allow conversion of the TimeUTC field from a string to the proper time format. With streams in place, I can add queries to connect them.

-- From Input to Local
INSERT INTO GpsFmtd
SELECT
    Entity ,Latitude ,Longitude
    ,Course ,Speed
    ,TO_TIMESTAMP(TimeUTC)
FROM GpsIn;
 
-- From Local to Output
INSERT INTO StreamOut
SELECT * FROM GpsFmtd;

Here is the flow view.

This is all for the Coral8 side, now the Python code. Being on my local network, there is no need for an authorization module. I will need only template and service modules:

As in all previous examples, the service has to be activated in the main gps_event.py module:

# select service: none, aws, rc, tweet, email, gtalk, coral8, wp, blog
use_service = 'coral8'
 
# define sampling delay in seconds
sample_delay = 5

And here is the result. You may notice different display of the TimeUTC field. The top one is a string — as sent to the server — while the bottom one is in the Coral8 TIMESTAMP format.

In Coral8 terminology this would be called a GPS out-of-process adapter.

To be continued.

GPS Event [12] – Coral8

Let’s see what does it take to send a message to Coral8. The latest available version on the website is 5.5 and I have 5.1 installed, so it is time for an upgrade.

The upgrade (installation) went smooth. There is a good "getting started guide" available too.  Coral8 Server is the engine of the application and Coral8 Studio is a developer’s GUI. I have installed both on a Windows Server 2003 for Small Business — 2.53GHz, 2.5GB RAM — which also runs a developer version of MS SQL Server 2005. The average CPU usage seems to be below 5% when idle, not bad.

Coral8 has quite a few adapters included, but none of them seems to fit my GPS example; I will have to find another way. There is a SDK for several languages — including Python — so this seems a way to go. Now I can sketch a road map:

  1. Create a simple Coral8 project;
  2. Study the Python SDK and figure out how to send a message to Coral8;
  3. Test;
  4. If it works, continue with the GPS example.

Step 1 – a simple Coral8 project

Coral8 CCL (Continuous Computation Language) is SQL like, so I have created two streams (think one-row tables) with just two columns: Entity and Val. Imagine a measurement device (Entity) sending a number (Val) each measurement. Streams are connected by a query which copies data from the input to the output.

-- Input
CREATE INPUT STREAM StreamIn
SCHEMA (
    Entity STRING
    ,Val INTEGER
);
 
-- Output
CREATE OUTPUT STREAM StreamOut
SCHEMA (
    Entity STRING
    ,Val INTEGER
);
 
-- Pass Through
INSERT INTO StreamOut
SELECT * FROM StreamIn;

This is how it looks in the Coral8 Flow View.

If I manage to insert data into the input stream, the same data should show up in the output stream.

Step 2 – the Python SDK

The SDK has an example which sends and receives data, and also includes Python multitasking. After some tinkering I have decided to simplify it a bit:

"""
c8_test_1.py
Test sending data to Coral8
"""
import time
from coral8 import Coral8
 
uri = 'ccl://damir5:6789/Stream/Default/'
uri += 'TestPythonInAdapter/StreamIn'
 
col_names = ['Entity','Val']
 
def send_to_c8(j):
    pb = Coral8.Publisher(uri)
    tp = Coral8.Tuple(col_names)  
    tp.setvalue('Entity', 'DS_1')
    tp.setvalue('Val', j)
    pb.write_tuple(tp)
 
if __name__ == '__main__':
    for j in range(10):
        send_to_c8(j)
        time.sleep(0.2)

The program is supposed to send ten numbers (0-9) to Coral8 Server. Note the uri variable, I got that one by inspecting the StreamIn properties and substituting “localhost” with the server name (damir5).

Step 3 – Test

Guess what, it works; just like that. Here is a screen capture of both stream viewers.

Step 4 – Continue

Now I can add service and template modules for Coral8 to my GPS Event example. I still have to figure out what to do with events inside the server. Maybe I can use it to write events to a database and send notifications when the GPS starts loosing satellite signals — we’ll see.

To be continued.

Disclosure:

Everything I write about Coral8 is my own opinion and experience.
There is no contract nor any kind of active business relationship between Coral8 and me (Damir Systems Inc).