Lint fixes and udpates.

Author: QuinnDamerell

.github/workflows/pylint.yml

@@ -42,15 +42,18 @@ jobs:
         pylint ./octoprint_octoeverywhere/
         pylint ./moonraker_octoeverywhere/
         pylint ./elegoo_octoeverywhere/
+        pylint ./elegoo_cc2_octoeverywhere/
         pylint ./bambu_octoeverywhere/
+        pylint ./prusalink_octoeverywhere/
         pylint ./linux_host/
         pylint ./py_installer/
         pylint ./docker_octoeverywhere/
+        pylint ./setup.py
 
     - name: 👉 Pyright
       run: |
         pyright
 
     - name: 🐶 Ruff
       run: |
-        ruff check
+        ruff check

.pylintrc

@@ -6,7 +6,6 @@ disable=
     C0103, # Disable bad attribute names
     W0511, # Ignore TODOs
     W0231, # Disable warns about init function calls
-    W0601, # OctoPrint global function issues
     R0901, # Too many ancestors
     R0904, # too-many-public-methods
     R0903, # too-few-public-methods
@@ -27,7 +26,6 @@ disable=
     R1724, # Unnecessary "else" after "break"
     R1715, # consider-using-get for dict gets.
     R1711, # Useless return at end of function or method (useless-return)
-    E1136, # Value 'CaseInsensitiveDict' is unsubscriptable (unsubscriptable-object) - pyright and ruff will catch these.
 
 
 
@@ -86,10 +84,6 @@ persistent=yes
 # version used to run pylint.
 # py-version=3.9
 
-# When enabled, pylint would attempt to guess common misconfiguration and emit
-# user-friendly hints instead of false-positive error messages.
-suggestion-mode=yes
-
 # Allow loading of arbitrary C extensions. Extensions are imported into the
 # active Python interpreter and may run arbitrary code.
 unsafe-load-any-extension=no
@@ -118,8 +112,6 @@ disable=raw-checker-failed,
         useless-suppression,
         deprecated-pragma,
         use-symbolic-message-instead,
-        too-many-function-args,
-        keyword-arg-before-vararg,
         use-dict-literal,
         too-many-lines
 

elegoo_cc2_octoeverywhere/__init__.py

@@ -1 +0,0 @@
-

elegoo_cc2_octoeverywhere/elegoocc2client.py

@@ -4,7 +4,7 @@
 import socket
 import threading
 import time
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, List, Optional, cast
 
 import paho.mqtt.client as mqtt
 
@@ -528,8 +528,11 @@ def _HandleResponseMessage(self, msg:Dict[str, Any]) -> None:
             return
 
         method = msg.get("method", None)
-        resultObj = msg.get("result", None)
-        if isinstance(resultObj, dict) and int(resultObj.get("error_code", 0)) == 0:
+        rawResultObj = msg.get("result", None)
+        resultObj:Optional[Dict[str, Any]] = None
+        if isinstance(rawResultObj, dict):
+            resultObj = cast(Dict[str, Any], rawResultObj)
+        if resultObj is not None and int(resultObj.get("error_code", 0)) == 0:
             if method == 1001:
                 self._HandleAttributesUpdate(resultObj)
             elif method == 1002:
@@ -547,11 +550,11 @@ def _HandleResponseMessage(self, msg:Dict[str, Any]) -> None:
 
             error = msg.get("error", None)
             if isinstance(error, dict):
-                context.SetResultAndEvent(resultObj if isinstance(resultObj, dict) else None, ResponseMsg.ELEGOO_CMD_ERROR_GENERIC, str(error))
-            elif isinstance(resultObj, dict) and int(resultObj.get("error_code", 0)) != 0:
+                context.SetResultAndEvent(resultObj, ResponseMsg.ELEGOO_CMD_ERROR_GENERIC, str(error))
+            elif resultObj is not None and int(resultObj.get("error_code", 0)) != 0:
                 context.SetResultAndEvent(resultObj, ResponseMsg.ELEGOO_CMD_ERROR_GENERIC, str(resultObj.get("error_msg", "Printer command failed.")))
             else:
