Endpoint and relation to track downsample datapoints.

Gemfile

@@ -28,6 +28,18 @@ gem 'bootsnap', '>= 1.4.4', require: false
 
 gem 'responders'
 
+# For scheduling cron tasks
+gem 'whenever', require: false
+
+# For rounding times
+gem 'rounding'
+
+# For serializing response
+gem 'jsonapi-serializer'
+
+# For paginating long series of data
+gem 'will_paginate', '~> 3.1.0'
+
 group :development, :test do
   # Call 'byebug' anywhere in the code to stop execution and get a debugger console
   gem 'byebug', platforms: %i[mri mingw x64_mingw]

Gemfile.lock

@@ -66,6 +66,7 @@ GEM
       msgpack (~> 1.0)
     builder (3.2.4)
     byebug (11.1.3)
+    chronic (0.10.2)
     concurrent-ruby (1.1.8)
     crass (1.0.6)
     diff-lcs (1.4.4)
@@ -83,6 +84,8 @@ GEM
     json_spec (1.1.5)
       multi_json (~> 1.0)
       rspec (>= 2.0, < 4.0)
+    jsonapi-serializer (2.2.0)
+      activesupport (>= 4.2)
     listen (3.5.1)
       rb-fsevent (~> 0.10, >= 0.10.3)
       rb-inotify (~> 0.9, >= 0.9.10)
@@ -151,6 +154,7 @@ GEM
       actionpack (>= 5.0)
       railties (>= 5.0)
     rexml (3.2.5)
+    rounding (1.0.1)
     rspec (3.10.0)
       rspec-core (~> 3.10.0)
       rspec-expectations (~> 3.10.0)
@@ -231,6 +235,9 @@ GEM
     websocket-driver (0.7.4)
       websocket-extensions (>= 0.1.0)
     websocket-extensions (0.1.5)
+    whenever (1.0.0)
+      chronic (>= 0.6.3)
+    will_paginate (3.1.8)
     zeitwerk (2.4.2)
 
 PLATFORMS
@@ -241,13 +248,15 @@ DEPENDENCIES
   byebug
   factory_bot_rails
   json_spec
+  jsonapi-serializer
   listen (~> 3.3)
   pg
   puma (~> 5.0)
   rack-mini-profiler (~> 2.0)
   rails (~> 6.1.3, >= 6.1.3.2)
   redis (~> 4.0)
   responders
+  rounding
   rspec-rails
   rubocop
   rubocop-performance
@@ -258,6 +267,8 @@ DEPENDENCIES
   tzinfo-data
   web-console (>= 4.1.0)
   webpacker (= 6.0.0.beta.7)
+  whenever
+  will_paginate (~> 3.1.0)
 
 RUBY VERSION
    ruby 3.0.2p107

app/controllers/api/data_points_controller.rb

@@ -7,5 +7,16 @@ def index
 
       respond_with data_points
     end
+
+    #
+    # down_sampled
+    #
+    def down_sampled
+      down_sampled_data = DownsampledDatapoint.paginate(page: params[:page],
+                                                        per_page: 20)
+      render json: DataPoints::DownSampledDataSerializer
+        .new(down_sampled_data)
+        .serializable_hash
+    end
   end
 end

app/models/channel.rb

@@ -3,4 +3,5 @@
 class Channel < ApplicationRecord
   belongs_to :device
   has_many :data_points, dependent: :destroy
+  has_many :downsampled_datapoints, dependent: :destroy
 end

app/models/data_point.rb

@@ -3,7 +3,9 @@
 class DataPoint < ApplicationRecord
   belongs_to :channel
 
-  after_commit :broadcast
+  attr_accessor :skip_broadcast
+
+  after_commit :broadcast, unless: proc { |data_point| data_point.skip_broadcast }
 
   private
 

app/models/downsampled_datapoint.rb

@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class DownsampledDatapoint < ApplicationRecord
+  belongs_to :channel
+
+  enum interval: { '5m' => 0, '10m' => 1, '15m' => 2 }
+end

app/serializers/data_points/down_sampled_data_serializer.rb

@@ -0,0 +1,6 @@
+# frozen_string_literal: true
+
+class DataPoints::DownSampledDataSerializer
+  include JSONAPI::Serializer
+  attributes :value, :time_interval, :interval, :channel_id
+end

app/services/data_points/rollup.rb

