<commit_before>import xmlrpclib
import time
import solr

class LabcasClient(object):
    '''
    Python client used to interact with a remote Labcas back-end.
    Mostly for demo and example purposes...
    Available methods are defined in Java class org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManager.
    '''
    
    def __init__(self, 
                 workflowManagerUrl='http://localhost:9001/',
                 fileManagerUrl='http://localhost:9000/',
                 solrUrl='http://localhost:8983/solr/oodt-fm',
                 verbose=False):
        
        self.workflowManagerServerProxy = xmlrpclib.ServerProxy(workflowManagerUrl, verbose=verbose)
        self.fileManaferServerProxy = xmlrpclib.ServerProxy(fileManagerUrl, verbose=verbose)
        self.solrServerProxy = solr.SolrConnection(solrUrl)
        
    def getWorkflowsByEvent(self, eventName):
        '''Retrieve a specific workflow by the triggering event.'''
        
        workflows =  self.workflowManagerServerProxy.workflowmgr.getWorkflowsByEvent(eventName)
        for workflow in workflows:
            self.printWorkflow(workflow)
        
    def getWorkflowById(self, workflowId):
        '''Retrieve a specific workflow by its unique identifier.'''
        
        workflow =  self.workflowManagerServerProxy.workflowmgr.getWorkflowById(workflowId)
        self.printWorkflow(workflow)
        
    def executeWorkflow(self, tasks, metadata):
        '''Submits a dynamic workflow composed of the specified tasks, using the specified metadata.'''
        
        # FIXME: pass metadata through: s.encode('ascii',errors='ignore')
        return self.workflowManagerServerProxy.workflowmgr.executeDynamicWorkflow(tasks, metadata)
        
    def waitForCompletion(self, wInstId):
        ''' Monitors a workflow instance until it completes.'''
    
        # wait for the server to instantiate this workflow before querying it
        time.sleep(4) 
    
        # now use the workflow instance id to check for status, wait until completed
        running_status  = ['CREATED', 'QUEUED', 'STARTED', 'PAUSED']
        pge_task_status = ['STAGING INPUT', 'BUILDING CONFIG FILE', 'PGE EXEC', 'CRAWLING']
        finished_status = ['FINISHED', 'ERROR', 'METMISS']
        while (True):
            response = self.workflowManagerServerProxy.workflowmgr.getWorkflowInstanceById(wInstId)
            status = response['status']
            if status in running_status or status in pge_task_status:
                print 'Workflow instance=%s running with status=%s' % (wInstId, status)
                time.sleep(1)
            elif status in finished_status:
                print 'Workflow instance=%s ended with status=%s' % (wInstId, status)
                break
            else:
                print 'UNRECOGNIZED WORKFLOW STATUS: %s' % status
                break
        print response
        
    def uploadCollection(self, datasetId, metadata, newVersion=False, inPlace=False):
        
        # add 'DatasetId' key, value to other metadata
        metadata['DatasetId'] = datasetId
                
        # optionally request a new version
        if newVersion:
            metadata['NewVersion'] = 'true'
    
        # NOTE: currently, if you start a named workflow, the XMLRPC interface only returns True/False, not a workflow instance identifier...
        #tf = serverProxy.workflowmgr.handleEvent('labcas-upload', { 'DatasetId':'mydata' } )
    
        # ... consequently, you must submit an equivalent dynamic workflow, which does return the workflow instance id<commit_msg>Making labcas_client more tolerant for XMLRPC errors.<commit_after>import xmlrpclib
import time
import solr

class LabcasClient(object):
    '''
    Python client used to interact with a remote Labcas back-end.
    Mostly for demo and example purposes...
    Available methods are defined in Java class org.apache.oodt.cas.workflow.system.XmlRpcWorkflowManager.
    '''
    
    def __init__(self, 
                 workflowManagerUrl='http://localhost:9001/',
                 fileManagerUrl='http://localhost:9000/',
                 solrUrl='http://localhost:8983/solr/oodt-fm',
                 verbose=False):
        
        self.workflowManagerServerProxy = xmlrpclib.ServerProxy(workflowManagerUrl, verbose=verbose)
        self.fileManaferServerProxy = xmlrpclib.ServerProxy(fileManagerUrl, verbose=verbose)
        self.solrServerProxy = solr.SolrConnection(solrUrl)
        
    def getWorkflowsByEvent(self, eventName):
        '''Retrieve a specific workflow by the triggering event.'''
        
        workflows =  self.workflowManagerServerProxy.workflowmgr.getWorkflowsByEvent(eventName)
        for workflow in workflows:
            self.printWorkflow(workflow)
        
    def getWorkflowById(self, workflowId):
        '''Retrieve a specific workflow by its unique identifier.'''
        
        workflow =  self.workflowManagerServerProxy.workflowmgr.getWorkflowById(workflowId)
        self.printWorkflow(workflow)
        
    def executeWorkflow(self, tasks, metadata):
        '''Submits a dynamic workflow composed of the specified tasks, using the specified metadata.'''
        
        # FIXME: pass metadata through: s.encode('ascii',errors='ignore')
        return self.workflowManagerServerProxy.workflowmgr.executeDynamicWorkflow(tasks, metadata)
        
    def waitForCompletion(self, wInstId):
        ''' Monitors a workflow instance until it completes.'''
    
        # wait for the server to instantiate this workflow before querying it
        time.sleep(4) 
    
        # now use the workflow instance id to check for status, wait until completed
        running_status  = ['CREATED', 'QUEUED', 'STARTED', 'PAUSED']
        pge_task_status = ['STAGING INPUT', 'BUILDING CONFIG FILE', 'PGE EXEC', 'CRAWLING']
        finished_status = ['FINISHED', 'ERROR', 'METMISS']
        while (True):
            try:
                response = self.workflowManagerServerProxy.workflowmgr.getWorkflowInstanceById(wInstId)
                status = response['status']
                if status in running_status or status in pge_task_status:
                    print 'Workflow instance=%s running with status=%s' % (wInstId, status)
                    time.sleep(1)
                elif status in finished_status:
                    print 'Workflow instance=%s ended with status=%s' % (wInstId, status)
                    break
                else:
                    print 'UNRECOGNIZED WORKFLOW STATUS: %s' % status
                    break
            except xmlrpclib.Fault as e:
                # must ignore XML-RPC exeptions that often arise when querying OODT for a specific workflow
                # just try again with the same workflow identifier
                print e
        print response
        
    def uploadCollection(self, datasetId, metadata, newVersion=False, inPlace=False):
        
        # add 'DatasetId' key, value to other metadata
        metadata['DatasetId'] = datasetId
                
        # optionally request a new version
        if newVersion:
            metadata['NewVersion'] = 'true'
    
        # NOTE: currently, if you start a named workflow, the XMLRPC interface only returns True/False, not a workflow instance identifier...
        #tf = serverProxy.workflowmgr.handleEvent('labcas-upload', { 'DatasetId':'mydata' } )
    
        # ... consequently, you must submit an equivalent dynamic workflow, which does return the workflow instance id
{'fraction_non_alphanumeric': 0.08292424041395781, 'fraction_numerical': 0.00371500597054531, 'mean_word_length': 2.475334255417243, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 7}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1462617', 'n_tokens_mistral': 2025, 'n_tokens_neox': 1944, 'n_words': 560}