Scheduling of control loops, chart improvements

Gemfile

@@ -33,8 +33,10 @@ gem 'jbuilder', '~> 2.5'
 # Use Capistrano for deployment
 # gem 'capistrano-rails', group: :development
 
+gem 'active_scheduler'
 gem 'devise'
 gem "haml-rails", "~> 1.0"
+gem 'kaminari'
 gem 'mqtt'
 gem 'paho-mqtt'
 gem 'redis'

Gemfile.lock

@@ -24,6 +24,9 @@ GEM
       erubi (~> 1.4)
       rails-dom-testing (~> 2.0)
       rails-html-sanitizer (~> 1.0, >= 1.0.3)
+    active_scheduler (0.5.0)
+      activejob (>= 4.2.0)
+      rake
     activejob (5.1.5)
       activesupport (= 5.1.5)
       globalid (>= 0.3.6)
@@ -102,7 +105,19 @@ GEM
     jbuilder (2.7.0)
       activesupport (>= 4.2.0)
       multi_json (>= 1.2)
-    libv8 (3.16.14.19-x86_64-linux)
+    kaminari (1.1.1)
+      activesupport (>= 4.1.0)
+      kaminari-actionview (= 1.1.1)
+      kaminari-activerecord (= 1.1.1)
+      kaminari-core (= 1.1.1)
+    kaminari-actionview (1.1.1)
+      actionview
+      kaminari-core (= 1.1.1)
+    kaminari-activerecord (1.1.1)
+      activerecord
+      kaminari-core (= 1.1.1)
+    kaminari-core (1.1.1)
+    libv8 (3.16.14.19)
     listen (3.1.5)
       rb-fsevent (~> 0.9, >= 0.9.4)
       rb-inotify (~> 0.9, >= 0.9.7)
@@ -256,6 +271,7 @@ PLATFORMS
   ruby
 
 DEPENDENCIES
+  active_scheduler
   byebug
   capybara (~> 2.13)
   coffee-rails (~> 4.2)
@@ -264,6 +280,7 @@ DEPENDENCIES
   factory_bot_rails
   haml-rails (~> 1.0)
   jbuilder (~> 2.5)
+  kaminari
   listen (>= 3.0.5, < 3.2)
   minitest-hooks
   mqtt

Rakefile

@@ -3,7 +3,24 @@
 
 require_relative 'config/application'
 require 'resque/tasks'
+require 'resque/scheduler/tasks'
+require 'active_scheduler'
 
-task 'resque:setup' => :environment
+task 'resque:setup' => :environment do
+  Resque.before_fork = Proc.new { 
+    ActiveRecord::Base.establish_connection
+
+    # Open the new separate log file
+    # logfile = File.open(File.join(Rails.root, 'log', 'resque.log'), 'a')
+
+    # Activate file synchronization
+    # logfile.sync = true
+
+    # Create a new buffered logger
+    # Resque.logger = ActiveSupport::BufferedLogger.new(logfile)
+    # Resque.logger.level = Logger::DEBUG
+    # Resque.logger.debug "Resque Logger Initialized!"
+  }
+end
 
 Rails.application.load_tasks

app/assets/javascripts/application.js

@@ -18,4 +18,5 @@
 //= require amcharts/dist/amcharts/amcharts.js
 //= require amcharts/dist/amcharts/serial.js
 //= require amcharts/dist/amcharts/themes/light.js
+// require amcharts/dist/amcharts/plugins/export/export.js
 //= require_tree .

app/assets/javascripts/channels/actuator.coffee

@@ -23,5 +23,5 @@ $(document).on 'turbolinks:load', ->
 
       createStatusElement: (data) ->
         """
-        <span>#{data['value']}</span>
+        <span>#{data}</span>
         """

app/assets/javascripts/channels/sensors.coffee