@@ -0,0 +1,65 @@
+# frozen_string_literal: true
+
+module DataPoints
+  class Rollup
+    #
+    # initialize
+    # @params {Integer} interval
+    # can be used to provide a different interval
+    #
+    def initialize(interval = nil)
+      @interval = interval || 5
+      @time_lower_bound = set_time_lower_bound
+    end
+
+    def call
+      downsampled_data = []
+      last_data_point_log_time = DataPoint.last.created_at
+      return if last_data_point_log_time < @time_lower_bound
+
+      while last_data_point_log_time > @time_lower_bound
+        channel_wise_average = DataPoint.where(created_at: @time_lower_bound..time_upper_bound)
+                                        .group(:channel_id)
+                                        .average(:value)
+        channel_wise_average.each do |channel_id, average|
+          downsampled_data << { channel_id: channel_id,
+                                value: average.to_f,
+                                time_interval: time_upper_bound,
+                                created_at: Time.now.utc,
+                                updated_at: Time.now.utc }
+        end
+
+        @time_lower_bound += interval_in_minutes
+      end
+      DownsampledDatapoint.insert_all!(downsampled_data)
+    end
+
+    private
+
+    #
+    # set_time_lower_bound
+    #
+    def set_time_lower_bound
+      if DownsampledDatapoint.any?
+        last_logged_time = DownsampledDatapoint.last.time_interval
+        last_logged_time + interval_in_minutes
+      else
+        DataPoint.first.created_at.floor_to(interval_in_minutes)
+      end
+    end
+
+    #
+    # time_upper_bound
+    #
+    def time_upper_bound
+      @time_lower_bound + interval_in_minutes
+    end
+
+    #
+    # interval_in_minutes
+    #
+    def interval_in_minutes
+      @interval.minutes
+    end
+  end
+end

config/routes.rb

@@ -2,7 +2,11 @@
   # For details on the DSL available within this file, see https://guides.rubyonrails.org/routing.html
   namespace :api do
     resources :channels, only: %i[index]
-    resources :data_points, only: %i[index]
+    resources :data_points, only: %i[index] do
+      collection do
+        get :down_sampled
+      end
+    end
     resources :devices, only: %i[index]
   end
 

config/schedule.rb

@@ -0,0 +1,12 @@
+# Use this file to easily define all of your cron jobs.
+#
+# It's helpful, but not entirely necessary to understand cron before proceeding.
+# http://en.wikipedia.org/wiki/Cron
+
+# Example:
+#
+set :output, "log/cron_log.log"
+
+every 2.minutes do
+  rake 'data_points:down_sample'
+end

db/migrate/20210823165746_create_rollups.rb

@@ -0,0 +1,12 @@
+class CreateRollups < ActiveRecord::Migration[6.1]
+  def change
+    create_table :rollups do |t|
+      t.string :name, null: false
+      t.string :interval, null: false
+      t.datetime :time, null: false
+      t.jsonb :dimensions, null: false, default: {}
+      t.float :value
+    end
+    add_index :rollups, [:name, :interval, :time, :dimensions], unique: true
+  end
+end

db/migrate/20210823184355_create_downsampled_datapoints.rb

@@ -0,0 +1,11 @@
+class CreateDownsampledDatapoints < ActiveRecord::Migration[6.1]
+  def change
+    create_table :downsampled_datapoints do |t|
+      t.references :channel
+      t.integer :interval
+      t.string :value
+
+      t.timestamps
+    end
+  end
+end

db/migrate/20210823195142_add_time_interval_to_downsapmpled_datapoints.rb

@@ -0,0 +1,5 @@
+class AddTimeIntervalToDownsapmpledDatapoints < ActiveRecord::Migration[6.1]
+  def change
+    add_column :downsampled_datapoints, :time_interval, :datetime, null: false, after: :value
+  end
+end

db/migrate/20210824175545_add_default_interval_value.rb

@@ -0,0 +1,5 @@
+class AddDefaultIntervalValue < ActiveRecord::Migration[6.1]
+  def change
+    change_column_default :downsampled_datapoints, :interval, 0
+  end
+end

db/migrate/20210824191349_drop_roll_ups_table.rb

@@ -0,0 +1,5 @@
+class DropRollUpsTable < ActiveRecord::Migration[6.1]
+  def change
+    drop_table :rollups
+  end
+end

db/seeds.rb

