3D-Druck/OctoPrint-BambuPrinter
1
0
Fork 0
Code Pull Requests Actions 1 Packages Releases 1 Activity

0.1.8rc11

Browse Source 
update pybambu/bambu_cloud.py from upstream HA project
This commit is contained in:
jneilliii 2024-11-14 23:35:03 -05:00
parent b54e372342
commit 98ab94b371
2 changed files with 102 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

128
octoprint_bambu_printer/printer/pybambu/bambu_cloud.py
View File

@ -3,7 +3,11 @@ from __future__ import annotations
import base64 import base64
import json import json
from curl_cffi import requests curl_available = True
try:
from curl_cffi import requests as curl_requests
except ImportError:
curl_available = False
from dataclasses import dataclass from dataclasses import dataclass
@ -33,6 +37,9 @@ class BambuCloud:
def _get_authentication_token(self) -> dict: def _get_authentication_token(self) -> dict:
LOGGER.debug("Getting accessToken from Bambu Cloud") LOGGER.debug("Getting accessToken from Bambu Cloud")
if not curl_available:
LOGGER.debug(f"Curl library is unavailable.")
return 'curlUnavailable'
# First we need to find out how Bambu wants us to login. # First we need to find out how Bambu wants us to login.
data = { data = {
@ -41,7 +48,14 @@ class BambuCloud:
"apiError": "" "apiError": ""
} }
response = requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) response = curl_requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER)
# Check specifically for cloudflare block
if response.status_code == 403:
if 'cloudflare' in response.text:
LOGGER.error('CloudFlare blocked connection attempt')
return 'cloudFlare'
if response.status_code >= 400: if response.status_code >= 400:
LOGGER.error(f"Login attempt failed with error code: {response.status_code}") LOGGER.error(f"Login attempt failed with error code: {response.status_code}")
LOGGER.debug(f"Response: '{response.text}'") LOGGER.debug(f"Response: '{response.text}'")
@ -80,7 +94,7 @@ class BambuCloud:
} }
LOGGER.debug("Requesting verification code") LOGGER.debug("Requesting verification code")
response = requests.post(get_Url(BambuUrl.EMAIL_CODE, self._region), json=data, impersonate=IMPERSONATE_BROWSER) response = curl_requests.post(get_Url(BambuUrl.EMAIL_CODE, self._region), json=data, impersonate=IMPERSONATE_BROWSER)
if response.status_code == 200: if response.status_code == 200:
LOGGER.debug("Verification code requested successfully.") LOGGER.debug("Verification code requested successfully.")
@ -96,7 +110,7 @@ class BambuCloud:
"code": code "code": code
} }
response = requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) response = curl_requests.post(get_Url(BambuUrl.LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER)
LOGGER.debug(f"Response: {response.status_code}") LOGGER.debug(f"Response: {response.status_code}")
if response.status_code == 200: if response.status_code == 200:
@ -128,7 +142,7 @@ class BambuCloud:
"tfaCode": code "tfaCode": code
} }
response = requests.post(get_Url(BambuUrl.TFA_LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER) response = curl_requests.post(get_Url(BambuUrl.TFA_LOGIN, self._region), json=data, impersonate=IMPERSONATE_BROWSER)
LOGGER.debug(f"Response: {response.status_code}") LOGGER.debug(f"Response: {response.status_code}")
if response.status_code == 200: if response.status_code == 200:
@ -209,13 +223,11 @@ class BambuCloud:
self._password = password self._password = password
result = self._get_authentication_token() result = self._get_authentication_token()
if result == 'verifyCode': if result is None:
return result
elif result == 'tfa':
return result
elif result is None:
LOGGER.error("Unable to authenticate.") LOGGER.error("Unable to authenticate.")
return None return None
elif len(result) < 20:
return result
else: else:
self._auth_token = result self._auth_token = result
self._username = self._get_username_from_authentication_token() self._username = self._get_username_from_authentication_token()
@ -223,7 +235,7 @@ class BambuCloud:
def login_with_verification_code(self, code: str): def login_with_verification_code(self, code: str):
result = self._get_authentication_token_with_verification_code(code) result = self._get_authentication_token_with_verification_code(code)
if result == 'codeExpired' or result == 'codeIncorrect': if len(result) < 20:
return result return result
self._auth_token = result self._auth_token = result
self._username = self._get_username_from_authentication_token() self._username = self._get_username_from_authentication_token()
@ -231,16 +243,29 @@ class BambuCloud:
def login_with_2fa_code(self, code: str): def login_with_2fa_code(self, code: str):
result = self._get_authentication_token_with_2fa_code(code) result = self._get_authentication_token_with_2fa_code(code)
if len(result) < 20:
return result
self._auth_token = result self._auth_token = result
self._username = self._get_username_from_authentication_token() self._username = self._get_username_from_authentication_token()
return 'success' return 'success'
def get_device_list(self) -> dict: def get_device_list(self) -> dict:
LOGGER.debug("Getting device list from Bambu Cloud") LOGGER.debug("Getting device list from Bambu Cloud")
response = requests.get(get_Url(BambuUrl.BIND, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) if not curl_available:
LOGGER.debug(f"Curl library is unavailable.")
raise None
response = curl_requests.get(get_Url(BambuUrl.BIND, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER)
if response.status_code == 403:
if 'cloudflare' in response.text:
LOGGER.error('CloudFlare blocked connection attempt')
raise ValueError(response.status_code)
if response.status_code >= 400: if response.status_code >= 400:
LOGGER.debug(f"Received error: {response.status_code}") LOGGER.debug(f"Received error: {response.status_code}")
LOGGER.error(f"Received error: '{response.text}'")
raise ValueError(response.status_code) raise ValueError(response.status_code)
return response.json()['devices'] return response.json()['devices']
# The slicer settings are of the following form: # The slicer settings are of the following form:
@ -311,13 +336,23 @@ class BambuCloud:
# } # }
def get_slicer_settings(self) -> dict: def get_slicer_settings(self) -> dict:
LOGGER.debug("Getting slicer settings from Bambu Cloud") LOGGER.debug("DISABLED: Getting slicer settings from Bambu Cloud")
response = requests.get(get_Url(BambuUrl.SLICER_SETTINGS, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) # Disabled for now since it may be contributing to cloudflare detection speed.
if response.status_code >= 400: #
LOGGER.error(f"Slicer settings load failed: {response.status_code}") # if curl_available:
LOGGER.error(f"Slicer settings load failed: '{response.text}'") # response = curl_requests.get(get_Url(BambuUrl.SLICER_SETTINGS, self._region), headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER)
return None # if response.status_code == 403:
return response.json() # if 'cloudflare' in response.text:
# LOGGER.error(f"Cloudflare blocked slicer settings lookup.")
# return None
# if response.status_code >= 400:
# LOGGER.error(f"Slicer settings load failed: {response.status_code}")
# LOGGER.error(f"Slicer settings load failed: '{response.text}'")
# return None
# return response.json()
return None
# The task list is of the following form with a 'hits' array with typical 20 entries. # The task list is of the following form with a 'hits' array with typical 20 entries.
# #
@ -362,24 +397,46 @@ class BambuCloud:
# }, # },
def get_tasklist(self) -> dict: def get_tasklist(self) -> dict:
LOGGER.debug("Getting full task list from Bambu Cloud")
if not curl_available:
LOGGER.debug(f"Curl library is unavailable.")
raise None
url = get_Url(BambuUrl.TASKS, self._region) url = get_Url(BambuUrl.TASKS, self._region)
response = requests.get(url, headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER) response = curl_requests.get(url, headers=self._get_headers_with_auth_token(), timeout=10, impersonate=IMPERSONATE_BROWSER)
if response.status_code == 403:
if 'cloudflare' in response.text:
LOGGER.error('CloudFlare blocked connection attempt')
return None
# Check specifically for cloudflare block
if response.status_code == 403:
if 'cloudflare' in response.text:
LOGGER.error('CloudFlare blocked connection attempt')
return None
if response.status_code >= 400: if response.status_code >= 400:
LOGGER.debug(f"Received error: {response.status_code}") LOGGER.debug(f"Received error: {response.status_code}")
LOGGER.debug(f"Received error: '{response.text}'") LOGGER.debug(f"Received error: '{response.text}'")
raise ValueError(response.status_code) raise None
return response.json() return response.json()
def get_latest_task_for_printer(self, deviceId: str) -> dict: def get_latest_task_for_printer(self, deviceId: str) -> dict:
LOGGER.debug(f"Getting latest task from Bambu Cloud") LOGGER.debug(f"Getting latest task for printer from Bambu Cloud")
data = self.get_tasklist_for_printer(deviceId) try:
if len(data) != 0: data = self.get_tasklist_for_printer(deviceId)
return data[0] if len(data) != 0:
LOGGER.debug("No tasks found for printer") return data[0]
LOGGER.debug("No tasks found for printer")
except:
LOGGER.debug("Unable to make call")
return None
return None return None
def get_tasklist_for_printer(self, deviceId: str) -> dict: def get_tasklist_for_printer(self, deviceId: str) -> dict:
LOGGER.debug(f"Getting task list from Bambu Cloud") LOGGER.debug(f"Getting full task list for printer from Bambu Cloud")
tasks = [] tasks = []
data = self.get_tasklist() data = self.get_tasklist()
for task in data['hits']: for task in data['hits']:
@ -394,10 +451,21 @@ class BambuCloud:
def download(self, url: str) -> bytearray: def download(self, url: str) -> bytearray:
LOGGER.debug(f"Downloading cover image: {url}") LOGGER.debug(f"Downloading cover image: {url}")
response = requests.get(url, timeout=10, impersonate=IMPERSONATE_BROWSER) if not curl_available:
LOGGER.debug(f"Curl library is unavailable.")
return None
response = curl_requests.get(url, timeout=10, impersonate=IMPERSONATE_BROWSER)
if response.status_code == 403:
if 'cloudflare' in response.text:
LOGGER.error('CloudFlare blocked connection attempt')
raise ValueError(response.status_code)
if response.status_code >= 400: if response.status_code >= 400:
LOGGER.debug(f"Received error: {response.status_code}") LOGGER.debug(f"Received error: {response.status_code}")
LOGGER.debug(f"Received error: {response.text}")
raise ValueError(response.status_code) raise ValueError(response.status_code)
return response.content return response.content
@property @property
@ -408,6 +476,10 @@ class BambuCloud:
def auth_token(self): def auth_token(self):
return self._auth_token return self._auth_token
@property
def bambu_connected(self) -> bool:
return self._auth_token != "" and self._auth_token != None
@property @property
def cloud_mqtt_host(self): def cloud_mqtt_host(self):
return "cn.mqtt.bambulab.com" if self._region == "China" else "us.mqtt.bambulab.com" return "cn.mqtt.bambulab.com" if self._region == "China" else "us.mqtt.bambulab.com"

2
setup.py
View File

@ -14,7 +14,7 @@ plugin_package = "octoprint_bambu_printer"
plugin_name = "OctoPrint-BambuPrinter" plugin_name = "OctoPrint-BambuPrinter"
# The plugin's version. Can be overwritten within OctoPrint's internal data via __plugin_version__ in the plugin module # The plugin's version. Can be overwritten within OctoPrint's internal data via __plugin_version__ in the plugin module
plugin_version = "0.1.8rc10" plugin_version = "0.1.8rc11"
# The plugin's description. Can be overwritten within OctoPrint's internal data via __plugin_description__ in the plugin # The plugin's description. Can be overwritten within OctoPrint's internal data via __plugin_description__ in the plugin
# module # module