Rename progress bars to progress indicators

Author: lukasbindreiter

tilebox-workflows/tests/runner/test_runner.py

@@ -9,7 +9,7 @@
 from tilebox.workflows import ExecutionContext, Task
 from tilebox.workflows.cache import InMemoryCache, JobCache
 from tilebox.workflows.client import Client
-from tilebox.workflows.data import JobState, ProgressBar, RunnerContext
+from tilebox.workflows.data import JobState, ProgressIndicator, RunnerContext
 from tilebox.workflows.runner.task_runner import TaskRunner
 
 
@@ -147,7 +147,7 @@ def test_runner_with_workflow_tracking_progress() -> None:
     runner.run_all()
     job = job_client.find(job)  # load current job state
     assert job.state == JobState.COMPLETED
-    assert job.progress_bars == [ProgressBar("test", 4, 4)]
+    assert job.progress == [ProgressIndicator("test", 4, 4)]
 
 
 def replay_client(replay_file: str, assert_request_matches: bool = True) -> Client:

tilebox-workflows/tests/tasks_data.py

@@ -31,7 +31,7 @@
     Idling,
     Job,
     JobState,
-    ProgressBar,
+    ProgressIndicator,
     StorageEventTrigger,
     StorageLocation,
     StorageType,
@@ -59,12 +59,12 @@ def clusters(draw: DrawFn) -> Cluster:
 
 
 @composite
-def progress_bars(draw: DrawFn) -> ProgressBar:
-    """A hypothesis strategy for generating random progress_bars"""
+def progress_indicators(draw: DrawFn) -> ProgressIndicator:
+    """A hypothesis strategy for generating random progress indicators"""
     label = draw(one_of(alphanumerical_text(), none()))
     total = draw(integers(min_value=50, max_value=1000))
     done = draw(integers(min_value=0, max_value=total))
-    return ProgressBar(label, total, done)
+    return ProgressIndicator(label, total, done)
 
 
 @composite
@@ -134,7 +134,7 @@ def jobs(draw: DrawFn, canceled: bool | None = None) -> Job:
     if canceled is None:
         canceled = draw(booleans())
 
-    progress = draw(lists(progress_bars(), min_size=0, max_size=3))
+    progress = draw(lists(progress_indicators(), min_size=0, max_size=3))
 
     return Job(
         job_id,
@@ -167,7 +167,7 @@ def computed_tasks(draw: DrawFn) -> ComputedTask:
     task_id = draw(uuids(version=4))
     display = draw(alphanumerical_text())
     subtasks: list[TaskSubmission] = draw(lists(task_submissions(), min_size=1, max_size=10))
-    progress_updates = draw(lists(progress_bars(), min_size=0, max_size=3))
+    progress_updates = draw(lists(progress_indicators(), min_size=0, max_size=3))
 
     return ComputedTask(task_id, display, subtasks, progress_updates)
 

tilebox-workflows/tests/test_data.py

@@ -6,7 +6,7 @@
     computed_tasks,
     idling_responses,
     jobs,
-    progress_bars,
+    progress_indicators,
     storage_locations,
     task_identifiers,
     task_leases,
@@ -19,7 +19,7 @@
     ComputedTask,
     Idling,
     Job,
-    ProgressBar,
+    ProgressIndicator,
     StorageLocation,
     Task,
     TaskIdentifier,
@@ -33,9 +33,9 @@ def test_task_identifiers_to_message_and_back(task_id: TaskIdentifier) -> None:
     assert TaskIdentifier.from_message(task_id.to_message()) == task_id
 
 
-@given(progress_bars())
-def test_progress_bars_to_message_and_back(progress_bar: ProgressBar) -> None:
-    assert ProgressBar.from_message(progress_bar.to_message()) == progress_bar
+@given(progress_indicators())
+def test_progress_indicators_to_message_and_back(progress_indicator: ProgressIndicator) -> None:
+    assert ProgressIndicator.from_message(progress_indicator.to_message()) == progress_indicator
 
 
 @given(tasks())

tilebox-workflows/tilebox/workflows/data.py

@@ -108,19 +108,19 @@ def to_message(self) -> core_pb2.TaskLease:
 
 
 @dataclass(order=True)
-class ProgressBar:
+class ProgressIndicator:
     label: str | None
     total: int
     done: int
 
     @classmethod
-    def from_message(cls, progress_bar: core_pb2.ProgressBar) -> "ProgressBar":
-        """Convert a ProgressBar protobuf message to a ProgressBar object."""
-        return cls(label=progress_bar.label or None, total=progress_bar.total, done=progress_bar.done)
+    def from_message(cls, progress_indicator: core_pb2.Progress) -> "ProgressIndicator":
+        """Convert a ProgressIndicator protobuf message to a ProgressIndicator object."""
+        return cls(label=progress_indicator.label or None, total=progress_indicator.total, done=progress_indicator.done)
 
-    def to_message(self) -> core_pb2.ProgressBar:
-        """Convert a ProgressBar object to a ProgressBar protobuf message."""
-        return core_pb2.ProgressBar(label=self.label, total=self.total, done=self.done)
+    def to_message(self) -> core_pb2.Progress:
+        """Convert a ProgressIndicator object to a ProgressIndicator protobuf message."""
+        return core_pb2.Progress(label=self.label, total=self.total, done=self.done)
 
 
 @dataclass(order=True)
@@ -204,7 +204,7 @@ class Job:
     submitted_at: datetime
     started_at: datetime | None
     canceled: bool
-    progress_bars: list[ProgressBar]
+    progress: list[ProgressIndicator]
 
     @classmethod
     def from_message(cls, job: core_pb2.Job) -> "Job":  # lets use typing.Self once we require python >= 3.11
@@ -217,7 +217,7 @@ def from_message(cls, job: core_pb2.Job) -> "Job":  # lets use typing.Self once
             submitted_at=timestamp_to_datetime(job.submitted_at),
             started_at=timestamp_to_datetime(job.started_at) if job.HasField("started_at") else None,
             canceled=job.canceled,
-            progress_bars=[ProgressBar.from_message(progress_bar) for progress_bar in job.progress_bars],
+            progress=[ProgressIndicator.from_message(progress) for progress in job.progress],
         )
 
     def to_message(self) -> core_pb2.Job:
