@Plan:name('Android-Activity-ExecutionPlan')

@Plan:description('Identify the start and end of each activities')

@Import('iot.per.device.stream.android.activity:1.0.0')
define stream ActivitySream (meta_owner string, meta_deviceId string, meta_timestamp long, action string);

@Export('iot.per.device.stream.android.activity.stats:1.0.0')
define stream ActivityStatsStream (meta_owner string, meta_deviceId string, meta_timestamp long, action string, duration long, year int, month int, day int, hour int, minute int);

partition with (meta_deviceId of ActivitySream)
begin
    from every preActivity = ActivitySream, postActivity = ActivitySream[preActivity.action != action AND preActivity.meta_timestamp <= meta_timestamp]
    select preActivity.meta_owner as meta_owner, preActivity.meta_deviceId as meta_deviceId,
    preActivity.action as preAction, postActivity.action as postAction, preActivity.meta_timestamp as preTimestamp, postActivity.meta_timestamp as postTimestamp
    insert into #ActivityChangeEventStream;

    from every preEvent = #ActivityChangeEventStream, postEvent = #ActivityChangeEventStream[preEvent.postAction == preAction]
    select preEvent.meta_owner as meta_owner, preEvent.meta_deviceId as meta_deviceId, postEvent.preTimestamp as meta_timestamp, preEvent.postAction as action, (postEvent.postTimestamp - preEvent.postTimestamp) as duration,
	time:extract(postEvent.preTimestamp, 'year') as year, time:extract(postEvent.preTimestamp, 'month') as month, time:extract(postEvent.preTimestamp, 'day') as day, time:extract(postEvent.preTimestamp, 'hour') as hour, time:extract(postEvent.preTimestamp, 'minute') as minute
    insert into ActivityStatsStream;
end;