Python构建一个pipeline算法框架

算法流水线框架需求：

（1）日志功能

（2）计算文件下载保持功能（针对图片）

（3）多线程/多进程功能

（4）算子相互解耦

用文件夹结构展示

----algo（算子模块）

---------__init__.py

---------algo_base.py

---------algo_database.py

---------algo_datadict.py

..............................

----data(测试数据)

--------image.raw

--------image_scene.png

----utility(工具模块)

-------utility_image.py

-------utility_log.py

----test(测试模块)

--------__init__.py

--------test_engine.py

--------test_image.py

-----config.py(配置文件)

-----core_engine.py（计算引擎）

algo_database.py

from copy import deepcopy
from weakref import ref


class AlgoDataDict:

    def __init__(self):
        self.__datadict = {}

    @property
    def datadict(self):
        return self.__datadict

    @datadict.setter
    def datadict(self, value):
        self.__datadict = deepcopy(value)

    def append(self, label, value):
        if label is not self.__datadict.keys():
            self.__datadict[label] = (value,)
        else:
            self.__datadict[label] = self.__datadict[label] + (value,)

    def get(self, labels):
        if not isinstance(labels, (tuple,set,list)):
            yield None
        else:
            for label in labels:
                if label in self.__datadict.keys():
                    yield self.__datadict[label]
                else:
                    yield None

    def clear(self):
        self.__datadict.clear()

    def remove(self, labels):
        for label in labels:
            if label in self.__datadict.keys():
                self.__datadict.pop(label)

algo_datadict.py

import threading
from weakref import WeakValueDictionary
from algo.algo_datadict import AlgoDataDict


def singleton(cls):
    _instance = {}

    def inner():
        if cls not in _instance:
            _instance[cls] = cls()

        return _instance[cls]

    return inner


@singleton
class AlgoDataBase(object):
    def __init__(self):
        self.pool = WeakValueDictionary()
        self.lock = threading.Lock()

    def create_dict(self, thread_id):
        with self.lock:
            self.pool[thread_id] = algo_dict = AlgoDataDict()
            return algo_dict

    def get_dict(self, task_id):
        with self.lock:
            if task_id in self.pool.keys():
                return self.pool.get(task_id)
            self.pool[task_id] = algo_dict = AlgoDataDict()
            return algo_dict

    def clear_dict(self, task_id):
        with self.lock:
            if task_id in self.pool.keys():
                algo_dict = self.pool.pop(task_id)
                algo_dict.clear()

from abc import ABC, abstractmethod
from copy import deepcopy
from algo.algo_database import AlgoDataBase


class AlgoBase(ABC):

    def __init__(self, algo_name, attribute, task_id):
        self._attribute = deepcopy(attribute)
        self.algo_name = deepcopy(algo_name)
        self.db = AlgoDataBase()
        self.data_dict = self.db.get_dict(task_id)

    @property
    def attribute(self):
        return self._attribute

    @attribute.setter
    def attribute(self, value):
        self._attribute = value

    @abstractmethod
    def before_run(self):
        """callback before process function"""

    @abstractmethod
    def process(self, *args, **kwargs):
        """algorithm core to process data"""

    @abstractmethod
    def after_run(self):
        """callback after process function"""

from algo.algo_base import AlgoBase
from utility.utility_log import log_enter_fun, UserError, print_error_msg, print_info
from utility.utility_image import read_image, bytes2array, show_array_in_window


class AlgoIn(AlgoBase):

    label = "input"

    def before_run(self):
        print_info(self.algo_name)

    @log_enter_fun
    def process(self, *args, **kwargs):
        """
        :param args: args[0] is path of image
                     args[1] is the row number of image
                     args[2] is the col number of image
        :param kwargs: {}
        :return: None
        """
        if len(args) < 3:
            return None
        else:
            path = args[0]
            row = args[1]
            col = args[2]
            try:
                buffer = read_image(path)
                array2 = bytes2array(buffer, row, col)
                show_array_in_window("input image", array2, 1, 1, 1)
                self.data_dict.append(self.label, array2)
            except UserError as e:
                print_error_msg(e)

    def after_run(self):
        print_info(self.algo_name)

import logging
import traceback
from functools import wraps
from datetime import datetime

global log_level


def set_log_level(level):
    global log_level
    log_level = level
    log_format = "%(levelname)s-%(asctime)s-[*] %(message)s"
    if log_level <= logging.DEBUG:
        logging.basicConfig(level=logging.DEBUG, format=log_format)
    elif log_level <= logging.INFO:
        logging.basicConfig(level=logging.INFO, format=log_format)
    else:
        logging.basicConfig(level=logging.ERROR, format=log_format)


def print_info(fmt):
    global log_level
    if log_level <= logging.INFO:
        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S,%f')[:-3]
        print('INFO-{0}-{1}'.format(now, fmt))


def log_debug(fmt):
    logging.debug(fmt)


def log_info(fmt):
    logging.info(fmt)


def log_error(fmt):
    logging.error(fmt)


def print_error_msg():
    global log_level
    if log_level <= logging.DEBUG:
        msg = traceback.format_exc()
        print(msg)


def log_enter_fun(fun):
    @wraps(fun)
    def wrapper(*args, **kwargs):
        print_info('[' + fun.__name__+']' + " enter.")
        fun(*args, **kwargs)
        print_info('['+fun.__name__+']' + " exit.")
    return wrapper


class UserError(Exception):
    def __init__(self, fmt):
        self.args = fmt

    def __str__(self):
        str_args = "".join(self.args)
        return repr("{0} at {1}".format(str_args, id(self)))

import cv2
import cv2 as cv
import numpy as np
import matplotlib.pyplot as plt
from utility.utility_log import UserError


def show_image_in_window(img, title="image"):
    cv.imshow(title, img)


def keep_image_in_window():
    while 1:
        if cv2.waitKey(0) == ord('q'):
            break


def close_image_in_window(win_name=''):
    if win_name == '':
        cv.destroyAllWindows()
    else:
        cv.destroyWindow(win_name)


def array2image(array2):
    pass


def image2array(image):
    pass


def create_figure(nrows, ncols):
    fig, axes = plt.subplots(nrows, ncols)
    return fig, axes


def show_array_in_window(title, array2, nrows, ncols, index):
    if array2 is None:
        return
    plt.figure()
    plt.subplot(nrows, ncols, index)
    plt.title(title)
    plt.imshow(array2, cmap="gray")


def keep_array_in_window():
    plt.show(block=True)


def bytes2array(buffer, row, col):
    try:
        array = np.frombuffer(buffer, dtype=np.uint16, count=row*col, offset=0)
        array2 = array.reshape(col, row)
        return array2
    except (TypeError, RuntimeError):
        raise UserError("Failed to convert buffer to image")


def array2bytes(array2):
    try:
        buffer = array2.tobytes()
        return buffer
    except (TypeError, RuntimeError):
        raise UserError("Failed to convert array2 to buffer")


def read_image(path):
    try:
        buffer = None
        with open(path, 'rb') as f:
            buffer = f.read()
        return buffer
    except (OSError, TypeError):
        raise UserError("Failed to read image")


def write_image(path, buffer):
    try:
        with open(path, 'wb') as f:
            f.write(buffer)
    except (OSError, TypeError):
        raise UserError("Failed to write image")

from concurrent.futures import ThreadPoolExecutor
from utility.utility_log import UserError, log_error
import algo.all_algo as algo
import config


class CoreEngine:

    def __init__(self):
        self.tasks = []
        self._executor = ThreadPoolExecutor(max_workers = config.global_thread_num)

    def start(self, arguments, task_id):
        try:
            task = self._executor.submit(CoreEngine.run(arguments, task_id))
            self.tasks.append({'id': task_id, 'task': task})
        except (UserError, RuntimeError) as e:
            log_error(e)

    @staticmethod
    def run(arguments, task_id):
        try:
            input_image = algo.AlgoIn("InputData", None, task_id)
            path = "./data/image.raw16"
            row = 8032
            col = 6000
            input_image.before_run()
            input_image.process(path, row, col)
            input_image.after_run()

            attribute = {'key': 1, 'key2': 2}
            awb = algo.AlgoAWB("AWB", attribute, 1)
            awb.before_run()
            awb.process()
            awb.after_run()

        except UserError as e:
            log_error(e)

import unittest
from core_engine import CoreEngine
from utility.utility_log import *
import __init__


class TestEngineFunctions(unittest.TestCase):
    def test_init(self):
        set_log_level(level=20)

        core_engine = CoreEngine()

        algo_list = [{'name': 'AlgoIn', 'attribute': {}}, {'name': 'algo_dmc', 'attribute': {}}]
        task_id = 1
        core_engine.start(algo_list, task_id)

        self.assertEqual(True, True)  # add assertion here


if __name__ == '__main__':
    unittest.main()