Static check

* use ruff in pre-commit and travis
* fix some linter issues
* use right requests version
* editorconfig

.editorconfig

@@ -0,0 +1,16 @@
+# EditorConfig <https://EditorConfig.org>
+root = true
+
+# elementary defaults
+[*]
+charset = utf-8
+end_of_line = lf
+indent_size = tab
+indent_style = space
+insert_final_newline = false
+max_line_length = 80
+tab_width = 4
+
+# Markup files
+[{*.html,*.xml,*.yaml,*.yml}]
+tab_width = 2

.pre-commit-config.yaml

@@ -14,6 +14,10 @@ repos:
   #  rev: v2.4.4
   #  hooks:
   #    - id: pylint
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: 'v0.0.269'
+    hooks:
+    - id: ruff
   - repo: https://gitlab.com/pycqa/flake8
     rev: 3.9.2
     hooks:

.travis.yml

@@ -1,19 +1,22 @@
 ---
+dist: bionic
+
 language: python
 
 python:
- - "3.8"
  - "3.9"
+ - "3.11"
 
 install:
  - pip3 install -U types-pkg_resources types-requests
- - pip3 install -U pytest pytest-pylint pytest-doctestplus requests
-   requests_mock pytest-mypy inotify_simple func-timeout "responses==0.20"
+ - pip3 install -U ruff pytest pytest-pylint pytest-doctestplus requests
+   requests_mock pytest-mypy inotify_simple func-timeout responses
 
 before_script:
  - export TZ=Europe/Prague
 
 script:
+ - ruff check .
  - pytest -v --mypy tests --log-cli-level=10
  - pytest -v --mypy --pylint --doctest-rst prusa/connect/printer
  - pytest -v --doctest-plus prusa

ChangeLog

@@ -1,6 +1,8 @@
 ChangeLog
 =========
 
+0.7.1dev0
+
 0.7.0 (2023-05-02)
     * Added new flat struct for filesystem, old one is renamed to legacy
     * transfer_id moved to root to all events

prusa/connect/printer/__init__.py

