ongoing removal of old stuff

Author: sanderegg

services/director-v2/src/simcore_service_director_v2/modules/comp_scheduler/_scheduler_dask.py

@@ -38,14 +38,12 @@
 )
 from ...utils.dask_client_utils import TaskHandlers, UnixTimestamp
 from ...utils.rabbitmq import (
-    publish_service_progress,
     publish_service_resource_tracking_stopped,
     publish_service_stopped_metrics,
 )
 from ..clusters_keeper import get_or_create_on_demand_cluster
 from ..dask_client import DaskClient, PublishedComputationTask
 from ..dask_clients_pool import DaskClientsPool
-from ..db.repositories.comp_runs import CompRunsRepository
 from ..db.repositories.comp_tasks import CompTasksRepository
 from ._scheduler_base import BaseCompScheduler
 
@@ -159,27 +157,28 @@ async def _get_tasks_status(
                 use_on_demand_clusters=comp_run.use_on_demand_clusters,
                 run_metadata=comp_run.metadata,
             ) as client:
-                tasks_statuses = await client.get_tasks_status(
-                    [f"{t.job_id}" for t in tasks]
-                )
-                # process dask states
-            running_states: list[RunningState] = []
-            for dask_task_state, task in zip(tasks_statuses, tasks, strict=True):
-                if dask_task_state is DaskClientTaskState.PENDING_OR_STARTED:
-                    running_states += [
-                        (
-                            RunningState.STARTED
-                            if task.progress is not None
-                            else RunningState.PENDING
-                        )
-                    ]
-                else:
-                    running_states += [
-                        _DASK_CLIENT_TASK_STATE_TO_RUNNING_STATE_MAP.get(
-                            dask_task_state, RunningState.UNKNOWN
-                        )
-                    ]
-            return running_states
+                return await client.get_tasks_status2([f"{t.job_id}" for t in tasks])
+            #     tasks_statuses = await client.get_tasks_status(
+            #         [f"{t.job_id}" for t in tasks]
+            #     )
+            #     # process dask states
+            # running_states: list[RunningState] = []
+            # for dask_task_state, task in zip(tasks_statuses, tasks, strict=True):
+            #     if dask_task_state is DaskClientTaskState.PENDING_OR_STARTED:
+            #         running_states += [
+            #             (
+            #                 RunningState.STARTED
+            #                 if task.progress is not None
+            #                 else RunningState.PENDING
+            #             )
+            #         ]
+            #     else:
+            #         running_states += [
+            #             _DASK_CLIENT_TASK_STATE_TO_RUNNING_STATE_MAP.get(
+            #                 dask_task_state, RunningState.UNKNOWN
+            #             )
+            #         ]
+            # return running_states
 
         except ComputationalBackendOnDemandNotReadyError:
             _logger.info("The on demand computational backend is not ready yet...")
@@ -351,30 +350,30 @@ async def _task_progress_change_handler(
         with log_catch(_logger, reraise=False):
             task_progress_event = TaskProgressEvent.model_validate_json(event[1])
             _logger.debug("received task progress update: %s", task_progress_event)
-            user_id = task_progress_event.task_owner.user_id
-            project_id = task_progress_event.task_owner.project_id
-            node_id = task_progress_event.task_owner.node_id
-            comp_tasks_repo = CompTasksRepository(self.db_engine)
-            task = await comp_tasks_repo.get_task(project_id, node_id)
-            if task.progress is None:
-                task.state = RunningState.STARTED
-                task.progress = task_progress_event.progress
-                run = await CompRunsRepository(self.db_engine).get(user_id, project_id)
-                await self._process_started_tasks(
-                    [task],
-                    user_id=user_id,
-                    project_id=project_id,
-                    iteration=run.iteration,
-                    run_metadata=run.metadata,
-                )
-            else:
-                await comp_tasks_repo.update_project_task_progress(
-                    project_id, node_id, task_progress_event.progress
-                )
-            await publish_service_progress(
-                self.rabbitmq_client,
-                user_id=user_id,
-                project_id=project_id,
-                node_id=node_id,
-                progress=task_progress_event.progress,
-            )
+            # user_id = task_progress_event.task_owner.user_id
+            # project_id = task_progress_event.task_owner.project_id
+            # node_id = task_progress_event.task_owner.node_id
+            # comp_tasks_repo = CompTasksRepository(self.db_engine)
+            # task = await comp_tasks_repo.get_task(project_id, node_id)
+            # if task.progress is None:
+            #     task.state = RunningState.STARTED
+            #     task.progress = task_progress_event.progress
+            #     run = await CompRunsRepository(self.db_engine).get(user_id, project_id)
+            #     await self._process_started_tasks(
+            #         [task],
+            #         user_id=user_id,
+            #         project_id=project_id,
+            #         iteration=run.iteration,
+            #         run_metadata=run.metadata,
+            #     )
+            # else:
+            #     await comp_tasks_repo.update_project_task_progress(
+            #         project_id, node_id, task_progress_event.progress
+            #     )
+            # await publish_service_progress(
+            #     self.rabbitmq_client,
+            #     user_id=user_id,
+            #     project_id=project_id,
+            #     node_id=node_id,
+            #     progress=task_progress_event.progress,
+            # )