Operacje na obrazkach - wzorzec pipe and filters

Posiadam projekt w architekturze pipe and filters
Bardzo dużo skomplikowanych obliczeń na obrazkach lecz wszystko się sprowadza do klas w Pythonie przyjmujących obrazek na wejściu i dających jakiś obrazek na wyjściu.
Może to też być ten sam obrazek.

Obecnie tworzę w konstruktorze dwa pola na input_image i output_image lecz uważam że coś tutaj jest nie tak.
Wywołuję walidację bezpośrednio w konstruktorze - czy plik wejściowy istnieje (pomijam jak jest w pamięci ram) oraz czy wyjściowy nie istnieje.
Chciałem użyć metod __enter__ oraz __exit__ wykorzystując słówko with aby wykonywać na obrazku obliczenia.

Czy takie podejście jest możliwe do wykorzystania jeżeli obraz wejściowy i wyjściowy są różne?

Cały ten pomysł wydaje mi sie dziwny. Dla mnie klasami byłyby tutaj konkretne implementacje kroków i taki krok miałby jakąś metodę def process(input: Image)->Image a nie że te obrazki są polami.
Jeśli chodzi o context manager i with to możesz pokazać przykład jak chciałbyś zeby wyglądała składnia tego twojego przetwarzania? Coś takiego nie byłoby lepsze?

new_image = pipeline(some_image)
  .withStep(step1)
  .withStep(step2)
  .withStep(step3)
  .process()

Klasa pipeline miałaby w sobie wtedy obrazek wejściowy + listę kroków a samo przetwarzanie to byłaby jedna pętla w stylu:

def process():
    result = self.input_image
    for step in self.steps:
        result = step.process(result)
    return result

Context manager i słowo with ma sens jak chcesz np. czyścić jakieś zasoby na koniec (np. zamknąć pliki, zerwać połączenia etc)

Jesto pseudokod:
Wersja z polami:

class CompilcatedProcess:
   def __init__(self, 
                input_image:pathlib.Path, 
                output_image:pathlib.Path):
       self.__input_image:pathlib.Path = input_image
       self.__output_image:pathlib.Path = output_image
       
    def run_compicated_process(self):
        open_comand(self.__input_image)
        load_additional_information(self.__input_image)
        generate dataset
        ...
        calculations
        ...
        save_command(self.__output_image)
        save_additional_information(self.__output_image)

Wersja z __enter__ oraz __exit__:

class CompilcatedProcess:
    def __init__(self,
                 input_image: pathlib.Path,
                 output_image: pathlib.Path):
        self.__input_image: pathlib.Path = input_image
        self.__output_image: pathlib.Path = output_image

    def __enter__(self):
        open_comand(self.__input_image)
        load_additional_information(self.__input_image)
        generate dataset to field

    def __exit__(self, exc_type, exc_val, exc_tb):
        save_command(self.__output_image)
        save_additional_information(self.__output_image)

    def run_compicated_process(self):
        calculations on dataset

Tak na kolanie:

from contextlib import contextmanager
from typing import List


def pipeline():
    return PipelineBuilder()


class Image:
    def close(self):
        pass


class Step:
    def process(self, infile: Image) -> Image:
        return infile


class Pipeline:
    def __init__(self, infile: Image, outfile: str, steps: List[Step]):
        self.steps = steps
        self.outfile = outfile
        self.infile = infile

    def process(self) -> Image:
        result = self.infile
        for step in self.steps:
            result = step.process(result)
        return result

    def cleanup(self):
        self.infile.close()


class PipelineBuilder:

    def __init__(self):
        self.outfile: str = None
        self.infile: Image = None
        self.steps: List[Step] = []

    def with_input_file(self, infile: Image):
        self.infile = infile
        return self

    def with_output_file(self, outfile: str):
        self.outfile = outfile
        return self

    def with_step(self, step: Step):
        self.steps.append(step)
        return self

    @contextmanager
    def build(self) -> Pipeline:
        pipeline_object = Pipeline(self.infile, self.outfile, self.steps)
        try:
            yield pipeline_object
        finally:
            pipeline_object.cleanup()