@@ -230,7 +230,7 @@ def to_message(self) -> core_pb2.Job:
             submitted_at=datetime_to_timestamp(self.submitted_at),
             started_at=datetime_to_timestamp(self.started_at) if self.started_at else None,
             canceled=self.canceled,
-            progress_bars=[progress_bar.to_message() for progress_bar in self.progress_bars],
+            progress=[progress.to_message() for progress in self.progress],
         )
 
 
@@ -303,7 +303,7 @@ class ComputedTask:
     id: UUID
     display: str | None
     sub_tasks: list[TaskSubmission]
-    progress_updates: list[ProgressBar]
+    progress_updates: list[ProgressIndicator]
 
     @classmethod
     def from_message(cls, computed_task: task_pb2.ComputedTask) -> "ComputedTask":
@@ -312,7 +312,7 @@ def from_message(cls, computed_task: task_pb2.ComputedTask) -> "ComputedTask":
             id=uuid_message_to_uuid(computed_task.id),
             display=computed_task.display,
             sub_tasks=[TaskSubmission.from_message(sub_task) for sub_task in computed_task.sub_tasks],
-            progress_updates=[ProgressBar.from_message(progress) for progress in computed_task.progress_updates],
+            progress_updates=[ProgressIndicator.from_message(progress) for progress in computed_task.progress_updates],
         )
 
     def to_message(self) -> task_pb2.ComputedTask:

tilebox-workflows/tilebox/workflows/runner/task_runner.py

@@ -28,7 +28,7 @@
 from _tilebox.grpc.error import InternalServerError
 from tilebox.datasets.sync.dataset import DatasetClient
 from tilebox.workflows.cache import JobCache
-from tilebox.workflows.data import ComputedTask, Idling, NextTaskToRun, ProgressBar, Task, TaskLease
+from tilebox.workflows.data import ComputedTask, Idling, NextTaskToRun, ProgressIndicator, Task, TaskLease
 from tilebox.workflows.interceptors import Interceptor, InterceptorType
 from tilebox.workflows.observability.logging import get_logger
 from tilebox.workflows.observability.tracing import WorkflowTracer
@@ -56,7 +56,7 @@
 _FALLBACK_JITTER_INTERVAL = timedelta(seconds=5)
 
 # Maximum number of progress bars per task, mirroring the limit on the server side
-_MAX_TASK_PROGRESS_BARS = 1000
+_MAX_TASK_PROGRESS_INDICATORS = 1000
 
 WrappedFnReturnT = TypeVar("WrappedFnReturnT")
 
@@ -217,7 +217,7 @@ def _external_interrupt_handler(self, signum: int, frame: FrameType | None) -> N
             if self._task is not None:
                 progress = []
                 if self._context is not None:
-                    progress = _mutable_progress_bars_to_progress_updates(self._context._progress_bars)  # noqa: SLF001
+                    progress = _finalize_mutable_progress_trackers(self._context._progress_indicators)  # noqa: SLF001
                 self._service.task_failed(
                     self._task,
                     RunnerShutdown("Task was interrupted"),
@@ -431,7 +431,7 @@ def _execute(self, task: Task, shutdown_context: _GracefulShutdown) -> Task | Id
 
             task_failed_retry = _retry_backoff(self._service.task_failed, stop=shutdown_context.stop_if_shutting_down())
             cancel_job = True
-            progress_updates = _mutable_progress_bars_to_progress_updates(context._progress_bars)  # noqa: SLF001
+            progress_updates = _finalize_mutable_progress_trackers(context._progress_indicators)  # noqa: SLF001
             task_failed_retry(task, e, cancel_job, progress_updates)
 
         return None
@@ -493,7 +493,7 @@ def _try_execute(
                         task.to_submission(self.tasks_to_run.cluster_slug)
                         for task in context._sub_tasks  # noqa: SLF001
                     ],
-                    progress_updates=_mutable_progress_bars_to_progress_updates(context._progress_bars),  # noqa: SLF001
+                    progress_updates=_finalize_mutable_progress_trackers(context._progress_indicators),  # noqa: SLF001
                 )
 
                 next_task_retry = _retry_backoff(self._service.next_task, stop=shutdown_context.stop_if_shutting_down())