@@ -2,32 +2,38 @@
 
     Copyright (C) 2023 PrusaResearch
 """
-
 import configparser
 import os
 import re
 from logging import getLogger
-from queue import Queue, Empty
-from time import time, sleep
-from typing import Optional, List, Any, Callable, Dict
+from queue import Empty, Queue
+from time import sleep, time
+from typing import Any, Callable, Dict, List, Optional
+
+from requests import RequestException, Response, Session
 
-from requests import Session, RequestException, Response
 # pylint: disable=redefined-builtin
 from requests.exceptions import ConnectionError
 
 from . import const, errors
 from .camera_controller import CameraController
+from .clock import ClockWatcher
 from .command import Command, CommandFailed
-from .conditions import CondState, API, TOKEN, HTTP, INTERNET
+from .conditions import API, HTTP, INTERNET, TOKEN, CondState
+from .download import DownloadMgr, Transfer
 from .files import Filesystem, InotifyHandler, delete
 from .metadata import get_metadata
-from .models import Event, Telemetry, Sheet, Register, LoopObject, \
-    CameraRegister
-from .clock import ClockWatcher
-from .download import DownloadMgr, Transfer
+from .models import (
+    CameraRegister,
+    Event,
+    LoopObject,
+    Register,
+    Sheet,
+    Telemetry,
+)
 from .util import RetryingSession, get_timestamp
 
-__version__ = "0.7.0"
+__version__ = "0.7.1dev0"
 __date__ = "02 May 2023"  # version date
 __copyright__ = "(c) 2023 Prusa 3D"
 __author_name__ = "Prusa Link Developers"
@@ -95,7 +101,7 @@ def __init__(self,
             "wifi_ssid": None,
             "hostname": None,
             "username": None,
-            "digest": None
+            "digest": None,
         }
         self.api_key: Optional[str] = None
         self.code: Optional[str] = None
@@ -138,7 +144,7 @@ def __init__(self,
         # Handler blocks communication with Connect in loop method!
         self.register_handler = default_register_handler
         self.__printed_file_cb = lambda: None
-        self.download_finished_cb = lambda Transfer: None
+        self.download_finished_cb = lambda transfer: None  # noaq: ARG005
 
         self.clock_watcher = ClockWatcher()
 
@@ -247,7 +253,7 @@ def make_headers(self, timestamp: Optional[float] = None) -> dict:
 
         headers = {
             "Fingerprint": self.fingerprint,
-            "Timestamp": str(timestamp)
+            "Timestamp": str(timestamp),
         }
         if self.token:
             headers['Token'] = self.token
@@ -382,7 +388,7 @@ def get_info(self) -> Dict[str, Any]:
             "fingerprint": self.fingerprint,
             "mbl": self.mbl,
             "sheet_settings": self.sheet_settings,
-            "active_sheet": self.active_sheet
+            "active_sheet": self.active_sheet,
         }
 
     def send_info(self, caller: Command) -> Dict[str, Any]:
@@ -657,7 +663,7 @@ def register(self):
             "sn": self.sn,
             "fingerprint": self.fingerprint,
             "printer_type": str(self.__type),
-            "firmware": self.firmware
+            "firmware": self.firmware,
         }
         res = self.conn.post(self.server + "/p/register",
                              headers=self.make_headers(),

prusa/connect/printer/camera.py

@@ -5,13 +5,22 @@
 from copy import deepcopy
 from threading import Event
 from time import time
-from typing import Any, Set, Optional, Dict
+from typing import Any, Dict, Optional, Set
 
 from requests import Session
 
-from .const import CapabilityType, TriggerScheme, DEFAULT_CAMERA_SETTINGS, \
-    NotSupported, CAMERA_WAIT_TIMEOUT, CameraBusy, CONNECTION_TIMEOUT, \
-    DriverError, ReadyTimeoutError, CAMERA_BUSY_TIMEOUT
+from .const import (
+    CAMERA_BUSY_TIMEOUT,
+    CAMERA_WAIT_TIMEOUT,
+    CONNECTION_TIMEOUT,
+    DEFAULT_CAMERA_SETTINGS,
+    CameraBusy,
+    CapabilityType,
+    DriverError,
+    NotSupported,
+    ReadyTimeoutError,
+    TriggerScheme,
+)
 from .util import make_fingerprint
 
 log = logging.getLogger("camera")
@@ -34,8 +43,11 @@ def __init__(self):
     def is_sendable(self) -> bool:
         """Is this snapshot complete and can it therefore be sent?"""
         required_attributes = [
-            self.camera_fingerprint, self.camera_token, self.camera_id,
-            self.timestamp, self.data
+            self.camera_fingerprint,
+            self.camera_token,
+            self.camera_id,
+            self.timestamp,
+            self.data,
         ]
         return all(attribute is not None for attribute in required_attributes)
 
@@ -50,7 +62,7 @@ def send(self, conn: Session, server):
             "Timestamp": str(self.timestamp),
             "Fingerprint": self.camera_fingerprint,
             "Token": self.camera_token,
-            'Content-Type': "image/jpg"
+            'Content-Type': "image/jpg",
         }
         params = {}
         if self.printer_uuid is not None:
@@ -139,6 +151,7 @@ def inner(camera: "Camera", value):
                     f" {capability_type.value} from {old_value} "
                     f"to {value}.") from exception
             camera._become_ready()
+
         return inner
 
     return value_setter_decorator

prusa/connect/printer/camera_configurator.py

@@ -3,13 +3,17 @@
 from configparser import ConfigParser
 from copy import deepcopy
 from multiprocessing import RLock
-from typing import Dict, Type, List, Set, Tuple
+from typing import Dict, List, Set, Tuple, Type
 
 from . import CameraController
 from .camera import Camera
 from .camera_driver import CameraDriver
-from .const import ConfigError, CameraAlreadyConnected, CameraConfigs, \
-    CameraNotFound
+from .const import (
+    CameraAlreadyConnected,
+    CameraConfigs,
+    CameraNotFound,
+    ConfigError,
+)
 
 log = logging.getLogger("camera_configurator")
 
@@ -25,9 +29,12 @@ class CameraConfigurator:
     connected to PrusaLink, or we would need one order for saving and another
     for the instance.
     """
+
     # pylint: disable=too-many-arguments