if __name__ == '__main__':
    infile = Image()
    with (pipeline()
            .with_input_file(infile)
            .with_output_file("x.png")
            .with_step(Step())
            .build()) as pipe:
        out: Image = pipe.process()
        out.close()

Nie wiem dokładnie co i jak chcesz tam cleanupować i jak to ma działać ;]

edit: łatwiejszy przykład, same obiekty Image które same się zamkną jak wyjdą ze scope:

@contextmanager
def image(path: str) -> 'Image':
    image_object = Image(path)
    try:
        yield image_object
    finally:
        image_object.close()


class Image:
    def __init__(self, path):
        self.path = path
        # loading logic

    def close(self):
        print(f"Closing image {self.path}")
        # cleanup logic
        pass

if __name__ == '__main__':
    with image('a.png') as infile:
        print("We have open file here")
    print("And here no more")

@Shalom: dziękuję za odpowiedź. Nie rozumiem jaka jest różnica pomiędzy definiowaniem context manager jako klasę lub jako funkcję.

Spróbowałem to zaimplementować:

import logging
import pathlib
from contextlib import contextmanager

import gdal
from scipy import ndimage
from osgeo import gdal
import numpy as np


class Refinement:
    input_image_array = None
    input_dataset = None
    output_array = None

    def __init__(self, input_image: pathlib.Path, output_image: pathlib.Path):
        self.input_image: pathlib.Path = input_image
        self.output_image: pathlib.Path = output_image

    def __enter__(self):
        self.input_dataset = gdal.Open(self.input_image.__str__(), gdal.GA_ReadOnly)
        self.input_image_array = np.array(self.input_dataset.GetRasterBand(1).ReadAsArray())

    def __exit__(self, exc_type, exc_val, exc_tb):
        # method after exit
        self.add_median_filter()
        file_format = "GTiff"
        driver = gdal.GetDriverByName(file_format)
        output_dataset = driver.CreateCopy(self.output_image.__str__(), self.input_dataset, strict=0,
                                           options=['COMPRESS=LZW'])
        output_dataset.GetRasterBand(1).WriteArray(self.output_array)
        # no_data_value: int = 1
        # output_dataset.GetRasterBand(1).SetNoDataValue(no_data_value)

    @contextmanager
    def add_median_filter(self, size: int = 3):
        self.output_array = ndimage.median_filter(self.input_image_array, size)


input_file: pathlib.Path = pathlib.Path("1.tif")
output_file: pathlib.Path = pathlib.Path("2.tif")
with Refinement(input_image=input_file, output_image=output_file) as image:
    # image.add_median_filter(size=3)
    pass

Czy można stworzyć metodę, którą będziemy wywoływać na obiekcie:

with Refinement(input_image=input_file, output_image=output_file) as image:
    image.add_median_filter(size=3)

Czy można stworzyć metodę, którą będziemy wywoływać na obiekcie:

Nie rozumiem pytania. W ten kodzie który wstawiłeś wyrzuć to @contextmanager i masz normalną metodę po prostu

Nie rozumiem jaka jest różnica pomiędzy definiowaniem context manager jako klasę lub jako funkcję.

Dekorator pozwala ci zdefiniować to zwalnianie poza klasą. Możesz w ogóle nie kontrolować kodu tej klasy i nie dodasz tam sobie __exit__ ani __enter__ ;)

class Refinement:
    input_image_array = None
    input_dataset = None
    output_array = None

    def __init__(self, input_image: pathlib.Path, output_image: pathlib.Path):
        self.input_image: pathlib.Path = input_image
        self.output_image: pathlib.Path = output_image

    def __enter__(self):
        self.input_dataset = gdal.Open(self.input_image.__str__(), gdal.GA_ReadOnly)
        self.input_image_array = np.array(self.input_dataset.GetRasterBand(1).ReadAsArray())

    def __exit__(self, exc_type, exc_val, exc_tb):
        file_format = "GTiff"
        driver = gdal.GetDriverByName(file_format)
        output_dataset = driver.CreateCopy(self.output_image.__str__(), self.input_dataset, strict=0,
                                           options=['COMPRESS=LZW'])
        output_dataset.GetRasterBand(1).WriteArray(self.output_array)
        # no_data_value: int = 1
        # output_dataset.GetRasterBand(1).SetNoDataValue(no_data_value)

    @contextmanager
    def add_median_filter(self, size: int = 3):
        self.output_array = ndimage.median_filter(self.input_image_array, size)