@@ -513,7 +513,7 @@ def __init__(self, runner: TaskRunner, task: Task, job_cache: JobCache) -> None:
         self.current_task = task
         self.job_cache = job_cache
         self._sub_tasks: list[FutureTask] = []
-        self._progress_bars: dict[str | None, ProgressUpdate] = {}
+        self._progress_indicators: dict[str | None, ProgressUpdate] = {}
 
     def submit_subtask(
         self,
@@ -548,20 +548,20 @@ def submit_batch(
         )
         return self.submit_subtasks(tasks, cluster, max_retries)
 
-    def progress(self, label: str | None) -> ProgressUpdate:
+    def progress(self, label: str | None = None) -> ProgressUpdate:
         if label == "":
             label = None
 
-        if label in self._progress_bars:
-            return self._progress_bars[label]
+        if label in self._progress_indicators:
+            return self._progress_indicators[label]
 
         # this is our server side limit to prevent mistakes / abuse, so let's not allow to go beyond that already
         # client side
-        if len(self._progress_bars) > _MAX_TASK_PROGRESS_BARS:
-            raise ValueError(f"Cannot create more than {_MAX_TASK_PROGRESS_BARS} progress bars per task.")
+        if len(self._progress_indicators) > _MAX_TASK_PROGRESS_INDICATORS:
+            raise ValueError(f"Cannot create more than {_MAX_TASK_PROGRESS_INDICATORS} progress indicators per task.")
 
         progress_bar = ProgressUpdate(label)
-        self._progress_bars[label] = progress_bar
+        self._progress_indicators[label] = progress_bar
         return progress_bar
 
     @property
@@ -577,8 +577,10 @@ def _dataset(self, dataset_id: str) -> DatasetClient:
         return client.dataset(dataset_id)
 
 
-def _mutable_progress_bars_to_progress_updates(progress_bars: dict[str | None, ProgressUpdate]) -> list[ProgressBar]:
-    return [ProgressBar(label, bar._total, bar._done) for label, bar in progress_bars.items()]  # noqa: SLF001
+def _finalize_mutable_progress_trackers(
+    progress_bars: dict[str | None, ProgressUpdate],
+) -> list[ProgressIndicator]:
+    return [ProgressIndicator(label, bar._total, bar._done) for label, bar in progress_bars.items()]  # noqa: SLF001
 
 
 def _execute(task: TaskInstance, context: ExecutionContext, additional_interceptors: list[Interceptor]) -> None:

tilebox-workflows/tilebox/workflows/runner/task_service.py

@@ -8,7 +8,7 @@
     ComputedTask,
     Idling,
     NextTaskToRun,
-    ProgressBar,
+    ProgressIndicator,
     Task,
     TaskLease,
     uuid_to_uuid_message,
@@ -48,7 +48,9 @@ def next_task(self, task_to_run: NextTaskToRun | None, computed_task: ComputedTa
             return Idling.from_message(response.idling)
         return None
 
-    def task_failed(self, task: Task, error: Exception, cancel_job: bool, progress_updates: list[ProgressBar]) -> None:
+    def task_failed(
+        self, task: Task, error: Exception, cancel_job: bool, progress_updates: list[ProgressIndicator]
+    ) -> None:
         # job ouptut is limited to 1KB, so truncate the error message if necessary
         error_message = repr(error)[: (1024 - len(task.display or "None") - 1)]
         display = f"{task.display}" if error_message == "" else f"{task.display}\n{error_message}"

tilebox-workflows/tilebox/workflows/task.py

@@ -254,10 +254,10 @@ def __init__(self, label: str | None) -> None:
         self._done = 0
 
     def add(self, count: int) -> None:
-        """Add a given amount of total work to be done to the progress bar.
+        """Add a given amount of total work to be done to the progress indicator.
 
         Args:
-            count: The amount of work to add to the progress bar.
+            count: The amount of work to add to the progress indicator.
         """
         self._total += count
 
@@ -307,8 +307,8 @@ def runner_context(self) -> RunnerContext:
         """Get the runner context for the task runner executing the task."""
 
     @abstractmethod
-    def progress(self, label: str | None) -> ProgressUpdate:
-        """Get a progress bar instance for tracking job progress."""
+    def progress(self, label: str | None = None) -> ProgressUpdate:
+        """Get a progress indicator instance for tracking job progress."""
 
 
 def serialize_task(task: Task) -> bytes: