ATLaS and CQL Stream Example: a Military Logistics Application

Introduction

This page provides data stream schema and query specifications from a military logistics application. Imagine a wartime scenario where military personnel and equipment such as tanks and missiles are deployed over a wide area. The military high command stationed at an on-site operations center would like to keep track of troop movements, equipment health, and enemy positions in real-time. The schema and queries were provided by members of the Aurora Project. The CQL queries are from Stanford Stream Query Repository.

Schema

Any mobile unit that should be tracked is called a platform (e.g., soldier, tank). Each platform has various sensors for different kind of measurements. Further, there are stationary sensors that perform various measurements but are not associated with any platform. The schema consists of streams for measurements from the various sensors. The schema also consists of a relation that contains details of each platform.

PlatformSpec: Relation containing information about a platform. Presently, this relation identifies the type of a platform (e.g., soldier, tank) and the platoon to which it belongs.

PlatformSpec(PRN                /* platform reference number */,
             type               /* type of the platform      */,
             platoonId          /* platoon to which belongs  */);

PlatformPositionReport: Continuous stream of a platform's position.

PlatformPositionReport(PRN       /* platform reference number */,
                       latitude  /* latitude of current position */, 
                       longitude /* longitude of current position */,  
                       timestamp /* time of measurement */);

PlatformStatusReport: Stream of other platform-specific information: speed, hull orientation, turret orientation, fuel status, ammunition status, engine status, blood pressure (for soldier), heart-rate (for soldier).

PlatformStatusReport(PRN       /* platform reference number */,
                     type      /* identifies what is being measured, 
                                  e.g., SPEED, FUEL-STATUS */,            
                     value     /* the measured value */, 
                     timestamp /* time of measurement */);

RFSensorPositionReport: Stream of speed measurements of other platforms measured by the RF Sensor of a platform.

RFSensorPositionReport(PRN       /* PRN of reporting platform */,
                       latitude  /* latitude of current position of 
                                    observed platform */, 
                       longitude /* longitude of current position of 
                                    observed platform */,  
                       speed     /* speed of observed platform */
                       timestamp /* time of observation */);

MagneticSensorPositionReport: Stream of measurements of the positions of other platforms measured by the Magnetic sensor of a platform.

MagneticSensorPositionReport(PRN       /* PRN of reporting platform */,
                             latitude  /* latitude of current position of 
                                          observed platform */, 
                             longitude /* longitude of current position of 
                                          observed platform */,  
                             direction /* direction of movement of
                                          observed platform */
                             timestamp /* time of observation */);

RadarSensorPositionReport: Stream of measurements of the positions of other platforms by the radar sensor of a platform.

RadarSensorPositionReport(PRN       /* PRN of reporting platform */,
                          latitude  /* latitude of current position of 
                                       observed platform */, 
                          longitude /* longitude of current position of 
                                       observed platform */,  
                          timestamp /* time of observation */);

InfraredTriggeredSensorReport:

InfraredTriggeredSensorReport (vfindex   /*  video-frame index */, 
                               coords    /* "heat/position coordinates" */, 
                               timestamp /* time of observation */);

MagneticTriggeredSensorReport:

MagneticTriggeredSensorReport (dpindex   /* delta-permeability index */, 
                               comp      /* "composition" */, 
                               timestamp /* time of observation */);

SeismicTriggeredSensorReport:

SeismicTriggeredSensorReport  (dindex    /* distance index */,  
                               range     /* "range" */,
                               timestamp /* time of observation */);

WeatherAlert: Stream of measurements from stationary sensors that monitor weather.

WeatherAlert(latitudeL,  /* latitude of the lower left corner of the location */
             longitudeL, /* longitude of the lower left corner of the location */
             latitudeR,  /* latitude of the upper right corner of the location */
             longitudeR, /* longitude of the upper right corner of the location */ 
             type        /* type of weather---RAIN, SNOW, etc. */,
             amount      /* value in inches per hour */, 
             timestamp   /* time when reported weather is expected */);

Queries

Center of Mass Query: Report the center-of-mass for a platoon of soldiers every time it moves by 10 meters.

ATLaS

aggregate center_of_mass(iPRN, iLatitude, iLongitude):(oLatitude, oLongitude){
table memo(PRN, latitude, longtitude);  /* Current positions of platforms */
table counter(c int) memory as values(0);    /* # of platforms in the platoon */
table result(rLatitude, rLongitude) as values(0,0);
table lastResult(lastLatitude, lastLongitude) as values(0,0);
initialize:iterate:{
  /* Update result by subtracting the old values and adding the new ones */
  update memo, result, counter
  set rLatitude = (rLatitude * c - latitude + iLatitude) / c,
      rLongitude = (rLongitude * c - longtitude + iLongitude) / c,
      latitude = iLatitude, 
      longtitude = iLongitude
  where iPRN = PRN;
  /* Adding the new platforms into the result */
  insert into memo values(iPRN, iLatitude, iLongitude)
  where SQLCODE > 0;
  update result, counter
  set rLatitude = (rLatitude * c + iLatitude) / (c+1),
      rLongitude = (rLongitude * c + iLongitude) / (c+1),
      c = c + 1
  where SQLCODE = 0;
  insert into return
    select rLatitude, rLongitude
    from result, lastResult
    where dist(rLatitude, rLongtitude, lastLatitude, lastLongitude) < 10;
  update lastResult, result
  set lastLatitude = rLatitude, lastLongitude = rLongitude;
}
};

select center_of_mass(PRN, latitude, longitude)
from PlatformPositionReport
where platoonID = 10;

Comments We assume center_of_mass (latitude, longitude) is the average of (latitude, longtitude) of all platforms. In aggregate center_of_mass, the "memo" table stores current positions of platforms. We can use RTREE index on this table. The "counter" table keep an updated counter on "memo". The "result" table stores the result while "lastResult" table is for comparison purpose. The user-defined function (UDF) dist() computes the distance between current and last center_of_mass.

CQL:

Q1: Select  Istream(center-of-mass(R1.latitude, R1.longitude) as com)
    From    PlatformSpec P, PlatformPositionReport [Range 5 Minute] R1
    Where   P.platoonID = 10 and P.PRN = R1.PRN and 
            R1.timestamp >= All (Select timestamp 
                                 From   PlatformPositionReport 
                                        [Range 5 Minute] R2
                                 Where  R2.PRN = R1.PRN)

