4.0.0 python-multibar refactoring

Author: InspiredImpact

multibar/__init__.py

@@ -1,3 +0,0 @@
-from .impl import *
-
-__all__ = impl.__all__

multibar/__main__.py

@@ -0,0 +1,82 @@
+import abc
+import typing
+
+from . import iterators
+
+
+class CalculationServiceAware(abc.ABC):
+    __slots__ = ()
+
+    @abc.abstractmethod
+    def calculate_filled_indexes(self) -> iterators.AbstractIterator[int]:
+        ...
+
+    @abc.abstractmethod
+    def calculate_unfilled_indexes(self) -> iterators.AbstractIterator[int]:
+        ...
+
+    @property
+    @abc.abstractmethod
+    def progressbar_length(self) -> int:
+        ...
+
+    @property
+    @abc.abstractmethod
+    def progress_percents(self) -> float:
+        ...
+
+    @property
+    @abc.abstractmethod
+    def start_value(self) -> typing.Union[int, float]:
+        ...
+
+    @property
+    @abc.abstractmethod
+    def end_value(self) -> typing.Union[int, float]:
+        ...
+
+    @staticmethod
+    @abc.abstractmethod
+    def get_progress_percentage(start: int, end: int, /) -> float:
+        ...
+
+
+class ProgressbarCalculationService(CalculationServiceAware):
+    def __init__(
+        self,
+        start_value: typing.Union[int, float],
+        end_value: typing.Union[int, float],
+        length: int,
+    ) -> None:
+        self._start_value = start_value
+        self._end_value = end_value
+        self._length = length
+
+    def calculate_filled_indexes(self) -> iterators.AbstractIterator[int]:
+        filled_range = range(round(self.progress_percents / (100 / self._length)))
+        return iterators.Iterator(iter(filled_range)).indexes()
+
+    def calculate_unfilled_indexes(self) -> iterators.AbstractIterator[int]:
+        filled_range_len = len(tuple(self.calculate_filled_indexes()))
+        unfilled_range = range(self._length - filled_range_len)
+        return iterators.Iterator(iter(unfilled_range)).indexes(conversion=lambda i: i + filled_range_len)
+
+    @property
+    def progressbar_length(self) -> int:
+        return self._length
+
+    @property
+    def progress_percents(self) -> float:
+        return self.get_progress_percentage(self._start_value, self._end_value)
+
+    @property
+    def start_value(self) -> typing.Union[int, float]:
+        return self._start_value
+
+    @property
+    def end_value(self) -> typing.Union[int, float]:
+        return self._end_value
+
+    @staticmethod
+    def get_progress_percentage(start: int, end: int, /) -> float:
+        return (start / end) * 100

multibar/calculation_service.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+import abc
+import typing
+
+from . import utils
+from . import writers
+from . import contracts
+from . import hooks as hooks_
+from . import types as progress_types
+
+if typing.TYPE_CHECKING:
+    from . import progressbars
+    from . import sectors
+
+
+class ProgressbarClientAware(abc.ABC):
+    @abc.abstractmethod
+    @typing.overload
+    def get_progress(
+        self,
+        start_value: int,
+        end_value: int,
+        /,
+    ) -> typing.Optional[progressbars.Progressbar[sectors.AbstractSector]]:
+        ...
+
+    @abc.abstractmethod
+    @typing.overload
+    def get_progress(
+        self,
+        start_value: int,
+        end_value: int,
+        /,
+        *,
+        length: int,
+    ) -> typing.Optional[progressbars.Progressbar[sectors.AbstractSector]]:
+        ...
+
+    @abc.abstractmethod
+    def get_progress(
+        self,
+        start_value: int,
+        end_value: int,
+        /,
+        *,
+        length: int = 20,
+    ) -> typing.Optional[progressbars.Progressbar[sectors.AbstractSector]]:
+        ...
+
+    @abc.abstractmethod
+    def set_hooks(self, hooks: hooks_.HooksAware) -> ProgressbarClientAware:
+        ...
+
+    @abc.abstractmethod
+    def update_hooks(self, hooks: hooks_.HooksAware) -> ProgressbarClientAware:
+        ...
+
+    @property
+    @abc.abstractmethod
+    def hooks(self) -> hooks_.HooksAware:
+        ...
+
+    @property
+    @abc.abstractmethod
+    def contract_manager(self) -> contracts.ContractManagerAware:
+        ...
+
+
+class ProgressbarClient(ProgressbarClientAware):
+    def __init__(
+        self,
+        *,
+        hooks: typing.Optional[hooks_.HooksAware] = None,
+        progress_writer: typing.Optional[writers.ProgressbarWriterAware] = None,
+        contract_manager: typing.Optional[contracts.ContractManagerAware] = None,
+    ) -> None:
+        self._hooks = utils.none_or(hooks_.Hooks(), hooks)
+        self._writer = utils.none_or(writers.ProgressbarWriter(), progress_writer)
+
+        if contract_manager is None:
+            contract_manager = contracts.ContractManager()
+            contract_manager.subscribe(contracts.WriteProgressContract())
+
+        self._contract_manager: contracts.ContractManagerAware = contract_manager
+
+    def _validate_contracts(self, *args: typing.Any, **kwargs: typing.Any) -> None:
+        try:
+            self._contract_manager.trigger_contracts(*args, **kwargs)
+        except Exception as exc:
+            self._hooks.trigger_on_error(*args, exc, **kwargs)
+
+    def get_progress(
+        self,
+        start_value: int,
+        end_value: int,
+        /,
+        *,
+        length: int = 20,
+    ) -> typing.Optional[progressbars.Progressbar[sectors.AbstractSector]]:
+        writer = self._writer
+        call_metadata: progress_types.ProgressMetadataType = {
+            "calculation_service_cls": writer.calculation_cls,
+            "progressbar": None,
+            "start_value": start_value,
+            "end_value": end_value,
+            "length": length,
+            "sig": writer.signature,
+        }
+
+        self._validate_contracts(self._writer, metadata=call_metadata)
+        self._hooks.trigger_pre_execution(self, metadata=call_metadata)
+
+        progressbar = self._writer.write(start_value, end_value, length=length)
+        call_metadata["progressbar"] = progressbar
+
+        self._hooks.trigger_post_execution(self, metadata=call_metadata)
+        return progressbar
+
+    def set_hooks(self, hooks: hooks_.HooksAware) -> ProgressbarClient:
+        self._hooks = hooks
+        return self
+
+    def update_hooks(self, hooks: hooks_.HooksAware) -> ProgressbarClient:
+        self._hooks = self._hooks | hooks
+        return self
+
+    @property
+    def hooks(self) -> hooks_.HooksAware:
+        return self._hooks
+
+    @property
+    def contract_manager(self) -> contracts.ContractManagerAware:
+        return self._contract_manager