laydi/workflows/test_workflow.py

import gtk
from fluents import dataset, logger, plots, workflow
#import geneontology
#import gostat
from scipy import array, randn, log, ones
import cPickle
import networkx

class TestWorkflow (workflow.Workflow):

    name = 'Test Workflow'
    ident = 'test'
    description = 'This workflow currently serves as a general testing workflow.'

    def __init__(self):
        workflow.Workflow.__init__(self)

        load = workflow.Stage('load', 'Test Data')
        load.add_task(TestDataTask)
        self.add_stage(load)

        
class TestDataTask(workflow.Task):  
    name = "Test data"

    def __init__(self, input):
        workflow.Task.__init__(self, input)

    def run(self):
        logger.log('notice', 'Injecting foo test data')
        x = randn(500,15)
        X = dataset.Dataset(x)
        p = plots.ScatterPlot(X, X, 'rows', 'rows', '0_1', '0_2',name='scatter')
#        p2 = plots.ScatterMarkerPlot(X, X, 'rows', 'rows', '0_1', '0_2',name='marker')
        graph = networkx.XGraph()
        for x in 'ABCDEF':
            for y in 'ADE':
                graph.add_edge(x, y, 3)
        ds = dataset.GraphDataset(array(networkx.adj_matrix(graph)))
        ds_plot = plots.NetworkPlot(ds)

        cds = dataset.CategoryDataset(ones([3, 3]))
        ds_scatter = plots.ScatterMarkerPlot(ds, ds, 
                                             'rows_0', 'rows_0', 
                                             '0_1', '0_2')
        lp = plots.LineViewPlot(X,major_axis=0)
        vp = plots.VennPlot()
        self.datasets = [p]
        return [X, ds, p, ds_plot, ds_scatter, p2, cds, lp, vp]