@@ -7,46 +7,87 @@ $(document).on 'turbolinks:load', ->
     console.log 'Streaming data for sensor #' + sensor_id
     chartData = []
     chart = AmCharts.makeChart('chart',
+    'pathToImages': '/assets/'
     'type': 'serial'
     'theme': 'light'
-    'dataDateFormat': 'YYYY-MM-DD'
+    'dataDateFormat': 'YYYY-MM-DD JJ:NN:SS'
     'valueAxes': [ {
       'id': 'v1'
       'position': 'left'
     } ]
     'graphs': [ {
       'id': 'g1'
+      #'type': 'smoothedline'
       'bullet': 'round'
       'valueField': 'value'
-      'balloonText': '[[category]]: [[value]]'
+      'balloonText': '[[value]]'
     } ]
-    'categoryField': 'created_at'
+    'chartScrollbar': {
+      'graph': 'g1'
+      'oppositeAxis': false
+      'offset': 30
+      'scrollbarHeight': 80
+      'backgroundAlpha': 0
+      'selectedBackgroundAlpha': 0.2
+      'selectedBackgroundColor': '#888888'
+      'graphFillAlpha': 0
+      'graphLineAlpha': 0.5
+      'selectedGraphFillAlpha': 0
+      'selectedGraphLineAlpha': 1
+      'autoGridCount': true
+      'color': '#000000'
+      'graphLineColor': '#FF0000'
+    }
+    'chartCursor': {
+      'pan': true
+      'valueLineEnabled': true
+      'valueLineBalloonEnabled': true
+      'cursorAlpha':1
+      'cursorColor':'#258cbb'
+      'limitToGraph':'g1'
+      'valueLineAlpha':0.2
+      'valueZoomable':true
+    }
+    'categoryField': 'measured_at_formatted'
     'categoryAxis':
       'parseDates': true
-      'equalSpacing': true
+      'minPeriod': 'ss',
       'dashLength': 1
       'minorGridEnabled': true
     'dataProvider': chartData)
+    chart.ignoreZoomed = false
+    chart.addListener 'zoomed', (event) ->
+      if chart.ignoreZoomed
+        chart.ignoreZoomed = false
+      else
+        chart.zoomStartDate = event.startDate
+        chart.zoomEndDate = event.endDate
+    chart.addListener 'dataUpdated', (event) ->
+      if chart.zoomStartDate && chart.zoomEndDate
+        the_end_date = new Date()
+        chart.zoomEndDate = the_end_date
+        chart.zoomToDates chart.zoomStartDate, the_end_date
+    $.getJSON "/sensors/#{sensor_id}/data"
+    .then (response) ->
+      chartData.push.apply(chartData, response.reverse())
+      chartData.sort (a, b) ->
+        new Date(a.measured_at_formatted).getTime() - new Date(b.measured_at_formatted).getTime()
+      chart.validateData()
+      end = new Date()
+      start = new Date()
+      start.setDate(end.getDate()-1)
+      chart.zoomToDates(start, end)
+    .catch (error) ->
+      console.log "Error getting sensor data #{sensor_id}"
+      console.log error
     App.cable.subscriptions.create { channel: "SensorChannel", id: sensor_id },
       received: (data) ->
         @addDataToChart data
-        #@appendLine data
 
       addDataToChart: (data) ->
-        console.log(data)
-        if chartData.length > 200
-          chartData.splice 0, chartData.length - 200
         chartData.push data
+        chartData.sort (a, b) ->
+          new Date(a.measured_at_formatted).getTime() - new Date(b.measured_at_formatted).getTime()
+        chart.ignoreZoomed = true
         chart.validateData()
-
-      appendLine: (data) ->
-        html = @createLine(data)
-        $('#sensor-readings').append(html)
-
-      createLine: (data) ->
-        """
-        <div class="sensor-reading">
-          <span class="timestamp">#{data["created_at"]}</span>
-          <span class="value">#{data["value"]}</span>
-        </div>
-        """
+        return

app/assets/javascripts/control_loops.coffee

