<commit_before>'''
@author <NAME> <<EMAIL>>
@file ion/services/dm/inventory/data_retriever_service.py
@description Data Retriever Service
'''

from ion.core.function.transform_function import TransformFunction
from ion.processes.data.replay.replay_process import ReplayProcess
from ion.services.dm.inventory.dataset_management_service import DatasetManagementService
from ion.services.dm.utility.granule import RecordDictionaryTool

from pyon.core.exception import BadRequest 
from pyon.public import PRED, RT
from pyon.util.arg_check import validate_is_instance, validate_true
from pyon.util.containers import for_name
from pyon.util.log import log
from pyon.event.event import EventSubscriber

from interface.objects import Replay 
from interface.services.dm.idata_retriever_service import BaseDataRetrieverService

import collections
import time
import gevent

class DataRetrieverService(BaseDataRetrieverService):
    REPLAY_PROCESS = 'replay_process'

    _refresh_interval = 10
    _cache_limit      = 5
    _retrieve_cache   = collections.OrderedDict()
    _cache_lock       = gevent.coros.RLock()
    
    def on_quit(self): #pragma no cover
        #self.clients.process_dispatcher.delete_process_definition(process_definition_id=self.process_definition_id)
        self.event_subscriber.stop()
        super(DataRetrieverService,self).on_quit()


    def on_start(self):
        self.event_subscriber = EventSubscriber(event_type='DatasetModified', callback=lambda event,m : self._eject_cache(event.origin), auto_delete=True)
        self.event_subscriber.start()


    @classmethod
    def _eject_cache(cls, dataset_id):
        with cls._cache_lock:
            try:
                cls._retrieve_cache.pop(dataset_id)
            except KeyError:
                pass
    
    def define_replay(self, dataset_id='', query=None, delivery_format='', stream_id=''):
        ''' Define the stream that will contain the data from data store by streaming to an exchange name.
        query: 
          start_time: 0    The beginning timestamp
          end_time:   N    The ending timestamp
          parameters: []   The list of parameters which match the coverages parameters
          tdoa: slice()    The slice for the desired indices to be replayed
        '''

        if not dataset_id:
            raise BadRequest('(Data Retriever Service %s): No dataset provided.' % self.name)
        validate_true(stream_id, 'No stream_id provided')


        res, _  = self.clients.resource_registry.find_resources(restype=RT.ProcessDefinition,name=self.REPLAY_PROCESS,id_only=True)
        if not len(res):
            raise BadRequest('No replay process defined.')<commit_msg>Update DataRetrieverService to use managed endpoints<commit_after>'''
@author <NAME> <<EMAIL>>
@file ion/services/dm/inventory/data_retriever_service.py
@description Data Retriever Service
'''

from ion.core.function.transform_function import TransformFunction
from ion.processes.data.replay.replay_process import ReplayProcess
from ion.services.dm.inventory.dataset_management_service import DatasetManagementService
from ion.services.dm.utility.granule import RecordDictionaryTool

from pyon.core.exception import BadRequest 
from pyon.public import PRED, RT
from pyon.util.arg_check import validate_is_instance, validate_true
from pyon.util.containers import for_name
from pyon.util.log import log
from pyon.event.event import EventSubscriber

from interface.objects import Replay 
from interface.services.dm.idata_retriever_service import BaseDataRetrieverService

import collections
import time
import gevent

class DataRetrieverService(BaseDataRetrieverService):
    REPLAY_PROCESS = 'replay_process'

    _refresh_interval = 10
    _cache_limit      = 5
    _retrieve_cache   = collections.OrderedDict()
    _cache_lock       = gevent.coros.RLock()

    def on_start(self):
        self.event_subscriber = EventSubscriber(event_type='DatasetModified', callback=lambda event,m : self._eject_cache(event.origin), auto_delete=True)
        self.add_endpoint(self.event_subscriber)

    @classmethod
    def _eject_cache(cls, dataset_id):
        with cls._cache_lock:
            try:
                cls._retrieve_cache.pop(dataset_id)
            except KeyError:
                pass
    
    def define_replay(self, dataset_id='', query=None, delivery_format='', stream_id=''):
        ''' Define the stream that will contain the data from data store by streaming to an exchange name.
        query: 
          start_time: 0    The beginning timestamp
          end_time:   N    The ending timestamp
          parameters: []   The list of parameters which match the coverages parameters
          tdoa: slice()    The slice for the desired indices to be replayed
        '''

        if not dataset_id:
            raise BadRequest('(Data Retriever Service %s): No dataset provided.' % self.name)
        validate_true(stream_id, 'No stream_id provided')


        res, _  = self.clients.resource_registry.find_resources(restype=RT.ProcessDefinition,name=self.REPLAY_PROCESS,id_only=True)
        if not len(res):
            raise BadRequest('No replay process defined.')
{'fraction_non_alphanumeric': 0.07438977140643162, 'fraction_numerical': 0.0015497869043006587, 'mean_word_length': 3.634649910233393, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '524573', 'n_tokens_mistral': 1480, 'n_tokens_neox': 1441, 'n_words': 370}