♻️ Let checks decide when to become inactive

That will allow us to check past M115 responses.

Author: foosel

octoprint_firmware_check/__init__.py

@@ -92,7 +92,6 @@ def on_firmware_info_received(self, comm_instance, firmware_name, firmware_data)
 		                 to_unicode(firmware_name, errors="replace"),
 		                 dict((to_unicode(key, errors="replace"), to_unicode(value, errors="replace"))
 		                      for key, value in firmware_data.items()))
-		self._scan_received = False
 
 	##~~ Firmware capability hook handler
 
@@ -136,6 +135,7 @@ def get_update_information(self):
 	def _run_checks(self, check_type, *args, **kwargs):
 		changes = False
 
+		still_active = False
 		for warning_type, check_data in FIRMWARE_CHECKS.items():
 			checks = check_data.get("checks")
 			message = check_data.get("message")
@@ -179,6 +179,11 @@ def _run_checks(self, check_type, *args, **kwargs):
 				check.evaluate_timeout()
 				self._logger.debug("Check {} active? {}".format(check, check.active))
 
+			still_active = any(check.active for check in checks) or still_active
+
+		self._scan_received = still_active
+		self._logger.debug("Scanning received lines enabled? {}".format(self._scan_received))
+
 		if changes:
 			self._ping_clients()
 

octoprint_firmware_check/checks/__init__.py

@@ -55,12 +55,83 @@ def reset(self):
 		self._start_time = None
 
 
+class LineCheck(Check):
+	def __init__(self):
+		Check.__init__(self)
+		self._matches = None
+
+	def received(self, line):
+		if not line:
+			return
+
+		if self._is_match(line):
+			self._matches = True
+		elif self._is_ruled_out(line):
+			self._matches = False
+
+		self._evaluate()
+
+	def m115(self, name, data):
+		# M115 usually means we stop scanning received lines
+		if self._matches is None:
+			self._matches = False
+		self._evaluate()
+
+	def _evaluate(self):
+		if self._matches is None:
+			return
+		self._triggered = self._matches
+		self._active = False
+
+	def reset(self):
+		Check.reset(self)
+		self._matches = None
+
+	def _is_match(self, line):
+		return False
+
+	def _is_ruled_out(self, line):
+		return False
+
+
+class CapCheck(Check):
+	CAP = None
+
+	def __init__(self):
+		Check.__init__(self)
+		self.cap_seen = False
+
+	def cap(self, cap, enabled):
+		self.cap_seen = True
+		if cap == self.CAP:
+			self._triggered = self._eval(enabled)
+			self._active = False
+
+	def received(self, line):
+		if self.cap_seen and not line.startswith("Cap:"):
+			# first non cap line after cap report: cap report over, deactivate
+			self._active = False
+
+	def _eval(self, enabled):
+		# trigger when enabled
+		return enabled
+
+
+class NegativeCapCheck(CapCheck):
+	def _eval(self, enabled):
+		# trigger when NOT enabled
+		return not enabled
+
 
 class AuthorCheck(Check):
 	authors = ()
 
 	AUTHOR = "| Author: ".lower()
 
+	def m115(self, name, data):
+		# M115 usually means we stop scanning received lines
+		self._active = False
+
 	def received(self, line):
 		if not line:
 			return

octoprint_firmware_check/checks/firmware_broken.py

@@ -5,7 +5,7 @@
 __copyright__ = "Copyright (C) 2020 The OctoPrint Project - Released under terms of the AGPLv3 License"
 
 from flask_babel import gettext
-from . import Check, Severity
+from . import LineCheck, Severity
 
 class FirmwareBrokenChecks(object):
 	@classmethod
@@ -17,36 +17,15 @@ def as_dict(cls):
 		            severity=Severity.INFO)
 
 
-class CbdCheck(Check):
+class CbdCheck(LineCheck):
 	name = "cbd"
 	url = "https://faq.octoprint.org/warning-firmware-broken-cbd"
+	FRAGMENT = "CBD make it".lower()
 
-	CRITICAL_FRAGMENT = "CBD make it".lower()
+	def _is_match(self, line):
+		return self.FRAGMENT in line.lower()
 
-	def __init__(self):
-		Check.__init__(self)
-		self._fragment_matches = None
-
-	def received(self, line):
-		if not line:
-			return
-
-		lower_line = line.lower()
-		if self.CRITICAL_FRAGMENT in lower_line:
-			self._fragment_matches = True
-
-		self._evaluate()
-
-	def _evaluate(self):
-		if self._fragment_matches is None:
-			return
-		self._triggered = self._fragment_matches
-		self._active = False
-
-	def reset(self):
-		Check.reset(self)
-		self._fragment_matches = None
 
 class ZwlfCheck(CbdCheck):
 	name = "zwlf"
-	CRITICAL_FRAGMENT = "ZWLF make it".lower()
+	FRAGMENT = "ZWLF make it".lower()

octoprint_firmware_check/checks/firmware_unsafe.py

@@ -6,7 +6,7 @@
 
 from flask_babel import gettext
 from octoprint.util.version import get_comparable_version
-from . import Check, AuthorCheck, Severity
+from . import Check, NegativeCapCheck, AuthorCheck, Severity
 
 class FirmwareUnsafeChecks(object):
 	@classmethod
@@ -52,6 +52,14 @@ def __init__(self):
 		self._author_matches = None
 		self._version_matches = None
 
+	def m115(self, name, data):
+		# make sure we stop scanning once we see the response to M115
+		if self._author_matches is None:
+			self._author_matches = False
+		if self._version_matches is None:
+			self._version_matches = False
+		self._evaluate()
+
 	def received(self, line):
 		if not line:
 			return
@@ -169,13 +177,9 @@ def _extract_repetier_version(self, name):
 			version = get_comparable_version(version, base=True)
 		return version
 
-class ThermalProtectionCapCheck(Check):
+class ThermalProtectionCapCheck(NegativeCapCheck):
 	"""
 	Firmware reporting disabled THERMAL_PROTECTION capability
 	"""
 	name = "capability"
-
-	def cap(self, cap, enabled):
-		if cap == "THERMAL_PROTECTION":
-			self._triggered = not enabled
-			self._active = False
+	CAP = "THERMAL_PROTECTION"