rabbitmq - pyCommon

frameCli.py

import pika
import sys
import pyCommon.data.serialize as serialize
import pyCommon.net.command    as cmd
import pyCommon.net.event      as evt


queueName = sys.argv[1]


def loadCsv(channel):
    path = raw_input('path: ')
    loadCmd = cmd.LOAD_DATASOURCE_REQUESTED({
        'type': 'CSV',
        'path': path
    })
    channel.basic_publish(exchange='',
                          routing_key=queueName,
                          body=serialize.send_zipped_pickle(loadCmd),
                          properties=pika.BasicProperties(
                              delivery_mode = 2, # make message persistent
                          ))


def getCsvInfo(channel):
    getCmd = cmd.GET_DATASOURCE_INFO_REQUESTED({})
    channel.basic_publish(exchange='',
                          routing_key=queueName,
                          body=serialize.send_zipped_pickle(getCmd),
                          properties=pika.BasicProperties(
                              delivery_mode = 2, # make message persistent
                          ))


connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='localhost')
    )
channel = connection.channel()
channel.queue_declare(queue=queueName, durable=True)

# /Users/rogovski/CodeRepos/dataset/elec-usage-ngma/elec-usage-ngma.csv

def cliLoop():
    while True:
      inp = raw_input('> ')
      if inp == 'exit':
          connection.close()
          break;

      if inp == 'loadcsv':
          loadCsv(channel)
      elif inp == 'info':
          getCsvInfo(channel)
      else:
          print 'unrecognized command'

cliLoop()

frameProc.py

import sys
import pika
import pyCommon.either.core    as either
import pyCommon.either.util    as eutil
import pyCommon.data.frame     as frame
import pyCommon.data.query     as query
import pyCommon.data.serialize as serialize
import pyCommon.net.command    as cmd
import pyCommon.net.event      as evt


#############################################################################
## data frame for process
df = frame.Frame();


#############################################################################
## command handlers for process
commandHandler = {}


#############################################################################
## load a datasource of a particular type into this processes frame
def loadDataSource(cmd):
    def loadDs(dsType):
        if dsType == 'CSV':
            epath = cmd.get('path')

            if epath.name == 'Left':
                return either.Left('csv path not provided')
            else:
                epath = epath.val

            return df.loadCsv(epath)

        else:
            return either.Left('invalid datasource type')

    result = either.pipe(
        lambda _: cmd.get('type'),
        lambda dsType: loadDs(dsType)
    )(None)

    if result.name == 'Left':
        return evt.LOAD_DATASOURCE_FAILED({
            'msg': result.val
        });
    else:
        return evt.LOAD_DATASOURCE_SUCCEEDED({
            'msg': result.val
        });


commandHandler['LOAD_DATASOURCE_REQUESTED'] = loadDataSource


#############################################################################
## get info about loaded datasource
def getDataSourceInfo(cmd):
    efrm = df.getFrame()

    if efrm.name == 'Left':
        return evt.GET_DATASOURCE_INFO_FAILED({
            'msg': efrm.val
        })
    else:
        return evt.GET_DATASOURCE_INFO_SUCCEEDED({
            'count': efrm.val.count(),
            'dimensions': efrm.val.columns.values
        })

commandHandler['GET_DATASOURCE_INFO_REQUESTED'] = getDataSourceInfo


#############################################################################
## queue name for worker to attach to


queueName = sys.argv[1]


connection = pika.BlockingConnection(
        pika.ConnectionParameters(host='localhost')
    )
channel = connection.channel()
channel.queue_declare(queue=queueName, durable=True)


def callback(ch, method, properties, body):
    try:
        cmd = serialize.recv_zipped_pickle(body)
        handler = eutil.dictGet(cmd.commandType, commandHandler)

        if handler.name == 'Left':
            print 'handler not found'
        else:
            print handler.val(cmd)

    except:
        print 'failed to depickle'

    ch.basic_ack(delivery_tag = method.delivery_tag)


print(' [*] Waiting for messages. To exit press CTRL+C')
channel.basic_qos(prefetch_count=1)
channel.basic_consume(callback, queue=queueName)
channel.start_consuming()