<commit_before>
"""
@author <NAME>
@file ion/processes/data/ingestion/ingestion_launcher.py
@description Ingestion Launcher
"""
from interface.objects import ProcessDefinition, ExchangeQuery
from interface.services.cei.iprocess_dispatcher_service import ProcessDispatcherServiceClient
from interface.services.dm.ipubsub_management_service import PubsubManagementServiceClient
from pyon.core import bootstrap
from pyon.service.service import BaseService
from interface.services.dm.itransform_management_service import TransformManagementServiceClient
from pyon.util.config import CFG

class CacheLauncher(BaseService):
    def on_start(self):
        super(CacheLauncher,self).on_start()
        tms_cli = TransformManagementServiceClient()
        pubsub_cli = PubsubManagementServiceClient()
        pd_cli = ProcessDispatcherServiceClient()

        proc_def = ProcessDefinition()
        proc_def.executable['module'] = 'ion.processes.data.ingestion.ingestion_cache'
        proc_def.executable['class'] = 'IngestionCache'
        proc_def_id = pd_cli.create_process_definition(process_definition=proc_def)

        xs_dot_xp = CFG.core_xps.science_data
        try:
            self.XS, xp_base = xs_dot_xp.split('.')
            self.XP = '.'.join([bootstrap.get_sys_name(), xp_base])
        except ValueError:
            raise StandardError('Invalid CFG for core_xps.science_data: "%s"; must have "xs.xp" structure' % xs_dot_xp)

        subscription_id = pubsub_cli.create_subscription(query=ExchangeQuery(), exchange_name='ingestion_aggregate')

        config = {
            'couch_storage' : {
                'datastore_name' : self.CFG.get_safe('process.datastore_name','dm_cache'),
                'datastore_profile' : self.CFG.get_safe('process.datastore_profile','SCIDATA')
            }
        }

        transform_id = tms_cli.create_transform(
            name='ingestion_cache',
            description='Ingestion that compiles an aggregate of metadata',
            in_subscription_id=subscription_id,
            process_definition_id=proc_def_id,
            configuration=config
        )


        tms_cli.activate_transform(transform_id=transform_id)<commit_msg>try this again...<commit_after>
"""
@author <NAME>
@file ion/processes/data/ingestion/ingestion_launcher.py
@description Ingestion Launcher
"""
from interface.objects import ProcessDefinition, ExchangeQuery
from interface.services.cei.iprocess_dispatcher_service import ProcessDispatcherServiceClient
from interface.services.dm.ipubsub_management_service import PubsubManagementServiceClient
from pyon.core import bootstrap
from pyon.service.service import BaseService
from interface.services.dm.itransform_management_service import TransformManagementServiceClient
from pyon.util.config import CFG
from ion.processes.data.ingestion.ingestion_cache import CACHE_DATASTORE_NAME

class CacheLauncher(BaseService):
    def on_start(self):
        super(CacheLauncher,self).on_start()
        tms_cli = TransformManagementServiceClient()
        pubsub_cli = PubsubManagementServiceClient()
        pd_cli = ProcessDispatcherServiceClient()
        dname = CACHE_DATASTORE_NAME
        number_of_workers = self.CFG.get_safe('process.number_of_workers')

        proc_def = ProcessDefinition()
        proc_def.executable['module'] = 'ion.processes.data.ingestion.ingestion_cache'
        proc_def.executable['class'] = 'IngestionCache'
        proc_def_id = pd_cli.create_process_definition(process_definition=proc_def)

        xs_dot_xp = CFG.core_xps.science_data
        try:
            self.XS, xp_base = xs_dot_xp.split('.')
            self.XP = '.'.join([bootstrap.get_sys_name(), xp_base])
        except ValueError:
            raise StandardError('Invalid CFG for core_xps.science_data: "%s"; must have "xs.xp" structure' % xs_dot_xp)

        subscription_id = pubsub_cli.create_subscription(query=ExchangeQuery(), exchange_name='ingestion_cache')

        config = {
            'couch_storage' : {
                'datastore_name' : dname,
                'datastore_profile' : 'SCIDATA'
            }
        }

        for i in xrange(number_of_workers):

            transform_id = tms_cli.create_transform(
                name='ingestion_cache%d' % i,
                description='Ingestion that compiles an aggregate of metadata',
                in_subscription_id=subscription_id,
                process_definition_id=proc_def_id,
                configuration=config
            )


            tms_cli.activate_transform(transform_id=transform_id)
{'fraction_non_alphanumeric': 0.08528831396623547, 'fraction_numerical': 0.0, 'mean_word_length': 3.455078125, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '3129635', 'n_tokens_mistral': 1343, 'n_tokens_neox': 1286, 'n_words': 216}