@@ -5,16 +5,21 @@
 #
 #   movies = Movie.create([{ name: 'Star Wars' }, { name: 'Lord of the Rings' }])
 #   Character.create(name: 'Luke', movie: movies.first)
+if ENV['VERSION'].present?
+  path = Rails.root.join('db', 'seeds', ENV['VERSION'])
+  raise 'Seed file not found by the provided identifier' unless File.exist?(path)
+  require path
+else
+  Device.create!(name: "ABJ-NM", category: 'balance', vendor: 'Kern').tap do |device|
+    device.channels.create! name: 'weight', unit: 'mg'
+  end
 
-Device.create!(name: "ABJ-NM", category: 'balance', vendor: 'Kern').tap do |device|
-  device.channels.create! name: 'weight', unit: 'mg'
-end
-
-Device.create!(name: "Entris Series", category: 'balance', vendor: 'Sartorius').tap do |device|
-  device.channels.create! name: 'weight', unit: 'mg'
-end
+  Device.create!(name: "Entris Series", category: 'balance', vendor: 'Sartorius').tap do |device|
+    device.channels.create! name: 'weight', unit: 'mg'
+  end
 
-Device.create!(name: "Hei-VAP", category: 'stirrer', vendor: 'Heidolph').tap do |device|
-  device.channels.create! name: 'rotation', unit: 'rpm'
-  device.channels.create! name: 'temperature', unit: 'C'
+  Device.create!(name: "Hei-VAP", category: 'stirrer', vendor: 'Heidolph').tap do |device|
+    device.channels.create! name: 'rotation', unit: 'rpm'
+    device.channels.create! name: 'temperature', unit: 'C'
+  end
 end

db/seeds/001_sample_data_points.rb

@@ -0,0 +1,17 @@
+# Can be executed using rails db:seed VERSION=001_sample_data_points.rb
+
+# Adding dummy data_points (10 per device via channel) for different devices
+# Each data_point for now is given a random float value
+
+Device.all.each do |device|
+  channel = device.channels.sample
+  minute_index = 0
+  10000.times do
+    minute_index += 1
+    DataPoint.create(channel: channel,
+                     value: rand(1.0..999.9),
+                     skip_broadcast: true,
+                     created_at: Time.now + minute_index.minutes)
+  end
+end
+

lib/tasks/data_points.rake

@@ -38,6 +38,10 @@ namespace :data_points do
     threads.map(&:join)
   end
 
+  task down_sample: :environment do
+    DataPoints::Rollup.new().call
+  end
+
   def wait_for_migration!
     loop do
       ActiveRecord::Migration.check_pending!

spec/controllers/api/data_points_controller_spec.rb

@@ -3,6 +3,14 @@
 require 'rails_helper'
 
 describe Api::DataPointsController, type: :controller do
+  let(:device) { create(:device) }
+  let(:channel) { create(:channel, device: device) }
+  let!(:down_sampled_data) do
+    create_list(:downsampled_datapoint, 100,
+                channel: channel,
+                value: rand(1.0..999.9))
+  end
+
   describe 'GET /api/data_points' do
     before do
       5.times do |index|
@@ -26,4 +34,19 @@
       expect(response.body).to have_json_size(3)
     end
   end
+
+  describe 'GET /api/down_sampled' do
+    it 'returns 20 down_sampled records per page' do
+      get :down_sampled, format: :json, params: { page: 1 }
+      parsed_response = JSON.parse(response.body)['data']
+      expect(parsed_response.size).to eq(20)
+    end
+
+    it 'returns appropriate attributes' do
+      get :down_sampled, format: :json, params: { page: 1 }
+      parsed_response = JSON.parse(response.body)['data']
+      expect(parsed_response.first['attributes'].keys)
+        .to eq(%w[value time_interval interval channel_id])
+    end
+  end
 end

spec/factories/downsampled_datapoints.rb

@@ -0,0 +1,8 @@
+# frozen_string_literal: true
+
+FactoryBot.define do
+  factory :downsampled_datapoint do
+    interval do '5m' end
+    time_interval { Time.current + 5.minutes }
+  end
+end

spec/models/downsampled_datapoint_spec.rb

@@ -0,0 +1,6 @@
+require 'rails_helper'
+# frozen_string_literal: true
+
+RSpec.describe DownsampledDatapoint, type: :model do
+  pending "add some examples to (or delete) #{__FILE__}"
+end