# Path: trtools/tools/processing.py
# class ParallelDataProcessor(DataProcessor):
# 
#     def process(self, func=None, *args, **kwargs):
#         batch = self.jobs
#         return self.process_parallel(batch, func, *args, **kwargs)
# 
#     def process_parallel(self, batch, func, num_consumers=None, verbose=False, process_vars=None):
#         tasks = self.generate_tasks(func, batch)
#         result_wrap = result_handler = self.result_handler
#         if result_handler:
#             # split up here so we don't have to job, data = result
#             result_wrap = lambda result: result_handler(result[0], result[1])
# 
#         parallel.farm(tasks, num_consumers=num_consumers, verbose=verbose,
#                             result_handler=result_wrap, process_vars=process_vars)
#         return result_handler
# 
#     def result_wrapper(self, result):
#         """
#             Just splits result in job / data
#         """
#         job, data = result
#         return self.result_handler(job, data)
# 
# class DataProcessor(object):
#     """
#         The basic idea is we setup a DataProcessor which is a mgr and jobs
#         In finance data that would be something like EOD and a list of symbols
# 
#         Note: mgr is really just a callable, so if your data/process step can be
#         better expressed in one function, DataPanel supports not calling the mgr
#     """
#     def __init__(self, jobs=None, mgr=None, result_handler=missing):
#         self.mgr = mgr
#         self.jobs = jobs
# 
#         if result_handler is missing:
#             result_handler = AggregateResultStore()
#         self.result_handler = result_handler
# 
#     def add_jobs(self, new_jobs):
#         self.jobs.extend(new_jobs)
# 
#     def process(self, func=None, *args, **kwargs):
#         wrap_func = func
#         if isinstance(func, str): # easy support of calling methods on data
#             wrap_func = lambda df: getattr(df, func)()
# 
#         batch = self.jobs
#         return self._process(batch, wrap_func, *args, **kwargs)
# 
#     def _process(self, batch, func, *args, **kwargs):
#         tasks = self.generate_tasks(func, batch)
#         result_handler = self.result_handler
#         for task in tasks:
#             job, data = task()
#             if result_handler:
#                 result_handler(job, data)
#         return result_handler
# 
#     def generate_tasks(self, func, batch):
#         tasks = [DataPanelTask(func, job, self.mgr) for job in batch]
#         return tasks
# 
#     def __getstate__(self): return self.__dict__
#     def __setstate__(self, d): self.__dict__.update(d)
# 
# Path: trtools/common.py



import operator
from trtools.tools.processing import ParallelDataProcessor, DataProcessor
from trtools.common import missing
    store_key is if the job needs translation
    """
    def __init__(self, jobs, store=None, mgr=None, job_trans=None, store_key=None, result_handler=missing):
        if job_trans is None:
            job_trans = lambda x: x

        self.job_trans = job_trans
        self.jobs = job_trans(jobs)
        self.mgr = mgr
        self.store = store

        if result_handler == missing:
            result_handler = StoreResultHandler(store, store_key)
        self.handler = result_handler

    def process(self, func, refresh=False, num=None, parallel=True, *args, **kwargs):
        if refresh:  
            self.store.delete_all()

        jobs = self.remaining_jobs()

        if num is None:
            num = len(jobs)
        jobs = jobs[:num]

        processor = self.get_processor(jobs, parallel=parallel)
        processor.process(func, *args, **kwargs)

    def get_processor(self, jobs, parallel=True):
        if parallel:

# Path: trtools/tools/processing.py
# class ParallelDataProcessor(DataProcessor):
# 
#     def process(self, func=None, *args, **kwargs):
#         batch = self.jobs
#         return self.process_parallel(batch, func, *args, **kwargs)
# 
#     def process_parallel(self, batch, func, num_consumers=None, verbose=False, process_vars=None):
#         tasks = self.generate_tasks(func, batch)
#         result_wrap = result_handler = self.result_handler
#         if result_handler:
#             # split up here so we don't have to job, data = result
#             result_wrap = lambda result: result_handler(result[0], result[1])
# 
#         parallel.farm(tasks, num_consumers=num_consumers, verbose=verbose,
#                             result_handler=result_wrap, process_vars=process_vars)
#         return result_handler
# 
#     def result_wrapper(self, result):
#         """
#             Just splits result in job / data
#         """
#         job, data = result
#         return self.result_handler(job, data)
# 
# class DataProcessor(object):
#     """
#         The basic idea is we setup a DataProcessor which is a mgr and jobs
#         In finance data that would be something like EOD and a list of symbols
# 
#         Note: mgr is really just a callable, so if your data/process step can be
#         better expressed in one function, DataPanel supports not calling the mgr
#     """
#     def __init__(self, jobs=None, mgr=None, result_handler=missing):
#         self.mgr = mgr
#         self.jobs = jobs
# 
#         if result_handler is missing:
#             result_handler = AggregateResultStore()
#         self.result_handler = result_handler
# 
#     def add_jobs(self, new_jobs):
#         self.jobs.extend(new_jobs)
# 
#     def process(self, func=None, *args, **kwargs):
#         wrap_func = func
#         if isinstance(func, str): # easy support of calling methods on data
#             wrap_func = lambda df: getattr(df, func)()
# 
#         batch = self.jobs
#         return self._process(batch, wrap_func, *args, **kwargs)
# 
#     def _process(self, batch, func, *args, **kwargs):
#         tasks = self.generate_tasks(func, batch)
#         result_handler = self.result_handler
#         for task in tasks:
#             job, data = task()
#             if result_handler:
#                 result_handler(job, data)
#         return result_handler
# 
#     def generate_tasks(self, func, batch):
#         tasks = [DataPanelTask(func, job, self.mgr) for job in batch]
#         return tasks
# 
#     def __getstate__(self): return self.__dict__
#     def __setstate__(self, d): self.__dict__.update(d)
# 
# Path: trtools/common.py



# Path: trtools/tools/datapanel.py
import operator
from trtools.tools.processing import ParallelDataProcessor, DataProcessor
from trtools.common import missing

    store_key is if the job needs translation
    """
    def __init__(self, jobs, store=None, mgr=None, job_trans=None, store_key=None, result_handler=missing):
        if job_trans is None:
            job_trans = lambda x: x

        self.job_trans = job_trans
        self.jobs = job_trans(jobs)
        self.mgr = mgr
        self.store = store

        if result_handler == missing:
            result_handler = StoreResultHandler(store, store_key)
        self.handler = result_handler

    def process(self, func, refresh=False, num=None, parallel=True, *args, **kwargs):
        if refresh:  
            self.store.delete_all()

        jobs = self.remaining_jobs()

        if num is None:
            num = len(jobs)
        jobs = jobs[:num]

        processor = self.get_processor(jobs, parallel=parallel)
        processor.process(func, *args, **kwargs)

    def get_processor(self, jobs, parallel=True):
        if parallel:

            processor = ParallelDataProcessor(jobs, result_handler=self.handler, 