<reponame>leodenault/mhw_optimizer<filename>progress_bar/progress_bar_process.py
from multiprocessing import Process, Value
from multiprocessing.connection import Connection

from progress_bar.progress_bar import ProgressBar

END_SIGNAL = "end_progress_bar_process"


class ProgressBarProcess(Process):
    def __init__(
        self,
        progress_bar: ProgressBar,
        end_signal_connection: Connection,
        shared_increment: Value
    ):
        super(ProgressBarProcess, self).__init__()
        self._progress_bar = progress_bar
        self._end_signal_connection = end_signal_connection
        self._shared_increment = shared_increment

    def run(self) -> None:
        current_increment = 0
        while not self._end_signal_received():
            self._progress_bar.increment(current_increment)
            with self._shared_increment.get_lock():
                current_increment = self._shared_increment.value
                self._shared_increment.value = 0

    def _end_signal_received(self) -> bool:
        return (
            self._end_signal_connection.poll()
            and self._end_signal_connection.recv() == END_SIGNAL
        )

from multiprocessing import Process, Value
from multiprocessing.connection import Connection

from progress_bar.progress_bar import ProgressBar

END_SIGNAL = "end_progress_bar_process"


class ProgressBarProcess(Process):
    def __init__(
        self,
        progress_bar: ProgressBar,
        end_signal_connection: Connection,
        shared_increment: Value
    ):
        super(ProgressBarProcess, self).__init__()
        self._progress_bar = progress_bar
        self._end_signal_connection = end_signal_connection
        self._shared_increment = shared_increment

    def run(self) -> None:
        current_increment = 0
        while not self._end_signal_received():
            self._progress_bar.increment(current_increment)
            with self._shared_increment.get_lock():
                current_increment = self._shared_increment.value
                self._shared_increment.value = 0

    def _end_signal_received(self) -> bool:
        return (
            self._end_signal_connection.poll()
            and self._end_signal_connection.recv() == END_SIGNAL
        )