pylint, black, ruff

alab_management/device_manager.py

@@ -79,9 +79,9 @@ def _raise(self, *args, **kwargs) -> NoReturn:  # pylint: disable=no-self-use
 
         __str__ = __repr__
 
-        __len__ = (
-            __getattr__
-        ) = __getitem__ = __add__ = __sub__ = __eq__ = __lt__ = __gt__ = _raise
+        __len__ = __getattr__ = __getitem__ = __add__ = __sub__ = __eq__ = __lt__ = (
+            __gt__
+        ) = _raise
 
     def __init__(self, name: str, devices_client: "DevicesClient"):
         self._name = name

alab_management/device_view/device_view.py

@@ -1,10 +1,10 @@
 """Wrapper over the ``devices`` collection."""
 
+import time
 from collections.abc import Collection
 from datetime import datetime
 from enum import Enum, auto, unique
 from typing import Any, TypeVar, cast
-import time
 
 import pymongo  # type: ignore
 from bson import ObjectId  # type: ignore
@@ -548,7 +548,10 @@ def pause_device(self, device_name: str):
             },
         )
         # wait until the device pause status has been updated
-        while self.get_device(device_name=device_name)["pause_status"].name != new_pause_status:
+        while (
+            self.get_device(device_name=device_name)["pause_status"].name
+            != new_pause_status
+        ):
             time.sleep(0.5)
 
     def unpause_device(self, device_name: str):
@@ -573,7 +576,9 @@ def unpause_device(self, device_name: str):
             {"$set": update_dict},
         )
         # wait until the device pause status has been updated
-        while self.get_device(device_name=device_name)["pause_status"].name != "RELEASED":
+        while (
+            self.get_device(device_name=device_name)["pause_status"].name != "RELEASED"
+        ):
             time.sleep(0.5)
 
     def __exit__(self, exc_type, exc_value, traceback):

alab_management/experiment_view/experiment_view.py

@@ -1,9 +1,9 @@
 """A wrapper over the ``experiment`` class."""
 
+import time
 from datetime import datetime
 from enum import Enum, auto
 from typing import Any, cast
-import time
 
 from bson import ObjectId  # type: ignore
 

alab_management/sample_view/sample_view.py

@@ -1,10 +1,10 @@
 """A wrapper over the ``samples`` and ``sample_positions`` collections."""
 
 import re
+import time
 from datetime import datetime
 from enum import Enum, auto
 from typing import Any, cast
-import time
 
 import pymongo  # type: ignore
 from bson import ObjectId  # type: ignore
@@ -245,7 +245,7 @@ def is_unoccupied_position(self, position: str) -> bool:
             self.get_sample_position_status(position)[0]
             is not SamplePositionStatus.OCCUPIED
         )
-    
+
     def is_locked_position(self, position: str) -> bool:
         """Tell if a sample position is locked or not."""
         sample_position = self.get_sample_position(position=position)
@@ -321,7 +321,6 @@ def lock_sample_position(self, task_id: ObjectId, position: str):
         # Wait until the position is locked successfully
         while not self.is_locked_position(position):
             time.sleep(0.5)
-
 
     def release_sample_position(self, position: str):
         """Unlock a sample position."""

alab_management/scripts/launch_lab.py

@@ -74,9 +74,9 @@ def launch_lab(host, port, debug):
     task_launcher_thread = Thread(target=launch_task_manager)
     device_manager_thread = Thread(target=launch_device_manager)
 
-    dashboard_thread.daemon = (
-        experiment_manager_thread.daemon
-    ) = task_launcher_thread.daemon = device_manager_thread.daemon = True
+    dashboard_thread.daemon = experiment_manager_thread.daemon = (
+        task_launcher_thread.daemon
+    ) = device_manager_thread.daemon = True
 
     dashboard_thread.start()
     device_manager_thread.start()

alab_management/task_manager/resource_requester.py

@@ -96,7 +96,9 @@ def update_request_status(self, request_id: ObjectId, status: RequestStatus):
             {"_id": request_id}, {"$set": {"status": status.name}}
         )
         # wait for the request to be updated
-        while self.get_request(request_id, projection=["status"])["status"] != status.name:
+        while (
+            self.get_request(request_id, projection=["status"])["status"] != status.name
+        ):
             time.sleep(0.5)
         return value_returned
 
@@ -250,7 +252,10 @@ def release_resources(self, request_id: ObjectId) -> bool:
         )
 
         # wait for the request to be released
-        while self.get_request(request_id, projection=["status"])["status"] != "NEED_RELEASE":
+        while (
+            self.get_request(request_id, projection=["status"])["status"]
+            != "NEED_RELEASE"
+        ):
             time.sleep(0.5)
 
         return result.modified_count == 1
@@ -287,7 +292,10 @@ def release_all_resources(self):
             },
         )
         # wait for all the requests to be released
-        while any(request["status"] == RequestStatus.NEED_RELEASE.name for request in self.get_requests_by_task_id(self.task_id)):
+        while any(
+            request["status"] == RequestStatus.NEED_RELEASE.name
+            for request in self.get_requests_by_task_id(self.task_id)
+        ):
             time.sleep(0.5)
 
     def _check_request_status_loop(self):

alab_management/task_manager/task_manager.py

@@ -357,7 +357,12 @@ def _handle_requested_resources(self, request_entry: dict[str, Any]):
                 },
             )
             # wait until the parsed_sample_positions_request is updated in the database
-            while self.get_request(request_entry["_id"], projection=["parsed_sample_positions_request"])["parsed_sample_positions_request"] is None:
+            while (
+                self.get_request(
+                    request_entry["_id"], projection=["parsed_sample_positions_request"]
+                )["parsed_sample_positions_request"]
+                is None
+            ):
                 time.sleep(0.5)
 
             sample_positions = self.sample_view.request_sample_positions(
@@ -379,8 +384,12 @@ def _handle_requested_resources(self, request_entry: dict[str, Any]):
                     }
                 },
             )
-            while self.get_request(request_entry["_id"], projection=["status"])[
-                "status"].name != "ERROR":
+            while (
+                self.get_request(request_entry["_id"], projection=["status"])[
+                    "status"
+                ].name
+                != "ERROR"
+            ):
                 time.sleep(0.5)
             return
 
@@ -397,8 +406,10 @@ def _handle_requested_resources(self, request_entry: dict[str, Any]):
             },
         )
         # Wait until the status of the request is updated in the database
-        while self.get_request(request_entry["_id"], projection=["status"])[
-            "status"] != "FULFILLED":
+        while (
+            self.get_request(request_entry["_id"], projection=["status"])["status"]
+            != "FULFILLED"
+        ):
             time.sleep(0.5)
         # label the resources as occupied
         self._occupy_devices(devices=devices, task_id=task_id)

alab_management/task_view/task_view.py

@@ -3,8 +3,8 @@
 provides some convenience methods to query and manipulate the tasks collection.
 """
 
-from datetime import datetime
 import time
+from datetime import datetime
 from typing import Any, cast
 
 from bson import ObjectId

alab_management/user_input.py

@@ -102,7 +102,9 @@ def update_request_status(self, request_id: ObjectId, response: str, note: str):
             },
         )
         # Wait until the status is updated
-        while self.get_request(request_id)["status"] != UserRequestStatus.FULLFILLED.value:
+        while (
+            self.get_request(request_id)["status"] != UserRequestStatus.FULLFILLED.value
+        ):
             time.sleep(1)
 
     def retrieve_user_input(self, request_id: ObjectId) -> str: