OOP multiprocessing on images

Question

I want to use multiprocessing to analyze several images in parallel, with my class:

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self, filename, path):
        self.filename = filename
        self.path = path
        self.input_data = None
        self.output_data = None


    def read_image(self):
        self.input_data =  cv2.imread(self.path + self.filename)[1]

    def write_image(self):
        cv2.imwrite(self.path + self.filename.split('.')[0] + '_' + self.DISPLAY_NAME + '.png', self.output_data)

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self):
        self.read_image()
        self.process()
        self.write_image()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        source = rgb_to_hsv(self.input_data)
        self.output_data = treshold_otsu(source)


class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        source = rgb_to_lab(self.input_data)
        self.output_data = global_threshold(source)


segmenter_class = {
'hsv': HSV_Segmentation,
'lab': LAB_Segmenter
}.get(procedure)

if not segmenter_class:
    raise ArgumentError("Invalid segmentation method '{}'".format(procedure))

for img in images:
    os.chdir(img_dir)
    processor =  = segmenter_class(img, img_dir, procedure)
    processor.start_pipeline()

However, I am not sure how to call the map funtion:

image_lst = os.listdir(my_image_path)

# We split the list into sublist with 5 elements because of 512 GB RAM limitation
if len(image_lst) > 4:
    nr_of_sublists = int(len(image_lst)/2.5)
    image_sub_lst  =np.array_split(image_lst, nr_of_sublists)
else:
    image_sub_lst = [image_lst]

# We do the analysis for each sublist
for sub_lst in image_sub_lst:
    print (sub_lst)
    pool = multiprocessing.Pool(8)
    
    # Call the processor 
    processor =  = segmenter_class(img, img_dir, procedure)
    processor.start_pipeline()
    # How to call map???
    pool.map(?, sub_lst)
    pool.terminate()
    

EDIT:

I tried to change the code to the comment but still getting an error:

import os
import multiprocessing

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self):

        print ('init')

    def read_image(self):
        print ('read')

    def write_image(self):
        print ('write')

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self, args):
        print ('ok starting')
        filename, path = args
        print(filename, path)
        self.process()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        print ('ok HSV')

class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        print ('ok LAB')

procedure = 'hsv'
segmenter_class = {
'hsv': HSV_Segmenter,
'lab': LabSegmenter
}.get(procedure)

images = ['01.png', '02.png', '03.png']
img_dir = 'C:/'

if __name__ == '__main__':
    pool = multiprocessing.Pool(3)
    pool.map(segmenter_class.start_pipeline, [images, img_dir])
    pool.terminate()

Error: The above exception was the direct cause of the following exception:

Traceback (most recent call last): File "C:/Users/lueck/PycharmProjects/hyphae_cmd/hyphae_cmd/multi.py", line 50, in pool.map(segmenter_class.start_pipeline, [images, img_dir]) File "C:\Users\lueck\AppData\Local\Continuum\anaconda3\envs\hyphae_env\lib\multiprocessing\pool.py", line 266, in map return self._map_async(func, iterable, mapstar, chunksize).get() File "C:\Users\lueck\AppData\Local\Continuum\anaconda3\envs\hyphae_env\lib\multiprocessing\pool.py", line 644, in get raise self._value TypeError: start_pipeline() missing 1 required positional argument: 'args'

Answer 1

You have to create list with pairs (filename, path)

data = [(img, img_dir) for img in images]

and then map will run every pair in separated process.

But you have to get args in start_pipeline

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

And you have to use () to create instance of class segmenter_class to use start_pipeline

pool.map(segmenter_class().start_pipeline, data)

---

BTW: In example code I also return result from process.

---

import os
import multiprocessing

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self):
        print('init')

    def read_image(self):
        print('read')

    def write_image(self):
        print('write')

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        print('ok HSV')
        return "result HSV"
    
class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        print('ok LAB')
        return "result LAB"