-                context.SetResultAndEvent(resultObj if isinstance(resultObj, dict) else {})
+                context.SetResultAndEvent(resultObj if resultObj is not None else {})
 
 
     def _HandleStatusMessage(self, msg:Dict[str, Any], isAsyncStatus:bool) -> None:
@@ -734,6 +737,8 @@ def _GetConnectionContextToTry(self, isConnectAttemptFromEventBump:bool) -> Cc2C
             raise Exception("An IP address or hostname must be provided in the config for Elegoo CC2 Connect.")
 
         self.SerialNumber = serialNumber
+        if self.PortStr is None:
+            raise Exception("A port must be provided in the config for Elegoo CC2 Connect.")
         return Cc2ConnectionContext(configIpOrHostname, self.PortStr, serialNumber, accessCode)
 
 

elegoo_cc2_octoeverywhere/elegoocc2discovery.py

@@ -2,7 +2,7 @@
 import logging
 import socket
 import time
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, cast
 
 from octoeverywhere.sentry import Sentry
 
@@ -69,6 +69,7 @@ def _ParseDiscoveryResponse(logger:logging.Logger, data:bytes, ip:str) -> Option
             if not isinstance(result, dict):
                 logger.debug("Elegoo CC2 discovery response missing result object: %s", msg)
                 return None
