<commit_before>#!/usr/bin/env python

'''
@author: <NAME> <<EMAIL>>
@file: pyon/ion/transform.py
@description: Implementation for TransformBase class
'''

from pyon.ion.streamproc import StreamProcess

class TransformBase(StreamProcess):
    """

    """
    def on_start(self):
        super(TransformBase,self).on_start()
        # Assign a name based on CFG, required for messaging
        self.name = self.CFG.get('process',{}).get('name',None)

        # Assign a list of streams available
        self.streams = self.CFG.get('process',{}).get('publish_streams',[])

    def callback(self):
        pass

    def process(self, packet):
        pass

class TransformProcessAdaptor(TransformBase):
    """
    Models a transform adaptor to an external transform process
    """
    def __init__(self, *args, **kwargs):
        super(TransformProcessAdaptor, self).__init__()
        #@todo: Initialize IO, pipes or some variant of IPC

    def on_start(self):
        pass

    def callback(self):
        pass

    def process(self, packet):
        pass



class TransformDataProcess(TransformBase):
    """Model for a TransformDataProcess

    """
    def __init__(self):
        super(TransformDataProcess,self).__init__()
        self._pub_init = False

    def on_start(self):
        super(TransformDataProcess,self).on_start()

    def process(self, packet):
        pass

    def callback(self):
        pass

    def publish(self,msg):
        self._publish_all(msg)


    def _publish_all(self, msg):
        '''Publishes a message on all output streams (publishers)
        '''
        # Ensure the publisher list is only initialized once
        if not self._pub_init:
            stream_names = list(k for k,v in self.streams.iteritems())
            self.publishers = []
            for stream in stream_names:
                self.publishers.append(getattr(self,stream))


        for publisher in self.publishers:
            publisher.publish(msg)



class TransformFunction(TransformDataProcess):
    """ Represents a transform function
    Input is given to the transform, it runs until the transform is complete
    and the output is returned.

    Input can be obtained from a stream, and output can be sent to a stream

    A TransformFunction is interchangeable with a TransformProcess but it is also able to be called explicitly to run

    """
    def execute(self, input):
        pass

    def process(self, packet):
        ret = self.execute(packet)
        if len(self.streams)>0:
            self.publish(ret)


<commit_msg>Fixed bug where streams aren't getting initiated.<commit_after>#!/usr/bin/env python

'''
@author: <NAME> <<EMAIL>>
@file: pyon/ion/transform.py
@description: Implementation for TransformBase class
'''

from pyon.ion.streamproc import StreamProcess

class TransformBase(StreamProcess):
    """

    """
    def on_start(self):
        super(TransformBase,self).on_start()
        # Assign a name based on CFG, required for messaging
        self.name = self.CFG.get('process',{}).get('name',None)

        # Assign a list of streams available
        self.streams = self.CFG.get('process',{}).get('publish_streams',[])

    def callback(self):
        pass

    def process(self, packet):
        pass

class TransformProcessAdaptor(TransformBase):
    """
    Models a transform adaptor to an external transform process
    """
    def __init__(self, *args, **kwargs):
        super(TransformProcessAdaptor, self).__init__()
        #@todo: Initialize IO, pipes or some variant of IPC

    def callback(self):
        pass

    def process(self, packet):
        pass



class TransformDataProcess(TransformBase):
    """Model for a TransformDataProcess

    """
    def __init__(self):
        super(TransformDataProcess,self).__init__()
        self._pub_init = False

    def on_start(self):
        super(TransformDataProcess,self).on_start()

    def process(self, packet):
        pass

    def callback(self):
        pass

    def publish(self,msg):
        self._publish_all(msg)


    def _publish_all(self, msg):
        '''Publishes a message on all output streams (publishers)
        '''
        # Ensure the publisher list is only initialized once
        if not self._pub_init:
            stream_names = list(k for k,v in self.streams.iteritems())
            self.publishers = []
            for stream in stream_names:
                self.publishers.append(getattr(self,stream))


        for publisher in self.publishers:
            publisher.publish(msg)



class TransformFunction(TransformDataProcess):
    """ Represents a transform function
    Input is given to the transform, it runs until the transform is complete
    and the output is returned.

    Input can be obtained from a stream, and output can be sent to a stream

    A TransformFunction is interchangeable with a TransformProcess but it is also able to be called explicitly to run

    """
    def execute(self, input):
        pass

    def process(self, packet):
        ret = self.execute(packet)
        if len(self.streams)>0:
            self.publish(ret)



{'fraction_non_alphanumeric': 0.09636434714620798, 'fraction_numerical': 0.00039093041438623924, 'mean_word_length': 2.691919191919192, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, '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': '7518924', 'n_tokens_mistral': 1454, 'n_tokens_neox': 1374, 'n_words': 433}