Q2: Select Istream(delta_compr(com)
    From   Q1

Comments: center-of-mass is a user-defined relational-aggregate for computing the center-of-mass. We consider the latest position report by each platform for computing the center-of-mass within the last 5 minutes. So if a platform has not reported its position in the last 5 minutes, it is not considered for the center-of-mass computation. Query Q1 continuously streams the center-of-mass and query Q2 streams the new center-of-mass every time it moves by 10 meters. delta_compr is a user-defined windowed-aggregate (see link for another example of delta compression).

Object Sighting Query: Report every time three observations are received from three different sources within a five minute interval of each other and the reported positions differ by not more than ten meters from each other.

ATLaS

select *
from RFSensorPositionReport O1,
     MagneticSensorPositionReport 
       over (range 5 minute preceding O1 range 5 minute following O1) O2,
     RadarSensorPositionReport 
       over(range 5 minute preceding O1 range 5 minute following O1) O3
where ABS(O2.timestamp, O3.timestamp) < 5 minute
      and distance(O1.latitude,O1.longitude,O2.latitude,O2.longitude) <= 10
      and distance(O1.latitude,O1.longitude,O3.latitude,O3.longitude) <= 10
      and distance(O2.latitude,O2.longitude,O3.latitude,O3.longitude) <= 10;

CQL:

ObservationStream:
  (Select PRN, latitude, longitude
   From   RFSensorPositionReport)
  Union
  (Select PRN, latitude, longitude
   From   MagneticSensorPositionReport)
  Union
  (Select PRN, latitude, longitude
   From   RadarSensorPositionReport)

ObjectSightingStream:
  Select  Rstream(*)
  From    ObservationStream [Now] O1,
          ObservationStream [Range 5 Minute] O2,
          ObservationStream [Range 5 Minute] O3
  Where   O1.PRN <> O2.PRN and O1.PRN <> O3.PRN and O2.PRN <> O3.PRN
          and distance(O1.latitude,O1.longitude,O2.latitude,O2.longitude) <= 10
          and distance(O1.latitude,O1.longitude,O3.latitude,O3.longitude) <= 10
          and distance(O2.latitude,O2.longitude,O3.latitude,O3.longitude) <= 10

Comments: ObservationStream is the derived stream of all observations from all sources. The object sighting query is written using the derived observation stream. The object sighting query checks for each observation (specified using the Now window) if there were two other observations in the last 5 minutes that satisfy the distance criteria. We use the user-defined function distance to compute the distance between two observations.

Boundary Crossing Query: Report when at least 40 soldiers have crossed a specified boundary.

ATLaS

aggregate online_count_distinct(iKey, latitude, longitude):(oCount int){
table memo(key);
table counter(c int) memory as values(0);
initialize:iterate:{
  insert into memo values(iKey)
  where  hasCrossed(latitude,longitude,500,600,550,650) and 
         not exists (select key from memo where key = iKey);
  update counter
  set c = c + 1
  where SQLCODE = 0;
  delete from memo
  where hasReturned(latitude,longitude,500,600,550,650) and
        exists (select key from memo where key = iKey);
  update counter
  set c = c - 1
  where SQLCODE = 0;
  insert into return
    select c from counter
}
};

select 'At least 40 soldiers have crossed a specified boundary!'
from PlatformPositionReport as R, PlatformSpecs as S
where S.type = 'SOLDIER' and S.PRN = R.PRN
having online_count_distinct(R.PRN) >= 40;

Comments: Since we can't use blocking aggregate count() on stream, we use User-defined Aggregate (UDA) online_count_distinct(). hasCrossed() and hasReturned() are 2 UDF to check if the soldier has crossed/returned the specified boundary, respectively.

CQL:

SoldierPositionStream:
   Select R.PRN as PRN, R.latitude as latitude, R.longitude as longitude
   From   PlatformPositionReport [Now] as R, PlatformSpecs as S
   Where  S.type = 'SOLDIER' and S.PRN = R.PRN

BoundaryCrossing:
  Select True
  From   (Select Count (Distinct PRN) as num
          From   SoldierPositionStream
          Where  hasCrossed(latitude,longitude,500,600,550,650))
  Where   num >= 40

Comments: SoldierPositionStream is the derived stream of positions of soldiers. The subquery in the From clause of the boundary crossing query computes the distinct number of soldiers who have ever crossed the boundary. This subquery uses a user-defined boolean function hasCrossed to check if a position is beyond the boundary. Note that the above query counts every soldier who has ever crossed the boundary, even if he crosses back subsequently. Therefore, it is possible that there are less than 40 soldiers beyond the boundary at the logical instant the query returns true. The following query reports the first instant when at least 40 people are simultaneously beyond the boundary.

Select True
From   (Select Count (Distinct (P.PRN)) as num
        From   PlatformPositionReport [Partition By PRN Rows 1] as P,
	       PlatformSpecs as S
        Where  P.PRN = S.PRN and S.type = 'SOLDIER' and
               hasCrossed(latitude,longitude,500,600,550,650))
Where  num >= 40

Front Line Query: Periodically compute the front line of a platoon. We assume the front line is a latitude value. It's the average of top-10 largest latitude values in the platoon;

ATLaS

aggregate top_k(iValue, k int):{
table memo(value) memory;
initialize:iterate:{
  insert into memo values(iValue)
  where exists (select value from memo 
                where iValue >=value) or
        (select count(value) from memo) < k;
  delete from memo
  where value = (select min(value) from memo) and 
        (select count(value) from memo) > k;
}
time:{
  insert into return 
    select * from memo;
}
};

select top_k(latitude, 10)
over (slide 1)
from PlatformPositionReport
where platoonID = 10;

CQL:

Fratricide Situation Query: Report whenever a tank is within 1000 meters of a friendly platform and the turret is pointed within 20 degrees of the platform.

ATLaS

/* maintain a table of current platforms' status */
table CurrentPlatformStatus(PRN, type, value);
insert into CurrentPlatformStatus C
  select R.PRN, R.latitude, R.longitude, R.value
  from PlatformStatusReport R
  where not exists (select PRN from C
                    where C.PRN = R.PRN); 
update CurrentPlatformStatus C, PlatformStatusReport R
set C.value = R.value, C.type = R.type
where C.PRN = R.PRN;

select *
from PlatformPositionReport P, PlatformSpecs PS, PlatformSpecs TS, 
     CurrentPlatformStatus R, 
     PlatformPositionReport T 
       over (range 10 minute preceding P range 10 minute following P)
where T.PRN = TS.PRN and TS.type = 'TANK' 
      and T.PRN = C.PRN and C.type = 'TURRET-ORIENTATION'
      and P.PRN = PS.PRN and TS.platoonID = PS.platoonID
      and distance(T.latitude, T.longitude, P.latitude, P.longitude) <= 1000
      and angle(C.value, T.latitude, T.longitude, P.latitude, P.longitude) <=20;

CQL:

Q1: Select *
    From   PlatformPositionReport [Range 10 Minute] R1
    Where  R1.timestamp >= All (Select timestamp 
                                From   PlatformPositionReport 
                                       [Range 10 Minute] R2
                                Where R2.PRN = R1.PRN)
Q2: Select *
    From   PlatformStatusReport [Range 10 Minute] R1
    Where  R1.timestamp >= All (Select timestamp 
                                From PlatformStatusReport 
                                     [Range 10 Minute] R2
                                Where R2.PRN = R1.PRN)

Q3: Select Istream(R1.PRN, R1.latitude, R1.longitude,
                   R2.PRN, R2.latitude, R2.longitude)
    From PlatformSpecs S1, Q1 R1, Q1 R2, Q2 T, PlatformSpecs S2
    Where S1.PRN = R1.PRN and S1.type = 'TANK' and R1.PRN != R2.PRN 
          and S2.PRN = R2.PRN and S1.platoonID = S2.platoonID
          and distance(R1.latitude,R1.longitude,R2.latitude,R2.longitude) <= 1000
          and T.PRN = R1.PRN and T.type = 'TURRET-ORIENTATION' and 
          angle(T.value,R1.latitude,R1.longitude,R2.latitude,R2.longitude) <= 20

Comments: We assume that the reports used to detect a fratricide situation must be within 10 minutes of each other. Among the reports within the last 10 minutes, we consider only the latest report. angle(turretOrientation, latitude1, longitude1, latitude2, longitude2) is a user-defined function that takes the turret orientation of the tank and the positions of the two platforms as input and returns the separation degree between the turret and the second platform. Further, since there is no other relevant information in the schema, we assume that two platforms are friendly if they belong to the same platoon.

Platform Report Query: Maintain a summary of the latest information available for each tank and update it every 10 minutes.

ATLaS

table TankSummary (PRN, latitude, longitude, fuel, ammo);

insert into TankSummary TS, PlatformSpecs S
  select R.PRN, -1, -1, R.value, -1
  from PlatformStatusReport R
  where R.PRN = S.PRN and S.type = 'TANK' and R.type = 'FUEL-STATUS'
        and not exists (select PRN from TS 
                        where TS.PRN = R.PRN);
update TankSummary TS, PlatformSpecs S, 
       PlatformStatusReport R
set TS.fuel = R.value
where TS.PRN = S.PRN and S.type = 'TANK'
      and TS.PRN = R.PRN and R.type = 'FUEL-STATUS';

insert into TankSummary TS, PlatformSpecs S
  select R.PRN, -1, -1, -1, R.value
  from PlatformStatusReport R
  where R.PRN = S.PRN and S.type = 'TANK' and R.type = AMMO-STATUS'
        and not exists (select PRN from TS 
                        where TS.PRN = R.PRN);
update TankSummary TS, PlatformSpecs S, 
       PlatformStatusReport R
set TS.ammo = R.value
where TS.PRN = S.PRN and S.type = 'TANK'
      and TS.PRN = R.PRN and R.type = AMMO-STATUS';

insert into TankSummary TS, PlatformSpecs S
  select R.PRN, R.latitude, R.longitude, -1, -1
  from PlatformPositionReport R
  where R.PRN = S.PRN and S.type = 'TANK' 
        and not exists (select PRN from TS 
                        where TS.PRN = R.PRN);
update TankSummary TS, PlatformSpecs S, 
       PlatformPositionReport R
set TS.latitude = R.latitude, TS.longitude = R.longitude
where TS.PRN = S.PRN and S.type = 'TANK' and TS.PRN = R.PRN;

Comments: We could add a window "over (range 10 minute slide 10 minute)" onto the stream PlatformStatusReport and PlatformPosistionReport, to "update every 10 minute". But it would be more effecient to update for every coming tuple, in which case we don't need to go back to query the window. Therefore we decide to update "TankSummary" table in this way. Users can output the result every 10 minutes.

CQL:

Q1: Select  Rstream(R1.PRN PRN, R1.value fuelStatus)
    From    PlatformStatusReport [Range 10 Minute
                                  Slide 10 Minute] R1, 
            PlatformSpecs S
    Where   R1.type = 'FUEL-STATUS' and S.PRN = R1.PRN and S.type = 'TANK'
            and R1.timestamp >= All (Select timestamp  
                                     From   PlatformStatusReport 
                                            [Range 10 Minute] R2
                                     Where  R2.PRN = R1.PRN and 
                                            R2.type = 'FUEL-STATUS')

Q2: Select  Rstream(R1.PRN PRN, R1.value ammoStatus)
    From    PlatformStatusReport [Range 10 Minute
                                  Slide 10 Minute] R1, 
            PlatformSpecs S
    Where   R1.type = 'AMMO-STATUS' and S.PRN = R1.PRN and S.type = 'TANK'
            and R1.timestamp >= All (Select timestamp 
                                     From   PlatformStatusReport 
                                            [Range 10 Minute] R2
                                     Where R2.PRN = R1.PRN and 
                                           R2.type = 'AMMO-STATUS')

Q3: Select  Rstream(R1.PRN PRN, R1.latitude latitude, R1.longitude longitude)
    From    PlatformPositionReport [Range 10 Minute
                                    Slide 10 Minute] R1, 
            PlatformSpecs S
    Where   S.PRN = R1.PRN and S.type = 'TANK' and 
            R1.timestamp >= ALL (Select timestamp 
                                 From   PlatformPositionReport 
                                        [Range 10 Minute] R2
                                 Where  R2.PRN = R1.PRN)
            
Q4: Select  *
    From    Q1 [Partition By PRN Rows 1],
            Q2 [Partition By PRN Rows 1],
            Q3 [Partition By PRN Rows 1]
    Where   Q1.PRN = Q2.PRN and Q2.PRN = Q3.PRN

Comments: We assume that the summary information for a tank consists of its position, fuel status, and ammunition status. Every ten minutes, query Q1 streams the latest fuel status, query Q2 streams the ammunition status, and query Q3 streams the latest position, respectively, for each tank. Query Q4 uses partitioned windows to maintain the latest available information for each tank.

Probabilistic Enemy Tracking Query: Determine the approximate probability that multiple enemy vehicle positions reported are actually the same vehicle.
ATLaS
CQL:
Prediction Query: Based on the history of movement of a specified platform, determine the likelihood of the platform crossing a specified border within a specified period of time.
ATLaS
CQL:
Platoon SITREP Query: Roll up the summary status computed for each platform into a situation report for the platoon.
ATLaS
CQL:
Platoon Trail Query: Draw a trail for a particular platform over the past 4 hours.
ATLaS
CQL:
Target Monitoring Query: For a specified list of enemy targets, maintain the best possible weapon system that can destroy the target. The system should respond in real-time to a command to fire at any target in the specified list. Targets have priorities associated with them.
ATLaS
CQL:
Continuous Guidance Query: For a particular vehicle that is driving towards its destination, alert the commander on obstacles that can cause delays. Obstacles include weather warnings, threat of the route coming under indirect fire, etc.
ATLaS
CQL:
Multi-sensor Fusion Query: Combine directional (e.g., infra-red-triggered sensors) and omni-directional (e.g., seismic-triggered and magnetic-triggered sensors) unattended ground sensor streams with dissimilar indexes.

Reference

Stanford Stream Query Repository, http://www-db.stanford.edu/stream/sqr/cql/milmon.html

Aurora Project, http://www.cs.brown.edu/research/aurora/