Adding a simple local web api to report the plugin's state.

Author: QuinnDamerell

.vscode/settings.json

@@ -131,6 +131,7 @@
         "localauth",
         "localfs",
         "localip",
+        "localwebapi",
         "Mailsail",
         "mainbaord",
         "Mainboard",

bambu_octoeverywhere/bambuclient.py

@@ -12,6 +12,7 @@
 
 from linux_host.config import Config
 from linux_host.networksearch import NetworkSearch
+from linux_host.localwebapi import LocalWebApi
 
 from .bambucloud import BambuCloud, LoginStatus
 from .bambumodels import BambuState, BambuVersion
@@ -207,6 +208,7 @@ def _ClientWorker(self):
                     # Random other errors.
                     Sentry.OnException(f"Failed to connect to the Bambu printer {ipOrHostname}:{self.PortStr}. We will retry in a bit.", e)
 
+            LocalWebApi.Get().SetPrinterConnectionState(False)
             # Sleep for a bit between tries.
             # The main consideration here is to not log too much when the printer is off. But we do still want to connect quickly, when it's back on.
             # Note that the system might also do a printer scan after many failed attempts, which can be CPU intensive.
@@ -332,6 +334,7 @@ def _OnSubscribe(self, client:Any, userdata:Any, mid:Any, reason_code_list:List[
 
             # At this point, we know the connection was successful, the access code is correct, and the SN is correct.
             self.ConsecutivelyFailedConnectionAttempts = 0
+            LocalWebApi.Get().SetPrinterConnectionState(True)
 
             # Sub success! Force a full state sync.
             self._ForceStateSyncAsync()

bambu_octoeverywhere/bambuhost.py

@@ -25,6 +25,7 @@
 from linux_host.secrets import Secrets
 from linux_host.version import Version
 from linux_host.logger import LoggerInit
+from linux_host.localwebapi import LocalWebApi
 
 from .bambucloud import BambuCloud
 from .bambuclient import BambuClient
@@ -115,6 +116,9 @@ def RunBlocking(self, configPath:str, localStorageDir:str, repoRoot:str, devConf
             # Init the mdns client
             MDns.Init(self.Logger, localStorageDir)
 
+            # Init the local web api. This will only start a thread if it's setup to run in the config.
+            LocalWebApi.Init(self.Logger, printerId, self.Config)
+
             # Init device id
             DeviceId.Init(self.Logger)
 
@@ -258,6 +262,7 @@ def ShowUiPopup(self, title:str, text:str, msgType:str, actionText:Optional[str]
     #
     def OnPrimaryConnectionEstablished(self, octoKey:str, connectedAccounts:List[str]) -> None:
         self.Logger.info("Primary Connection To OctoEverywhere Established - We Are Ready To Go!")
+        LocalWebApi.Get().OnPrimaryConnectionEstablished(len(connectedAccounts) > 0)
 
         # Give the octoKey to who needs it.
         if self.NotificationHandler is not None:

elegoo_octoeverywhere/elegooclient.py

@@ -15,6 +15,7 @@
 
 from linux_host.config import Config
 from linux_host.networksearch import NetworkSearch
+from linux_host.localwebapi import LocalWebApi
 
 from .elegoomodels import PrinterState, PrinterAttributes
 from .interfaces import IStateTranslator, IFileManager, IWebsocketMux
@@ -336,6 +337,8 @@ def _ClientWorker(self):
             except Exception as e:
                 Sentry.OnException("Elegoo client exception in main WS loop.", e)
 
+            LocalWebApi.Get().SetPrinterConnectionState(False)
+
             # Sleep for a bit between tries.
             # The main consideration here is to not log too much when the printer is off. But we do still want to connect quickly, when it's back on.
             # Note that the system might also do a printer scan after many failed attempts, which can be CPU intensive.
@@ -365,6 +368,7 @@ def _CleanupStateOnDisconnect(self):
     # Fired when the websocket is connected.
     def _OnWsConnect(self, ws:IWebSocketClient):
         self.Logger.info("Connection to the Elegoo printer established!")
+        LocalWebApi.Get().SetPrinterConnectionState(True)
 
         # Set the connected flag now, so we can send messages.
         self.WebSocketConnected = True

elegoo_octoeverywhere/elegoohost.py

@@ -24,7 +24,7 @@
 from linux_host.secrets import Secrets
 from linux_host.version import Version
 from linux_host.logger import LoggerInit
-
+from linux_host.localwebapi import LocalWebApi
 
 from .slipstream import Slipstream
 from .elegooclient import ElegooClient
@@ -116,6 +116,9 @@ def RunBlocking(self, configPath:str, localStorageDir:str, repoRoot:str, devConf
             # Init the mdns client
             MDns.Init(self.Logger, localStorageDir)
 
+            # Init the local web api. This will only start a thread if it's setup to run in the config.
+            LocalWebApi.Init(self.Logger, printerId, self.Config)
+
             # Init device id
             DeviceId.Init(self.Logger)
 
@@ -257,6 +260,7 @@ def ShowUiPopup(self, title:str, text:str, msgType:str, actionText:Optional[str]
     #
     def OnPrimaryConnectionEstablished(self, octoKey:str, connectedAccounts:List[str]) -> None:
         self.Logger.info("Primary Connection To OctoEverywhere Established - We Are Ready To Go!")
+        LocalWebApi.Get().OnPrimaryConnectionEstablished(len(connectedAccounts) > 0)
 
         # Give the octoKey to who needs it.
         self.NotificationHandler.SetOctoKey(octoKey)

linux_host/config.py

@@ -30,7 +30,8 @@ class Config:
     GeneralSection = "general"
     GeneralBedCooldownThresholdTempC = "bed_cooldown_threshold_temp_celsius"
     GeneralBedCooldownThresholdTempCDefault = 40.0
-
+    GeneralLocalWebApiPort = "local_web_api_port"
+    GeneralLocalWebApiBindIp = "local_web_api_bind_ip"
 
     #
     # Used for the local Moonraker plugin and companions.

linux_host/localwebapi.py

@@ -0,0 +1,121 @@
+import json
+import time
+import logging
+import threading
+
+from http.server import HTTPServer, BaseHTTPRequestHandler
+
+from octoeverywhere.sentry import Sentry
+
+from linux_host.config import Config
+
+# Creates a simple web server for plugins and systems to use.
+class LocalWebApi:
+
+    # The singleton instance of the LocalWebApi.
+    _Instance:"LocalWebApi" = None #pyright: ignore[reportAssignmentType]
+
+    @staticmethod
+    def Init(logger:logging.Logger, pluginId:str, config:Config) -> None:
+        LocalWebApi._Instance = LocalWebApi(logger, pluginId, config)
+
+
+    @staticmethod
+    def Get() -> "LocalWebApi":
+        return LocalWebApi._Instance
+
+
+    def __init__(self, logger:logging.Logger, pluginId:str, config:Config) -> None:
+        self.Logger = logger
+        self.PluginId = pluginId
+        self.WebServerThread = None
+
+        # Indicates if the plugin is connected to OctoEverywhere.
+        self.IsConnectedToOctoEverywhere = False
+        # Indicates if the plugin is connected to the account.
+        self.IsAccountLinked = False
+        # Indicates if the plugin is connected to the printer.
+        self.IsConnectedToPrinter = False
+
+        # The existence of the port enables or disables the web server.
+        self.HttpPort = config.GetInt(Config.GeneralSection, Config.GeneralLocalWebApiPort, None)
+        self.HostName = config.GetStr(Config.GeneralSection, Config.GeneralLocalWebApiBindIp, None)
+        if self.HostName is None:
+            self.HostName = "0.0.0.0"
+        self._Start()
+
+
+    # Called when we are connected and we know if there's an account setup with this addon
+    def OnPrimaryConnectionEstablished(self, hasConnectedAccount:bool):
+        self.IsAccountLinked = hasConnectedAccount
+        self.IsConnectedToOctoEverywhere = True
+
+
+    # Called when the OctoEverywhere connection is lost.
+    def OnOctoEverywhereConnectionLost(self):
+        self.IsConnectedToOctoEverywhere = False
+
+
+    # Called when the printer connection changes
+    def SetPrinterConnectionState(self, isConnected:bool):
+        self.IsConnectedToPrinter = isConnected
+
+
+    def _Start(self):
+        if self.HttpPort is None or self.HttpPort <= 0:
+            self.Logger.debug("Local web api is disabled, because the port is not set.")
+            return
+        # Start the web server worker thread.
+        self.WebServerThread = threading.Thread(target=self._WebServerWorker)
+        self.WebServerThread.start()
+
+
+    def _WebServerWorker(self):
+        backoff = 0
+        while True:
+            # Try to run the webserver forever.
+            webServer = None
+            try:
+                if self.HostName is None or self.HttpPort is None:
+                    self.Logger.debug("Local web api is disabled, because the host name or port is not set.")
+                    return
+                self.Logger.debug(f"Web Server Starting {self.HostName}:{self.HttpPort}")
+                webServer = HTTPServer((self.HostName, self.HttpPort), LocalWebApi.WebServerHandler)
+                self.Logger.debug(f"Web Server Started {self.HostName}:{self.HttpPort}")
+                webServer.serve_forever()
+            except Exception as e:
+                self.Logger.error("Web server exception. "+str(e))
+
+            # If we fail, close it.
+            try:
+                if webServer is not None:
+                    webServer.server_close()
+            except Exception as e:
+                Sentry.OnException("Failed to close the addon webserver.", e)
+
+            # Try again after some time.
+            backoff = min(backoff + 1, 20)
+            time.sleep(backoff * 0.5)
+
+
+    class WebServerHandler(BaseHTTPRequestHandler):
+        # Silence access logs
+        def log_message(self, format, *args): #pylint: disable=redefined-builtin
+            pass
+
+        def do_GET(self):
+            self.send_response(200)
+            self.send_header("Content-type", "application/json")
+            self.end_headers()
+
+            # Build the JSON response
+            localWebApi = LocalWebApi.Get()
+            response = {
+                "PluginId": localWebApi.PluginId,
+                "IsConnectedToOctoEverywhere": localWebApi.IsConnectedToOctoEverywhere,
+                "IsConnectedToPrinter": localWebApi.IsConnectedToPrinter,
+                "IsAccountLinked": localWebApi.IsAccountLinked,
+            }
+
+            # Write the response
+            self.wfile.write(bytes(json.dumps(response), 'utf-8'))

moonraker_octoeverywhere/moonrakerclient.py

@@ -20,6 +20,7 @@
 from octoeverywhere.interfaces import IWebSocketClient, IPrinterStateReporter, WebSocketOpCode
 
 from linux_host.config import Config
+from linux_host.localwebapi import LocalWebApi
 
 from .filemetadatacache import FileMetadataCache
 from .moonrakercredentialmanager import MoonrakerCredentialManager
@@ -335,6 +336,7 @@ def _WebSocketSend(self, jsonStr:str) -> bool:
     # This is called on a background thread, so we can block this.
     def _OnWsOpenAndKlippyReady(self) -> None:
         self.Logger.info("Moonraker client setting up default notification hooks")
+        LocalWebApi.Get().SetPrinterConnectionState(True)
         # First, we need to setup our notification subs
         # https://moonraker.readthedocs.io/en/latest/web_api/#subscribe-to-printer-object-status
         # https://moonraker.readthedocs.io/en/latest/printer_objects/
@@ -508,6 +510,7 @@ def _WebSocketWorkerThread(self) -> None:
 
             # Inform that we lost the connection.
             self.Logger.info("Moonraker client websocket connection lost. We will try to restart it soon.")
+            LocalWebApi.Get().SetPrinterConnectionState(False)
 
             # Set that the websocket is disconnected.
             with self.WebSocketLock:

moonraker_octoeverywhere/moonrakerhost.py

@@ -25,6 +25,7 @@
 from linux_host.secrets import Secrets
 from linux_host.version import Version
 from linux_host.logger import LoggerInit
+from linux_host.localwebapi import LocalWebApi
 
 from .smartpause import SmartPause
 from .uipopupinvoker import UiPopupInvoker
@@ -139,6 +140,9 @@ def RunBlocking(self, klipperConfigDir:str, isCompanionMode:bool, localStorageDi
             # Init the mdns client
             MDns.Init(self.Logger, localStorageDir)
 
+            # Init the local web api. This will only start a thread if it's setup to run in the config.
+            LocalWebApi.Init(self.Logger, printerId, self.Config)
+
             # Init device id
             DeviceId.Init(self.Logger)
 
@@ -303,6 +307,7 @@ def Rekey(self, reason:str) -> None:
     #
     def OnPrimaryConnectionEstablished(self, octoKey:str, connectedAccounts:List[str]) -> None:
         self.Logger.info("Primary Connection To OctoEverywhere Established - We Are Ready To Go!")
+        LocalWebApi.Get().OnPrimaryConnectionEstablished(len(connectedAccounts) > 0)
 
         # Check if this printer is unlinked, if so add a message to the log to help the user setup the printer if desired.
         # This would be if the skipped the printer link or missed it in the setup script.