-    def __init__(self, camera_controller: CameraController,
-                 config: ConfigParser, config_file_path: str,
+    def __init__(self,
+                 camera_controller: CameraController,
+                 config: ConfigParser,
+                 config_file_path: str,
                  drivers: List[Type[CameraDriver]],
                  auto_detect=True) -> None:
         self.camera_controller = camera_controller

prusa/connect/printer/camera_controller.py

@@ -1,15 +1,19 @@
 """Implementation of CameraController"""
 import logging
 from functools import partial
-from queue import Queue, Empty
+from queue import Empty, Queue
 from time import time
-from typing import Set, Dict, List, Callable, Iterator, Optional
+from typing import Callable, Dict, Iterator, List, Optional, Set
 
 from requests import Session
 
-from .camera import Snapshot, Camera
-from .const import TriggerScheme, TIMESTAMP_PRECISION, \
-    TRIGGER_SCHEME_TO_SECONDS, CameraBusy
+from .camera import Camera, Snapshot
+from .const import (
+    TIMESTAMP_PRECISION,
+    TRIGGER_SCHEME_TO_SECONDS,
+    CameraBusy,
+    TriggerScheme,
+)
 from .models import CameraRegister, LoopObject
 
 log = logging.getLogger("camera_controller")
@@ -86,9 +90,8 @@ def disconnect_stuck_cameras(self):
                 to_disconnect.append(camera)
 
         for camera in to_disconnect:
-            log.warning(
-                "Camera: %s id: %s looks stuck. Disconnecting",
-                camera.name, camera.camera_id)
+            log.warning("Camera: %s id: %s looks stuck. Disconnecting",
+                        camera.name, camera.camera_id)
             camera.disconnect()
         return len(to_disconnect)
 

prusa/connect/printer/camera_driver.py

@@ -5,11 +5,11 @@
 import logging
 from copy import deepcopy
 from threading import Thread
-from typing import Set, Optional, Callable, Iterable, Dict
+from typing import Callable, Dict, Iterable, Optional, Set
 
 from . import get_timestamp
 from .camera import Resolution, Snapshot
-from .const import CapabilityType, ALWAYS_REQURIED, ConfigError, CameraConfigs
+from .const import ALWAYS_REQURIED, CameraConfigs, CapabilityType, ConfigError
 
 log = logging.getLogger("camera_driver")
 
@@ -87,13 +87,12 @@ def _get_initial_resolution(available_resolutions, config):
         """Gets the configured resolution and validates it, if invalid gives
         the highest possible one"""
         highest_resolution = sorted(available_resolutions)[-1]
-        configured_resolution = config.get(
-            "resolution", str(highest_resolution))
+        configured_resolution = config.get("resolution",
+                                           str(highest_resolution))
         if configured_resolution not in available_resolutions:
             configured_resolution = highest_resolution
         return configured_resolution
 
-
     @staticmethod
     def make_hash(plaintext_id: str) -> str:
         """Hashes the camera ID"""
@@ -179,8 +178,9 @@ def disconnect(self) -> None:
         try:
             self._disconnect()
         except Exception:  # pylint: disable=broad-except
-            log.exception("Driver %s for a camera %s threw an error while "
-                          "disconnecting", self.name, self.camera_id)
+            log.exception(
+                "Driver %s for a camera %s threw an error while "
+                "disconnecting", self.name, self.camera_id)
         if self._connected:
             # If we got stuck taking a photo and are returning late, the
             # driver is already stopped, so we must not call the disconnected

prusa/connect/printer/command.py

@@ -1,10 +1,10 @@
 """Command class representation."""
-from multiprocessing import Event
-from typing import Optional, List, Any, Dict, Callable
 from logging import getLogger
+from multiprocessing import Event
+from typing import Any, Callable, Dict, List, Optional
 
 from . import const
-from .const import PRIORITY_COMMANDS, ONE_SECOND_TIMEOUT
+from .const import ONE_SECOND_TIMEOUT, PRIORITY_COMMANDS
 from .models import EventCallback
 
 log = getLogger("connect-printer")

prusa/connect/printer/conditions.py

@@ -14,7 +14,7 @@
 import math
 from enum import Enum
 from multiprocessing import RLock
-from typing import Callable, Set, Optional
+from typing import Callable, Optional, Set
 
 cond_lock = RLock()
 

prusa/connect/printer/const.py

@@ -200,14 +200,15 @@ class CameraNotDetected(RuntimeError):
     """Raised when trying to ad a camera only by its ID and it is not
     in the detected cameras"""
 
+
 class ReadyTimeoutError(RuntimeError):
     """Raised when we time out waiting for a camera to become ready"""
 
 
 # These settings are always required to instance a camera
 ALWAYS_REQURIED = {
     "name": "The camera name. The more unique the better.",
-    "driver": "Which driver to give this setting to"
+    "driver": "Which driver to give this setting to",
 }
 
 
@@ -242,7 +243,7 @@ class TriggerScheme(Enum):
 DEFAULT_CAMERA_SETTINGS = {
     CapabilityType.TRIGGER_SCHEME.value: TriggerScheme.THIRTY_SEC,
     CapabilityType.EXPOSURE.value: 0,
-    CapabilityType.ROTATION.value: 0
+    CapabilityType.ROTATION.value: 0,
 }
 
 CameraConfigs = Dict[str, Dict[str, str]]