status progress bar works on develop

rowers/views.py

@@ -14,6 +14,9 @@ from django.views.generic.base import TemplateView
 from django.db.models import Q
 from django import template
 from django.db import IntegrityError, transaction
+#from django.contrib.sessions.backends.db import SessionStore
+from importlib import import_module
+from django.contrib.sessions.models import Session
 from django.shortcuts import render
 from django.http import (
     HttpResponse, HttpResponseRedirect,
@@ -130,17 +133,71 @@ import django_rq
 queue = django_rq.get_queue('default')
 queuelow = django_rq.get_queue('low')
 queuehigh = django_rq.get_queue('low')
- 
+
+import redis
+import threading
 from redis import StrictRedis,Redis
 from rq.exceptions import NoSuchJobError
 from rq.registry import StartedJobRegistry
 from rq import Queue,cancel_job
 
+from django.core.cache import cache
+
+# Redis related
+session_engine = import_module(settings.SESSION_ENGINE)
+
+
+def getvalue(data):
+    perc = 0
+    total = 1
+    done = 0
+    id = 0
+    session_key = 'noot'
+    for i in data.iteritems():
+        if i[0] == 'total':
+            total = float(i[1])
+        if i[0] == 'done':
+            done = float(i[1])
+        if i[0] == 'id':
+            id = i[1]
+        if i[0] == 'session_key':
+            session_key = i[1]
+
+    return total,done,id,session_key
+
+class SessionTaskListener(threading.Thread):
+    def __init__(self, r, channels):
+        threading.Thread.__init__(self)
+        self.redis = r
+        self.pubsub = self.redis.pubsub()
+        self.pubsub.subscribe(channels)
+    
+    def work(self, item):
+
+        try:
+            data =  json.loads(item['data'])
+            total,done,id,session_key = getvalue(data)
+            perc = 100.*done/total
+            cache.set(id,perc)
+            
+        except TypeError:
+            pass
+            
+    def run(self):
+        for item in self.pubsub.listen():
+            if item['data'] == "KILL":
+                self.pubsub.unsubscribe()
+                print self, "unsubscribed and finished"
+                break
+            else:
+                self.work(item)
+
+
 queuefailed = Queue("failed",connection=Redis())
 redis_connection = StrictRedis()
 r = Redis()
-from .longtask import Listener
-client = Listener(r,['tasks'])
+
+client = SessionTaskListener(r,['tasks'])
 client.start()
 
 
@@ -187,9 +244,8 @@ def remove_asynctask(request,id):
         
     newtasks = []
     for task in oldtasks:
-        print task[0]
         if id not in task[0]:
-            newtasks += [(task[0],task[1])]
+            newtasks += [(task[0],task[1],task[2])]
 
     request.session['async_tasks'] = newtasks
             
@@ -223,8 +279,6 @@ def get_job_status(jobid):
     if settings.DEBUG:
         job = celery_result.AsyncResult(jobid)
         jobresult = job.result
-        channel = 'task:<'+job.id+'>:progress'
-        channel = 'noot'
         
         if 'fail' in job.status.lower():
             jobresult = '0'
@@ -271,6 +325,7 @@ def kill_async_job(request,id='aap'):
             pass
             
     remove_asynctask(request,id)
+    cache.delete(id)
     url = reverse(session_jobs_status)
 
     return HttpResponseRedirect(url)
@@ -278,11 +333,16 @@ def kill_async_job(request,id='aap'):
 @login_required()
 def test_job_view(request,aantal=100):
 
-    job = myqueue(queuehigh,long_test_task,int(aantal))
+    session_key = request.session._session_key
+
+    job = myqueue(queuehigh,long_test_task,int(aantal),
+                  session_key=session_key)
+
+
     try:
-        request.session['async_tasks'] += [(job.id,'long_test_task')]
+        request.session['async_tasks'] += [(job.id,'long_test_task',0)]
     except KeyError:
-        request.session['async_tasks'] = [(job.id,'long_test_task')]     
+        request.session['async_tasks'] = [(job.id,'long_test_task',0)]     
 
     url = reverse(session_jobs_status)
 
@@ -333,15 +393,32 @@ def get_stored_tasks_status(request):
     except KeyError:
         taskids = []
 
-    taskstatus = [{
-        'id':id,
-        'status':get_job_status(id)['status'],
-        'failed':get_job_status(id)['failed'],
-        'finished':get_job_status(id)['finished'],
-        'func_name':func_name,
-        'verbose': verbose_job_status[func_name]
-    } for id,func_name in taskids]
+    taskstatus = []
+    for id,func_name,session_progress in taskids:
+        progress = 0
+        cached_progress = cache.get(id)
+        finished = get_job_status(id)['finished']
+        if finished:
+            cache.set(id,100)
+            progress = 100
+        elif cached_progress>0:
+            progress = cached_progress
+        else:
+            progress = session_progress
 
+        this_task_status = {
+            'id':id,
+            'status':get_job_status(id)['status'],
+            'failed':get_job_status(id)['failed'],
+            'finished':get_job_status(id)['finished'],
+            'func_name':func_name,
+            'verbose': verbose_job_status[func_name],
+            'progress': progress,
+        }
+
+        taskstatus.append(this_task_status)
+
+    
     return taskstatus
 
 @login_required()
@@ -3268,9 +3345,9 @@ def otwrankings_view(request,theuser=0,
         )
         request.session['job_id'] = job.id
         try:
-            request.session['async_tasks'] += [(job.id,'updatecpwater')]
+            request.session['async_tasks'] += [(job.id,'updatecpwater',0)]
         except KeyError:
-            request.session['async_tasks'] = [(job.id,'updatecpwater')]        
+            request.session['async_tasks'] = [(job.id,'updatecpwater',0)]        
         messages.info(request,'New calculation queued. Refresh page or resubmit the date form to get the result')
     
     powerdf = pd.DataFrame({
@@ -3525,9 +3602,9 @@ def oterankings_view(request,theuser=0,
         )
         request.session['job_id'] = job.id
         try:
-            request.session['async_tasks'] += [(job.id,'updatecp')]
+            request.session['async_tasks'] += [(job.id,'updatecp',0)]
         except KeyError:
-            request.session['async_tasks'] = [(job.id,'updatecp')]
+            request.session['async_tasks'] = [(job.id,'updatecp',0)]
         messages.info(request,'New calculation queued.')
 
     powerdf = pd.DataFrame({
@@ -5659,9 +5736,9 @@ def workout_otwsetpower_view(request,id=0,message="",successmessage=""):
                           ratio=r.cpratio)
 
 	    try:
-                request.session['async_tasks'] += [(job.id,'otwsetpower')]
+                request.session['async_tasks'] += [(job.id,'otwsetpower',0)]
             except KeyError:
-                request.session['async_tasks'] = [(job.id,'otwsetpower')]
+                request.session['async_tasks'] = [(job.id,'otwsetpower',0)]
 
 	    successmessage = 'Your calculations have been submitted. You will receive an email when they are done. You can check the status of your calculations <a href="/rowers/jobs-status/">here</a>'
             messages.info(request,successmessage)
@@ -7437,9 +7514,9 @@ def workout_add_chart_view(request,id,plotnr=1):
             imagename=imagename
         )
         try:
-            request.session['async_tasks'] += [(jobid,'make_plot')]
+            request.session['async_tasks'] += [(jobid,'make_plot',0)]
         except KeyError:
-            request.session['async_tasks'] = [(jobid,'make_plot')]
+            request.session['async_tasks'] = [(jobid,'make_plot',0)]
 
     try:
         url = request.session['referer']