Mongodb aggregation query to subtract and grouping of cumulative value

Question

{
"_id" : ObjectId("58f5a22d22679039176d2ee8"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:01:01.000+05:30"), 
"Utilization" : NumberInt("63654480"),
"RunStatus" : NumberInt("1"),   
"ProductsCount" : NumberInt("681350")   
},
{
"_id" : ObjectId("58f5a22d22679039176d2ee9"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:02:02.000+05:30"), 
"Utilization" : NumberInt("63655480"),
"RunStatus" : NumberInt("1"),   
"ProductsCount" : NumberInt("681370")   
},
{
"_id" : ObjectId("58f5a22d22679039176d2eea"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:03:02.000+05:30"), 
"Utilization" : NumberInt("63656480"),
"RunStatus" : NumberInt("0"),   
"ProductsCount" : NumberInt("681390")   
},
{
"_id" : ObjectId("58f5a22d22679039176d2eeb"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:04:02.000+05:30"), 
"Utilization" : NumberInt("63657480"),
"RunStatus" : NumberInt("1"),   
"ProductsCount" : NumberInt("681420")   
},
{
"_id" : ObjectId("58f5a22d22679039176d2eec"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:05:02.000+05:30"), 
"Utilization" : NumberInt("63658480"),
"RunStatus" : NumberInt("1"),   
"ProductsCount" : NumberInt("681450"),  
},
{
"_id" : ObjectId("58f5a22d22679039176d2eed"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:06:02.000+05:30"), 
"Utilization" : NumberInt("63659480"),
"RunStatus" : NumberInt("1"),   
"ProductsCount" : NumberInt("681470")   
},
{
"_id" : ObjectId("58f5a22d22679039176d2eee"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:07:02.000+05:30"), 
"Utilization" : NumberInt("63659780"),
"RunStatus" : NumberInt("0"),   
"ProductsCount" : NumberInt("681490")   
},
{
"_id" : ObjectId("58f5a22d22679039176d2eef"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:08:03.000+05:30"), 
"Utilization" : NumberInt("63659880"),
"RunStatus" : NumberInt("1"),   
"ProductsCount" : NumberInt("681525")   
},
{
"_id" : ObjectId("58f5a22d22679039176d2ef0"),
"MachineID" : NumberInt("1001"),
"Timestamp" : ISODate("2017-04-18T07:09:03.000+05:30"), 
"Utilization" : NumberInt("63659980"),
"RunStatus" : ("0"),
"ProductsCount" : NumberInt("681563")
}

From the above collection, Utilization and ProductsCount are cumulative values and incremental over time.

Need to subtract Utilization of current row with Utilization of next row which is sorted in ascending order. As well as same operation for ProductsCount based on RunStatus.

If RunStatus of current row is 1 and next row is 0, then the difference in Utilization and ProductsCount should be mapped to RunStatus of next row which is 0.

Then group it based on MachineID and RunStatus

Expected Result

/* 1 */
{   
"MachineID" : 1001,
"RunStatus" : 1,
"Utilization" : 4100,
"ProducedCount" : 135
},

/* 2 */
{   
"MachineID" : NumberInt("1001"),
"RunStatus" : NumberInt("0"),
"Utilization" : 1400,
"ProducedCount" : 78
}

Result is required in aggregation framework. Please help.

This is what i tried,

db.collection.aggregate([
{ "$match" : { "$and" : [ { "MachineID" : { "$in" : [ 1001]}} , 
    { "Timestamp" : { "$gte" : ISODate("2017-04-18T01:30:00.000Z"), 
    "$lte" : ISODate("2017-04-19T01:30:00.000Z")}},]}
}, 
{
    "$addFields": {"lastUtilization": 0}
},
{
    "$addFields": {"lastProductsCount" : 0}
},
{
    "$group": {
        "_id": 
        {
             MachineID : '$MachineID',
            "RunStatus": "$RunStatus"
        },
        "Utilization" : 
        {
            "$sum" : 
            {
                "$cond": [
                        { "$ne": [ "$lastUtilization", 0 ] },
                         {"$subtract" : ["$Utilization", 
"$lastUtilization"]}, 0
                    ]
            }
        }, 
        "ProductsCount" : 
        {
            "$sum" : 
            {
                "$cond": [
                        { "$ne": [ "$lastProductsCount", 0 ] },
                         {"$subtract" : ["$ProductsCount", 
"$lastProductsCount"]}, 0
                    ]
            }
        }, 
        "lastProductsCount" : { "$avg" : "$ProductsCount"}, 
        "lastUtilization" : { "$avg" : "$Utilization"}
}
},
{
    "$project": 
    {
        "MachineID": "$_id.MachineID", 
        "RunStatus" : "$_id.RunStatus", 
        "Utilization" : "$Utilization", 
        "ProductsCount" : "$ProductsCount"
    }
},
]);

Answer 1

How's this? It doesn't compute the hour, but it does everything else.

[
    {
      $match: {
          $and: [
              {MachineID: {$in: [1001]}},
              {
                Timestamp: {
                    $gte: ISODate("2017-04-18T01:30:00.000Z"),
                    $lte: ISODate("2017-04-19T01:30:00.000Z")
                }
              }
          ]
      }
    },
    // Add all data to one array.
    {$group: {_id: "$MachineID", all: {$push: "$$ROOT"}}},
    // Create an array of (element, array index) pairs.
    {$addFields: {allWithIndex: {$zip: {inputs: ["$all", {$range: [0, {$size: "$all"}]}]}}}},
    // Create an array of {current: <element>, previous: <previous element>} pairs.
    {
      $project: {
          pairs: {
              $map: {
                  input: "$allWithIndex",
                  in : {
                      current: {$arrayElemAt: ["$$this", 0]},
                      prev: {
                          $arrayElemAt: [
                              "$all",
                              // Set prev == current for the first element.
                              {$max: [0, {$subtract: [{$arrayElemAt: ["$$this", 1]}, 1]}]}
                          ]
                      }
                  }
              }
          }
      }
    },
    // Compute the deltas.
    {$unwind: "$pairs"},
    {
      $group: {
          _id: {MachineID: "$_id", RunStatus: "$pairs.current.RunStatus"},
          ProductsCount:
              {$sum: {$subtract: ["$pairs.current.ProductsCount", "$pairs.prev.ProductsCount"]}},
          Utilization:
              {$sum: {$subtract: ["$pairs.current.Utilization", "$pairs.prev.Utilization"]}},
      }
    }
]