Skip to content

Commit

Permalink
0.1.8rc11
Browse files Browse the repository at this point in the history 
update pybambu/bambu_cloud.py from upstream HA project
  • Loading branch information
@jneilliii
jneilliii committed Nov 15, 2024
1 parent b54e372 commit 98ab94b
Show file tree
Hide file tree
Showing 2 changed files with 102 additions and 30 deletions.
130 changes: 101 additions & 29 deletions octoprint_bambu_printer/printer/pybambu/bambu_cloud.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,11 @@
import base64
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

Expand Down Expand Up @@ -33,6 +37,9 @@ def _get_headers_with_auth_token(self) -> dict:

def _get_authentication_token(self) -> dict:
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.
data = {
Expand All @@ -41,7 +48,14 @@ def _get_authentication_token(self) -> dict:
"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:
LOGGER.error(f"Login attempt failed with error code: {response.status_code}")
LOGGER.debug(f"Response: '{response.text}'")
Expand Down Expand Up @@ -80,7 +94,7 @@ def _get_email_verification_code(self):
}

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:
LOGGER.debug("Verification code requested successfully.")
Expand All @@ -96,7 +110,7 @@ def _get_authentication_token_with_verification_code(self, code) -> dict:
"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}")
if response.status_code == 200:
Expand Down Expand Up @@ -128,7 +142,7 @@ def _get_authentication_token_with_2fa_code(self, code: str) -> dict:
"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}")
if response.status_code == 200:
Expand Down Expand Up @@ -209,38 +223,49 @@ def login(self, region: str, email: str, password: str) -> str:
self._password = password

result = self._get_authentication_token()
if result == 'verifyCode':
return result
elif result == 'tfa':
return result
elif result is None:
if result is None:
LOGGER.error("Unable to authenticate.")
return None
elif len(result) < 20:
return result
else:
self._auth_token = result
self._username = self._get_username_from_authentication_token()
return 'success'

def login_with_verification_code(self, code: str):
result = self._get_authentication_token_with_verification_code(code)
if result == 'codeExpired' or result == 'codeIncorrect':
if len(result) < 20:
return result
self._auth_token = result
self._username = self._get_username_from_authentication_token()
return 'success'

def login_with_2fa_code(self, code: str):
result = self._get_authentication_token_with_2fa_code(code)
if len(result) < 20:
return result
self._auth_token = result
self._username = self._get_username_from_authentication_token()
return 'success'

def get_device_list(self) -> dict:
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:
LOGGER.debug(f"Received error: {response.status_code}")
LOGGER.error(f"Received error: '{response.text}'")
raise ValueError(response.status_code)

return response.json()['devices']

# The slicer settings are of the following form:
Expand Down Expand Up @@ -311,13 +336,23 @@ def get_device_list(self) -> dict:
# }

def get_slicer_settings(self) -> dict:
LOGGER.debug("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)
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()
LOGGER.debug("DISABLED: Getting slicer settings from Bambu Cloud")
# Disabled for now since it may be contributing to cloudflare detection speed.
#
# if curl_available:
# response = curl_requests.get(get_Url(BambuUrl.SLICER_SETTINGS, 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(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.
#
Expand Down Expand Up @@ -362,24 +397,46 @@ def get_slicer_settings(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)
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:
LOGGER.debug(f"Received error: {response.status_code}")
LOGGER.debug(f"Received error: '{response.text}'")
raise ValueError(response.status_code)
raise None

return response.json()

def get_latest_task_for_printer(self, deviceId: str) -> dict:
LOGGER.debug(f"Getting latest task from Bambu Cloud")
data = self.get_tasklist_for_printer(deviceId)
if len(data) != 0:
return data[0]
LOGGER.debug("No tasks found for printer")
LOGGER.debug(f"Getting latest task for printer from Bambu Cloud")
try:
data = self.get_tasklist_for_printer(deviceId)
if len(data) != 0:
return data[0]
LOGGER.debug("No tasks found for printer")
except:
LOGGER.debug("Unable to make call")
return None

return None

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 = []
data = self.get_tasklist()
for task in data['hits']:
Expand All @@ -394,10 +451,21 @@ def get_device_type_from_device_product_name(self, device_product_name: str):

def download(self, url: str) -> bytearray:
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:
LOGGER.debug(f"Received error: {response.status_code}")
LOGGER.debug(f"Received error: {response.text}")
raise ValueError(response.status_code)

return response.content

@property
Expand All @@ -408,6 +476,10 @@ def username(self):
def auth_token(self):
return self._auth_token

@property
def bambu_connected(self) -> bool:
return self._auth_token != "" and self._auth_token != None

@property
def cloud_mqtt_host(self):
return "cn.mqtt.bambulab.com" if self._region == "China" else "us.mqtt.bambulab.com"
2 changes: 1 addition & 1 deletion setup.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
plugin_name = "OctoPrint-BambuPrinter"

# 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
# module
Expand Down

0 comments on commit 98ab94b

Please sign in to comment.