GPS Event [17] – Coral8 [5]

Last time we detected when a device stops streaming data, however the message remained within Coral8. Today, I would like to attach an output adapter to send the message out. Coral8 comes with several output adapters, including XML over HTTP. This one seems appropriate for any kind of server based application. Just for fun, I decided to transfer data to an application running on Google App Engine.

Note:
Starting with Google App Engine is fairly easy; detailed instructions are available, so I will not describe it here.

The to-do list:

  1. Add a XML over HTTP output adapter to the Coral8 example;
  2. Write an application on Google App Engine to receive, store and display Coral8 messages;
  3. Test.

1. The Output Adapter

--Adapter for GoogleAppEngine
ATTACH OUTPUT ADAPTER WriteXMLOverHTTP 
    TYPE WriteXmlOverHttpAdapterType
TO STREAM sAlarm
PROPERTIES
 URL = "//192.168.xxx.76:8080",
 XSLTEMPLATE = 
[[<xsl:stylesheet xmlns:xsl="//www.w3.org/1999/XSL/Transform" version="1.0">
<xsl:output method="text" />
<xsl:template match="/">
<xsl:text>{</xsl:text>
<xsl:for-each select="*">
<xsl:for-each select="*">
<xsl:text>'</xsl:text>
<xsl:value-of select="@Name" />
<xsl:text>':'</xsl:text>
<xsl:value-of select="." />
<xsl:text>',</xsl:text>
</xsl:for-each>
</xsl:for-each>
<xsl:text>}</xsl:text>
</xsl:template>
</xsl:stylesheet>
]]
;

 

The adapter has two properties:

  • URL of the server, including the port number;
  • XSLTEMPLATE for an optional XSL transformation.

If the XSLTEMPLATE is not specified, Coral8 sends a raw XML string like this one:

<tuple xmlns="//www.coral8.com/cpx/2004/03/" timestamp="1232757671788000">
	<field name="Entity">DS_1</field>
	<field name="Msg">Stream stopped</field>
	<field name="AlmTime">2009-01-23 19:41:09.788000</field>
</tuple>

 

The template in the adapter transforms the XML to a dictionary-like string:
{'Entity':'DS_1','Msg':'Stream stopped','AlmTime':'2009-01-23 19:41:09.788000',}.
Using the XSL I have simply avoided parsing the XML on the server side.

2. The Application

"""
Capture Coral8 messages to DB.
Display last 10 messages.
"""
from google.appengine.ext import webapp
from google.appengine.ext.webapp.util \
     import run_wsgi_app
from google.appengine.ext import db
 
class Dta(db.Model):
  content = db.StringProperty(multiline=True)
  date = db.DateTimeProperty(auto_now_add=True)
  
class MainPage(webapp.RequestHandler):
  def get(self):
    self.response.headers \
      ['Content-Type'] = 'text/plain'
    dt = db.GqlQuery \
      ("SELECT * FROM Dta ORDER BY date DESC LIMIT 10")
    for d in dt:
      try:
        dic = eval(d.content)
        self.response.out.write('\n%s' % ('-' * 40))
        for k in dic.keys():
          self.response.out.write \
            ('\n%s = %s' % (k, dic[k]))
      except:
        pass
 
  def post(self):
    rec = Dta()
    doc = str(self.request.body)
    rec.content = ' '.join(doc.split())
    rec.put()
     
application = webapp.WSGIApplication \
              ([('/', MainPage)], debug=True)
 
def main():
  run_wsgi_app(application)
 
if __name__ == "__main__":
  main()

May seem complicated, but here is the basics:

class Dta(db.Model) defines a “database table” with two columns:

  • content – a string column for Coral8 data;
  • date – a timestamp column;

post(self) method is activated when Coral8 sends data to the application’s web page;

get(self) method is activated when a browser visits the web page.

3. Test

For testing, I have used the same example as in the previous post — send several messages to Coral 8 and then stop. Here is the result in a browser window:

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.