messages from tasks

6 files changed, 88 insertions, 6 deletions

diff --git a/app/client/modules/samplernn/samplernn.tasks.js b/app/client/modules/samplernn/samplernn.tasks.js
new file mode 100644
index 0000000..fa41fe8
--- /dev/null
+++ b/app/client/modules/samplernn/samplernn.tasks.js
@@ -0,0 +1,37 @@
+import uuidv1 from 'uuid/v1'
+
+import socket from '../../socket'
+import types from '../../types'
+
+import actions from '../../actions'
+
+export const train_task = (dataset, epochs=1) => dispatch => {
+  const task = {
+    module: 'samplernn',
+    activity: 'train',
+    dataset: dataset,
+    epochs: epochs,
+    opt: {
+      sample_length: 44100 * 5,
+      n_samples: 6,
+      keep_old_checkpoints: false,
+    }
+  }
+  return actions.queue.add_task(task)
+}
+export const fetch_task = (dataset) => dispatch => {
+  const task = {
+    module: 'samplernn',
+    activity: 'fetch',
+    dataset: dataset,
+  }
+  return actions.queue.add_task(task)
+}
+export const clear_cache_task = (dataset) => dispatch => {
+  const task = {
+    module: 'samplernn',
+    activity: 'clear_cache',
+    dataset: dataset,
+  }
+  return actions.queue.add_task(task)
+}
\ No newline at end of file
diff --git a/app/client/socket/socket.task.js b/app/client/socket/socket.task.js
index 2f3c65a..3a78c6c 100644
--- a/app/client/socket/socket.task.js
+++ b/app/client/socket/socket.task.js
@@ -40,6 +40,12 @@ socket.on('task_res', (data) => {
       break
     case 'set_priority':
       break
+    case 'progress':
+      dispatch({ type: types.task.progress, task: data.task })
+      break
+    case 'epoch':
+      dispatch({ type: types.task.progress, task: data.task })
+      break
     case 'error':
       return console.log('task error', data)
     default:
diff --git a/app/relay/modules/samplernn.js b/app/relay/modules/samplernn.js
index d3b5012..64a6b88 100644
--- a/app/relay/modules/samplernn.js
+++ b/app/relay/modules/samplernn.js
@@ -6,11 +6,14 @@ const cwd = process.env.SAMPLERNN_CWD || path.join(process.env.HOME, 'code/' + n
 const fetch = {
   type: 'perl',
   script: 'get.pl',
-  // here i need to bridge again... get the filename that comes back from youtube-dl
-  // and tell the cortex that URL -> fn and add the filename!s
   params: (task) => {
     console.log(task)
     return [ task.opt.url ]
+  },
+  listen: (task, line, i) => {
+    // here i need to bridge again... get the filename that comes back from youtube-dl
+    // and tell the cortex that URL -> fn and add the filename!s
+    return null
   }
 }
 const train = {
diff --git a/app/relay/modules/test.js b/app/relay/modules/test.js
index b0753f8..57a324a 100644
--- a/app/relay/modules/test.js
+++ b/app/relay/modules/test.js
@@ -17,6 +17,15 @@ const live = {
 	type: 'python',
   script: 'test.py',
   live: true,
+  listen: (task, line, i) => {
+    if (line.match('orange')) {
+      return { type: 'progress', task, mode: 'orange', progress: i }
+    }
+    if ( (i % 10) === 9) {
+      return { type: 'epoch', task, epoch: (i/10)|0 }
+    }
+    return null
+  }
 }
 
 export default {
diff --git a/app/relay/queue.js b/app/relay/queue.js
new file mode 100644
index 0000000..6e67b80
--- /dev/null
+++ b/app/relay/queue.js
@@ -0,0 +1,17 @@
+let queue = []
+let active = true
+let status = 'waiting'
+
+const is_active = () => active
+const get_status = () => status
+const activate = () => {
+  active = true
+  status = 'active'
+}
+const deactivate = reason => {
+  active = false
+  status = reason
+}
+const add_task = task => queue.push(task)
+const remove_task = task => queue = queue.filter(t => t.id !== task.id)
+const get_next_task = () => queue.shift()
diff --git a/app/relay/runner.js b/app/relay/runner.js
index 94b7779..c04419d 100644
--- a/app/relay/runner.js
+++ b/app/relay/runner.js
@@ -240,11 +240,18 @@ export function run_task_with_activity(task, module, activity, preempt=false, wa
   
   watch && console.log("watching stdout..")
 
-  subprocess.stdout.on('data', data => {
-    watch && remote.emit('task_res', { type: 'stdout', processor, data: data.toString('utf8') })
+  let count = 0
+  subprocess.stdout.on('data', buf => {
+    const data = buf.toString('utf8')
+    watch && remote.emit('task_res', { type: 'stdout', processor, data })
+    if (activity.listen) {
+      const res = activity.listen(task, data, count++)
+      if (res) remote.emit('task_res', res)
+    }
   })
-  subprocess.stderr.on('data', data => {
-    watch && remote.emit('task_res', { type: 'stderr', processor, data: data.toString('utf8') })
+  subprocess.stderr.on('data', buf => {
+    const data = buf.toString('utf8')
+    watch && remote.emit('task_res', { type: 'stderr', processor, data })
   })
 
   let finished = false
@@ -253,6 +260,8 @@ export function run_task_with_activity(task, module, activity, preempt=false, wa
     if (finished) return
     finished = true
     task.processing = false
+    task.completed = true
+    task.success = false
 		console.log('task error', subprocess.exitCode, err)
     finish({
       type: 'task_error',
@@ -267,6 +276,7 @@ export function run_task_with_activity(task, module, activity, preempt=false, wa
     set_connected(false)
     task.processing = false
     task.completed = true
+    task.success = true
     finish({
       type: 'task_finish',
       task,