input_file: pathlib.Path = pathlib.Path("1.tif")
output_file: pathlib.Path = pathlib.Path("2.tif")
with Refinement(input_image=input_file, output_image=output_file) as image:
    image.add_median_filter(size=3)

W jaki sposób wykonywać tą metodę w ten sposób:

with Refinement(input_image=input_file, output_image=output_file) as image:
    image.add_median_filter(size=3)

Zwraca błąd:

    image.add_median_filter(size=3)
AttributeError: 'NoneType' object has no attribute 'add_median_filter'

W jaki sposób sprawić aby metoda add_median_filter była dostępna w context managerze

aaaa brakuje ci return self w __enter__ a to @contextmanager jest tam zupełnie bez sensu.

edit: Kolejny powód żeby zrobić sobie funkcje z @contextmanager która yielduje ci obiekt i woła cleanup, a nie nadpisywać specjalne metody jak __enter__ bo jak widać łatwo coś poposuć ;)

class Refinement:
    input_image_array = None
    input_dataset = None
    output_array = None
    no_data_value = None

    def __init__(self, input_image: pathlib.Path, output_image: pathlib.Path):
        """
        Add Refinement - median filter with variable size

        Examples
        --------
        | >>> input_file: pathlib.Path = pathlib.Path("1.tif"):
        | >>> output_file: pathlib.Path = pathlib.Path("2.tif")
        | >>> with Refinement(input_image=input_file, output_image=output_file) as image:
        | >>>   image.add_median_filter(size=3)

        :param input_image: input image for processing, must exist
        :type input_image: pathlib.Path
        :param output_image: output image for processing, must not exist
        :type output_image: pathlib.Path
        """
        assert input_image.exists() is True, logging.error("Input image: " + input_image.__str__() + " not exist.")
        self.input_image: pathlib.Path = input_image
        assert output_image.exists() is False, logging.error("Output image: " + output_image.__str__() + " not exist.")
        self.output_image: pathlib.Path = output_image

    def __enter__(self):
        """
        Open image and cast to numpy array
        :return: self
        :rtype: Refinement
        """
        self.input_dataset = gdal.Open(self.input_image.__str__(), gdal.GA_ReadOnly)
        self.no_data_value = self.input_dataset.GetRasterBand(1).GetNoDataValue()
        self.input_image_array = np.array(self.input_dataset.GetRasterBand(1).ReadAsArray())
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """
        Save file to output image
        Set no data value
        :param exc_type:
        :param exc_val:
        :param exc_tb:
        :return:
        """
        file_format = "GTiff"
        driver = gdal.GetDriverByName(file_format)
        output_dataset = driver.CreateCopy(self.output_image.__str__(), self.input_dataset, strict=0,
                                           options=['COMPRESS=LZW'])
        output_dataset.GetRasterBand(1).WriteArray(self.output_array)

        # set no data value
        if self.no_data_value is not None:
            output_dataset.GetRasterBand(1).SetNoDataValue(self.no_data_value)
        assert self.output_image.exists() is True, logging.error(
            "Output image: " + self.output_image.__str__() + " calculate fail")
        logging.info("Finish adding median filter. Output image: " + self.output_image.__str__())

    def add_median_filter(self, size: int = 3):
        """
        Apply median filter to image
        :param size: size of median filter
        :type size: int
        :return: None
        :rtype: None
        """
        logging.info("Start adding median filter to: " + self.input_image.__str__())
        self.output_array = ndimage.median_filter(input=self.input_image_array, size=size)


input_file: pathlib.Path = pathlib.Path("1.tif")
output_file: pathlib.Path = pathlib.Path("2.tif")
with Refinement(input_image=input_file, output_image=output_file) as image:
    image.add_median_filter(size=3)

Czy taka implementacja jest poprawna według wzorca pipe and filters oraz użycia with w ramach context managera w pythonie?
Co byś tu zmienił? Czy opisy w docstring są poprawne i zrozumiałe?
P.S. staram się sztywno typować czy pola

    input_image_array = None
    input_dataset = None
    output_array = None