@@ -27,7 +27,7 @@ $(document).on 'turbolinks:load', ->
       'balloonText': '[[category]]: [[value]]'
       }
     ]
-    'categoryField': 'created_at'
+    'categoryField': 'measured_at'
     'categoryAxis':
       'parseDates': true
       'equalSpacing': true

app/assets/stylesheets/application.css

@@ -12,4 +12,5 @@
  *
  *= require_tree .
  *= require_self
+ *= require amcharts/dist/amcharts/plugins/export/export.css
  */

app/channels/sensor_channel.rb

@@ -1,10 +1,7 @@
 class SensorChannel < ApplicationCable::Channel
   def subscribed
-    Rails.logger.debug "DEBUG: someone subscribed"
     sensor = Sensor.find(params[:id])
     stream_for sensor, coder: ActiveSupport::JSON do |message|
-      Rails.logger.debug "Received message #{message}"
-
       transmit message
     end
   end

app/controllers/control_loops_controller.rb

@@ -69,6 +69,9 @@ def set_control_loop
 
     # Never trust parameters from the scary internet, only allow the white list through.
     def control_loop_params
-      params.require(:control_loop).permit(:name, :mode, :reference, :parameters, :sensor_id, :actuator_id)
+      params.require(:control_loop).permit(:name, :mode, :reference,
+                                           :parameters, :next_run,
+                                           :sampling_rate, :sensor_id,
+                                           :actuator_id)
     end
 end

app/controllers/sensors/data_controller.rb

@@ -6,7 +6,7 @@ def index
 
   private
     def fetch_data
-      @data = Sensor.find(params[:sensor_id]).data.limit(20)
+      @data = Sensor.find(params[:sensor_id]).data.order(measured_at: :desc) #.page(params[:page]).per(300)
     end
 
     def sensor_params

app/jobs/application_job.rb

@@ -1,2 +1,5 @@
 class ApplicationJob < ActiveJob::Base
+  #before_perform do |job|
+  #  ActiveRecord::Base.clear_active_connections!
+  #end
 end

app/jobs/control_loop_job.rb

@@ -1,26 +1,26 @@
 class ControlLoopJob < ApplicationJob
-  queue_as :control_loops
+  queue_as :run_control_loops
 
-  def perform
-    puts "TESTE"
-    Rails.logger.debug "*********** DEBUG: I'm here"
-  end
+  def perform(control_loop_id)
+    require Rails.root.join('lib', 'cervejator_mqtt').to_s
 
-  def do_perform(control_loop_id)
-    require 'mqtt'
     controller = ControlLoop.find(control_loop_id)
     action = controller.run
 
     return if action == :inactive
 
-    MQTT::Client.connect(host: ENV['MQTT_BROKER_HOST'], 
-                         username: 'mosquitto', 
-                         port: ENV['MQTT_BROKER_PORT']) do |client|
-      # parameters are: topic, payload, retain flag
-      # the retain flag means that the broker will store the last message for
-      # clients that subscribe later. Those will receive immediatelly the last
-      # retained message.
-      client.publish("actuators/#{controller.actuator.write_key}", action, true, 1)
-    end
+    cli = Cervejator::MQTT.instance
+
+    # parameters are: topic, payload, retain flag
+    # the retain flag means that the broker will store the last message for
+    # clients that subscribe later. Those will receive immediatelly the last
+    # retained message.
+    # IDEALLY WE WOULD SEND RETAINED MESSAGES, BUT UNTIL I FIND A WAY FOR THE
+    # MOSQUITTO BROKER TO ONLY RETAIN THE LAST MESSAGE PER TOPIC, I WILL REFRAIN
+    # FROM USING THIS, BECAUSE IT GENERATES A HUGE BACKLOG OF MESSAGES WHEN THE
+    # SUBSCRIBER IS OFFLINE
+    cli.publish("actuator/#{controller.actuator.write_key}", action, false, 1)
+
+    controller.update_attribute(:next_run, Time.now + controller.sampling_rate.seconds)
   end
 end