if __name__ == '__main__':

    procedure = 'hsv'
    
    segmenter_class = {
        'hsv': HSV_Segmenter,
        'lab': LabSegmenter,
    }.get(procedure)
    
    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data = [(img, img_dir) for img in images]
    
    pool = multiprocessing.Pool(3)

    # example 1

    results = pool.map(segmenter_class().start_pipeline, data)
    print('Results:', results)

    # example 2

    for result in pool.map(segmenter_class().start_pipeline, data):
        print('result:', result)

    pool.terminate()

---

EDIT:

You can also create function which gets procedure and data and then use it in map - this way every process will create own instance of procedure or you can send different procedures to different processes.

import os
import multiprocessing

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self):
        print('init')

    def read_image(self):
        print('read')

    def write_image(self):
        print('write')

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        print('ok HSV')
        return "result HSV"
    
class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        print('ok LAB')
        return "result LAB"

segmenters = {
    'hsv': HSV_Segmenter,
    'lab': LabSegmenter,
}

def start_process(args):

    procedure = args[0]
    data = args[1:]

    segmenter_class = segmenters.get(procedure)
    result = segmenter_class().start_pipeline(data)

    return result
    
if __name__ == '__main__':

    procedure = 'hsv'
    
    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data = [(procedure, img, img_dir) for img in images]
    
    pool = multiprocessing.Pool(3)

    # example 1

    results = pool.map(start_process, data)
    print('Results:', results)

    # example 2

    for result in pool.map(segmenter_class().start_pipeline, data):
        print('result:', result)

    pool.terminate()

---

Example with different procedures

if __name__ == '__main__':

    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    pool = multiprocessing.Pool(3)

    data = [('hsv', img, img_dir) for img in images]
    results = pool.map(start_process, data)
    print('Results HSV:', results)

    data = [('lab', img, img_dir) for img in images]
    results = pool.map(start_process, data)
    print('Results LAB:', results)

    pool.terminate()

And the same with one map(). There are 6 processes to start and Pool(3) so it will run only 3 process at the same time and when it will have free process then map will get next values from list and run process.

if __name__ == '__main__':

    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data_hsv = [('hsv', img, img_dir) for img in images]
    data_lab = [('lab', img, img_dir) for img in images]
    
    data = data_hsv + data_lab

    pool = multiprocessing.Pool(3)

    # example 1

    results = pool.map(start_process, data)
    print('Results:', results)

    # example 2

    for result in pool.map(start_process, data):
        print('results:', result)

    pool.terminate()

---

EDIT:

It works also with Ray

It needs only

from ray.util import multiprocessing

instead of

import multiprocessing

---

I didn't tested it with Dask, PySpark or Joblib

---

EDIT:

Example with Joblib

from joblib import Parallel, delayed

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self):
        print('init')

    def read_image(self):
        print('read')

    def write_image(self):
        print('write')

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        print('ok HSV')
        return "result HSV"

class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        print('ok LAB')
        return "result LAB"

segmenters = {
    'hsv': HSV_Segmenter,
    'lab': LabSegmenter,
}

def start_process(args):
    
    procedure = args[0]
    data = args[1:]
    
    segmenter_class = segmenters.get(procedure)
    result = segmenter_class().start_pipeline(data)
    
    return result

if __name__ == '__main__':

    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data_hsv = [('hsv', img, img_dir) for img in images]
    data_lab = [('lab', img, img_dir) for img in images]
    
    data = data_hsv + data_lab

    # --- version 1 ---

    #pool = Parallel(n_jobs=3, backend='threading')
    #pool = Parallel(n_jobs=3, backend='multiprocessing')
    pool = Parallel(n_jobs=3)
    
    # example 1
    
    results = pool( delayed(start_process)(args) for args in data )
    print('Results:', results)

    # example 2
    
    for result in pool( delayed(start_process)(args) for args in data ):
        print('result:', result)

    # --- version 2 ---
    
    #with Parallel(n_jobs=3, backend='threading') as pool:
    #with Parallel(n_jobs=3, backend='multiprocessing') as pool:
    with Parallel(n_jobs=3) as pool:

        # example 1
        
        results = pool( delayed(start_process)(args) for args in data )
        print('Results:', results)

        # example 1

        for result in pool( delayed(start_process)(args) for args in data ):
            print('result:', result)