powinny posiadać silne typowanie?

1. Nie tak sie robi pola obiektu w pythonie ;) To co zrobiłeś w tej klasie to statyczne pola klasy. Python to nie Java czy C++. Pola obiektu robisz w __init__ (wg konwencji) po prostu robiąc tam self.pole=cośtam
2. Moim zdaniem to co napisałeś nijak sie ma do pipes and filters. Ty masz po prostu jakiś mutowalny obiekt, a powinieneś mieć coś takiego jak pokazywałem na początku -> że budujesz sobie pipeline z pewnymi operacjami a potem przez ten pipeline przepuszczasz sobie dane. Coś na zasadzie:

pipe = pipeline()
  .withStep(step1)
  .withStep(step2)
  .withStep(step3)
  .build()

image_out = pipe.process(image_in)
image_out2 = pipe.process(image_in2)

Ja kolegę szanują za rzeczową odpowiedź ale się nie zgadzam.

o_O bo idea pipes and filters to chyba jest taka, ze możesz sobie poskładać różne przepływy z kawałków a potem przepuścić przez nie dane.
https://www.enterpriseintegrationpatterns.com/patterns/messaging/PipesAndFilters.html

Because all component use the same external interface they can be composed into different solutions by connecting the components to different pipes. We can add new filters, omit existing ones or rearrange them into a new sequence -- all without having to change the filters themselves.

Tzn np. robisz sobie:

1. Wczytaj plik PNG rozłóż ma tablicę pixeli RGB
2. Zamień R z G
3. Zamień pixele na Image i zapisz jako PNG

I teraz przez takie coś możesz pchać sobie obrazki. Możesz też podmienić pierwszy step na "Wczytaj BMP" a wszystko inne zostawić tak jak jest. Albo zamienić ostatni step na "Zapisz BMP" i znów nie zmieniasz w ogóle istniejącego kodu poszczególnych kroków, bo one o sobie nie wiedzą w ogóle. Mozesz też dodać jakieś kroki w środek jak chcesz. Co więcej, taki pipeline jest stateless i po prostu przetwarza dane które wchodzą na wejście.

To co napisałeś w ogóle nijak sie ma do wzorca o którym mówimy. Ty zrobiłeś jakiś dziwny mutowalny obiekt i nic wiecej.

Czy możliwa jest implementacja obiektowo a nie funkcyjnie wzorca pipe and filters w pythonie?

Oczywiście że jest. Pisałem o niej od samego początku. Każdy "krok" przetwarzania to jedna klasa, która implementuje jakiś wspólny interfejs, tzn ma np. jakąś metodę def process(input: T)->T i masz klasę która trzyma sobie listę takich kroków i przepycha dane przez nie jakimś

    result = input
    for step in self.steps:
        result = step.process(result)
    return result

W klasach mam różne ilości wejścia, wyjścia, różne logiki zapisu, odczytu, parametrów, warunków oraz ilości metod
Kolejność nie jest przypadkowa.

Próbuję to zrozumieć jednak jestem sam i nie mam nad sobą "seniora"

Nie wiem na ile dobry jesteś w shellu, ale myśl o tym właśnie jak o przetwarzaniu w bashu w jakimś unixie.
find . -name '*.id' | xargs cat | sort | uniq | grep ^Kowa\.\* | cut -f 2 -d ','
Tutaj tym wspólnym wejściem i wyjściem jest tekst. Per analogiam to może być strumień bajtów z obrazka, dźwięku... cokolwiek, ale idea jest taka, że dowolną operację możesz podmienić i to dalej działa. W przypadku powyżej możesz np. grepa wymienić na taila, cut na awk itd.

W jaki sposób wyobrazić sobie wzorzec pipe and filters? Moja analogia to linia produkcyjna gdzie dany produkt przechodzi przez kilka stanowisk aż do finalnego efektu
Czy jest to dobra analogia?
@alagner rozumiem, bash nie taki straszny ale nie wiem jakbym miał to zastosować. To bardziej przetwarzanie testu i kilka operacji linijka po linijce na nim