+            result = cast(Dict[str, Any], result)
             return ElegooCc2DiscoveryResult(
                 ip,
                 result.get("sn", None),

elegoo_cc2_octoeverywhere/elegoocc2models.py

@@ -1,6 +1,6 @@
 import logging
 import time
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple, cast
 
 from octoeverywhere.sentry import Sentry
 
@@ -149,6 +149,7 @@ def __init__(self, logger:logging.Logger) -> None:
     def OnUpdate(self, state:Dict[str, Any]) -> None:
         machineStatus = state.get("machine_status", {})
         if isinstance(machineStatus, dict):
+            machineStatus = cast(Dict[str, Any], machineStatus)
             self.MachineStatus = self._GetIntOrNone(machineStatus, "status", self.MachineStatus)
             self.SubStatus = self._GetIntOrNone(machineStatus, "sub_status", self.SubStatus)
             exceptionStatus = machineStatus.get("exception_status", None)
@@ -158,6 +159,7 @@ def OnUpdate(self, state:Dict[str, Any]) -> None:
 
         printStatus = state.get("print_status", {})
         if isinstance(printStatus, dict):
+            printStatus = cast(Dict[str, Any], printStatus)
             self.FileName = self._GetStrOrNone(printStatus, "filename", self.FileName)
             self.TaskId = self._GetStrOrNone(printStatus, "uuid", self.TaskId)
             self.CurrentLayer = self._GetIntOrNone(printStatus, "current_layer", self.CurrentLayer)
@@ -169,21 +171,25 @@ def OnUpdate(self, state:Dict[str, Any]) -> None:
 
         extruder = state.get("extruder", {})
         if isinstance(extruder, dict):
+            extruder = cast(Dict[str, Any], extruder)
             self.HotendActual = self._GetFloatOrNone(extruder, "temperature", self.HotendActual)
             self.HotendTarget = self._GetFloatOrNone(extruder, "target", self.HotendTarget)
 
         heaterBed = state.get("heater_bed", {})
         if isinstance(heaterBed, dict):
+            heaterBed = cast(Dict[str, Any], heaterBed)
             self.BedActual = self._GetFloatOrNone(heaterBed, "temperature", self.BedActual)
             self.BedTarget = self._GetFloatOrNone(heaterBed, "target", self.BedTarget)
 
         chamber = state.get("ztemperature_sensor", state.get("chamber", {}))
         if isinstance(chamber, dict):
+            chamber = cast(Dict[str, Any], chamber)
             self.ChamberActual = self._GetFloatOrNone(chamber, "temperature", self.ChamberActual)
             self.ChamberTarget = self._GetFloatOrNone(chamber, "target", self.ChamberTarget)
 
         led = state.get("led", {})
         if isinstance(led, dict):
+            led = cast(Dict[str, Any], led)
             ledStatus = led.get("status", None)
             if ledStatus is not None:
                 self.ChamberLightOn = int(ledStatus) > 0
@@ -239,7 +245,7 @@ def GetCurrentStatus(self) -> Tuple[Optional[str], Optional[str]]:
 
         if self.MachineStatus == PrinterState.MACHINE_PRINTING:
             subState = self.SubStatus
-            subStateStr = PrinterState.SubStatusMap.get(subState, None)
+            subStateStr = PrinterState.SubStatusMap.get(subState, None) if subState is not None else None
             if subState in [
                 PrinterState.SUB_EXTRUDER_PREHEATING,
                 PrinterState.SUB_EXTRUDER_PREHEATING_2,
@@ -370,6 +376,7 @@ def OnUpdate(self, msg:Dict[str, Any]) -> None:
         self.HardwareVersion = msg.get("hardware_version", self.HardwareVersion)
         softwareVersion = msg.get("software_version", None)
         if isinstance(softwareVersion, dict):
+            softwareVersion = cast(Dict[str, Any], softwareVersion)
             self.OtaVersion = softwareVersion.get("ota_version", self.OtaVersion)
             self.McuVersion = softwareVersion.get("mcu_version", self.McuVersion)
             self.SocVersion = softwareVersion.get("soc_version", self.SocVersion)

elegoo_octoeverywhere/elegooclient.py

@@ -326,6 +326,8 @@ def _ClientWorker(self):
 
                 # Get the current IP we want to try to connect with.
                 self.WebSocketConnectionIp = self._GetIpForConnectionAttempt(isConnectAttemptFromEventBump)
+                if self.WebSocketConnectionIp is None:
+                    raise Exception("No Elegoo printer IP address is available for this connection attempt.")
 
                 # We must update the local IP to what we are trying to connect to.
                 # This var is system wides and helps other systems access the target client IP.

moonraker_octoeverywhere/systemconfigmanager.py

@@ -1,6 +1,7 @@
 import os
 import subprocess
 import logging
+from typing import Set
 
 class SystemConfigManager:
 
@@ -122,7 +123,7 @@ def EnsureAllowedServicesFile(logger:logging.Logger, klipperConfigDir:str, servi
             # Check if we are already in the file.
             with open(allowedServiceFile, "r", encoding="utf-8") as file:
                 lines = file.readlines()
-                existingEntries = set()
+                existingEntries:Set[str] = set()
                 for line in lines:
                     entry = line.strip()
                     if len(entry) != 0:

octoeverywhere/Webcam/quickcam.py

@@ -495,7 +495,7 @@ def Connect(self, url:str) -> None:
 
         # Send the auth packet
         if authData is not None:
-            self.SslSocket.write(authData)
+            self.SslSocket.sendall(authData)
 
 
     # ~~ Interface Function ~~

octoeverywhere/Webcam/webcamstreaminstance.py

@@ -77,7 +77,8 @@ def _CustomBodyStreamRead(self) -> Optional[Buffer]:
                     # TODO - I don't know why, but chrome seems to delay the rendering of the image until it gets two?
                     # This could be something in the pipeline not flushing correctly, or other things. But for now, on the first send we double the image to make it render instantly.
                     if self.IsFirstSend:
-                        imageChunkBuffer = Buffer(imageChunkBuffer.Get() + imageChunkBuffer.Get())
+                        imageChunkBufferBytes = imageChunkBuffer.ForceAsBytes()
+                        imageChunkBuffer = Buffer(imageChunkBufferBytes + imageChunkBufferBytes)
                         self.IsFirstSend = False
                         if self.Logger.isEnabledFor(logging.DEBUG):
                             self.Logger.debug("QuickCam took %s seconds from octostream stream open to first image sent.", round(time.time() - self.StreamOpenTimeSec, 3))