app/jobs/schedule_control_loops_job.rb

@@ -0,0 +1,21 @@
+class ScheduleControlLoopsJob < ApplicationJob
+  queue_as :control_loops
+
+  def perform
+    ControlLoop.all.each do |cl|
+      ActiveJob::Base.logger.debug "Checking #{cl.name} for run"
+      if Time.now >= cl.next_run
+        # enqueue assynchronous action, we don't want to hang around
+        # here for too long
+        ControlLoopJob.perform_later(cl.id)
+        ActiveJob::Base.logger.info "Controle Loop ##{cl.id}:#{cl.name} enqueued to run"
+      end
+
+      # Let's test for missed deadlines
+      if Time.now >= cl.next_run + cl.sampling_rate.seconds
+        # this means the ControlLoopJob is not keeping up
+        ActiveJob::Base.logger.error "Controle Loop ##{cl.id}:#{cl.name} is missing deadlines!"
+      end
+    end
+  end
+end

app/jobs/update_actuator_job.rb

@@ -0,0 +1,11 @@
+class UpdateActuatorJob < ApplicationJob
+  queue_as :default
+
+  def perform(actuator_id, output)
+    if actuator = Actuator.find(actuator_id)
+      actuator.update_attribute(:output, output) if Actuator.outputs.keys.include?(output)
+    else
+      Rails.logger.error "UpdateActuatorJob#perform could not find Actuator with id #{actuator_id}"
+    end
+  end
+end

app/models/datum.rb

@@ -1,3 +1,9 @@
 class Datum < ApplicationRecord
   belongs_to :sensor
+
+  #serialize :measured_at, FormatedDateTime
+
+  def measured_at_formatted
+    self[:measured_at].strftime('%Y-%m-%d %H:%M:%S')
+  end
 end

app/views/control_loops/_form.html.haml

@@ -18,6 +18,12 @@
   .field
     = f.label :parameters
     = f.text_field :parameters
+  .field
+    = f.label :next_run
+    = f.time_select :next_run
+  .field
+    = f.label :sampling_rate
+    = f.number_field :sampling_rate
   .field
     = f.label :sensor
     = f.collection_select :sensor_id, Sensor.all, :id, :name, prompt: true

app/views/control_loops/show.html.haml

@@ -12,6 +12,12 @@
 %p
   %b Parameters:
   = @control_loop.parameters
+%p
+  %b Next run:
+  = @control_loop.next_run
+%p
+  %b Sampling Rate:
+  = @control_loop.sampling_rate
 %p
   %b Sensor:
   = link_to @control_loop.sensor.name, @control_loop.sensor

app/views/sensors/_sensor.json.jbuilder

@@ -1,2 +1,2 @@
-json.extract! sensor, :id, :name, :created_at, :updated_at
+json.extract! sensor, :id, :name, :measured_at, :created_at, :updated_at
 json.url sensor_url(sensor, format: :json)

app/views/sensors/data/_datum.json.jbuilder

@@ -1,2 +1,2 @@
-json.extract! datum, :created_at, :value
+json.extract! datum, :measured_at_formatted, :value
 #json.url sensor_data_url(datum, format: :json)

app/views/sensors/data/index.json.jbuilder

@@ -1 +1,5 @@
-json.array! @data, partial: 'sensors/data/datum', as: :datum
+#json.array! @data, partial: 'sensors/data/datum', as: :datum
+json.array! @data do |datum|
+  json.measured_at_formatted datum.measured_at_formatted
+  json.value datum.value
+end

app/views/sensors/show.html.haml

@@ -4,6 +4,7 @@
   = @sensor.name
   = @sensor.write_key
 .chart#chart
+= link_to 'load more data', sensor_data_path(@sensor, page: 2), remote: true
 .sensor-readings#sensor-readings{ "data-sensor-id": @sensor.id }
 = link_to 'Edit', edit_sensor_path(@sensor)
 |