Commit 65c62fa6 authored by Pierre Paleo's avatar Pierre Paleo
Browse files

Move decorators to nabu.app.utils

parent 475ad90f
Pipeline #27043 passed with stages
in 2 minutes and 42 seconds
from os import path
import numpy as np
from .logger import LoggerOrPrint
from .utils import use_options, pipeline_step
from ..utils import check_supported
from ..io.reader import ChunkReader
from ..preproc.ccd import FlatField, Log
......@@ -103,47 +104,11 @@ class FullFieldPipeline:
self.z_max = sub_region[-1]
def _get_process_name(self):
# TODO support "incomplete" processing pipeline
return "reconstruction"
#
# Decorators and callback mechanism
#
def use_options(step_name, step_attr):
def decorator(func):
def wrapper(*args, **kwargs):
self = args[0]
if step_name not in self.processing_steps:
self.__setattr__(step_attr, None)
return
self._steps_name2component[step_name] = step_attr
self._steps_component2name[step_attr] = step_name
return func(*args, **kwargs)
return wrapper
return decorator
def pipeline_step(step_attr, step_desc):
def decorator(func):
def wrapper(*args, **kwargs):
self = args[0]
if self.__getattribute__(step_attr) is None:
return
self.logger.info(step_desc)
res = func(*args, **kwargs)
self.logger.debug("End " + step_desc)
callback = self._callbacks.get(self._steps_component2name[step_attr], None)
if callback is not None:
callback()
return res
return wrapper
return decorator
def register_callback(self, step_name, callback):
"""
Register a callback for a pipeline processing step.
......
#
# Decorators and callback mechanism
#
def use_options(step_name, step_attr):
def decorator(func):
def wrapper(*args, **kwargs):
self = args[0]
if step_name not in self.processing_steps:
self.__setattr__(step_attr, None)
return
self._steps_name2component[step_name] = step_attr
self._steps_component2name[step_attr] = step_name
return func(*args, **kwargs)
return wrapper
return decorator
def pipeline_step(step_attr, step_desc):
def decorator(func):
def wrapper(*args, **kwargs):
self = args[0]
if self.__getattribute__(step_attr) is None:
return
self.logger.info(step_desc)
res = func(*args, **kwargs)
self.logger.debug("End " + step_desc)
callback = self._callbacks.get(self._steps_component2name[step_attr], None)
if callback is not None:
callback()
return res
return wrapper
return decorator
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment