OctoPrint-BambuPrinter/octoprint_bambu_printer/printer/pybambu/bambu_cloud.py

from __future__ import annotations

import base64
import json
import httpx

from dataclasses import dataclass

from .const import LOGGER

@dataclass
class BambuCloud:

    def __init__(self, region: str, email: str, username: str, auth_token: str):
        self._region = region
        self._email = email
        self._username = username
        self._auth_token = auth_token

    def _get_authentication_token(self) -> dict:
        LOGGER.debug("Getting accessToken from Bambu Cloud")
        if self._region == "China":
            url = 'https://api.bambulab.cn/v1/user-service/user/login'
        else:
            url = 'https://api.bambulab.com/v1/user-service/user/login'
        headers = {'User-Agent' : "OctoPrint Plugin"}
        data = {'account': self._email, 'password': self._password}
        with httpx.Client(http2=True) as client:
            response = client.post(url, headers=headers, json=data, timeout=10)
        if response.status_code >= 400:
            LOGGER.debug(f"Received error: {response.status_code}")
            raise ValueError(response.status_code)
        return response.json()['accessToken']

    def _get_username_from_authentication_token(self) -> str:
        # User name is in 2nd portion of the auth token (delimited with periods)
        b64_string = self._auth_token.split(".")[1]
        # String must be multiples of 4 chars in length. For decode pad with = character
        b64_string += "=" * ((4 - len(b64_string) % 4) % 4)
        jsonAuthToken = json.loads(base64.b64decode(b64_string))
        # Gives json payload with "username":"u_<digits>" within it
        return jsonAuthToken['username']

    # Retrieves json description of devices in the form:
    # {
    #     'message': 'success',
    #     'code': None,
    #     'error': None,
    #     'devices': [
    #         {
    #             'dev_id': 'REDACTED',
    #             'name': 'Bambu P1S',
    #             'online': True,
    #             'print_status': 'SUCCESS',
    #             'dev_model_name': 'C12',
    #             'dev_product_name': 'P1S',
    #             'dev_access_code': 'REDACTED',
    #             'nozzle_diameter': 0.4
    #             },
    #         {
    #             'dev_id': 'REDACTED',
    #             'name': 'Bambu P1P',
    #             'online': True,
    #             'print_status': 'RUNNING',
    #             'dev_model_name': 'C11',
    #             'dev_product_name': 'P1P',
    #             'dev_access_code': 'REDACTED',
    #             'nozzle_diameter': 0.4
    #             },
    #         {
    #             'dev_id': 'REDACTED',
    #             'name': 'Bambu X1C',
    #             'online': True,
    #             'print_status': 'RUNNING',
    #             'dev_model_name': 'BL-P001',
    #             'dev_product_name': 'X1 Carbon',
    #             'dev_access_code': 'REDACTED',
    #             'nozzle_diameter': 0.4
    #             }
    #     ]
    # }

    def test_authentication(self, region: str, email: str, username: str, auth_token: str) -> bool:
        self._region = region
        self._email = email
        self._username = username
        self._auth_token = auth_token
        try:
            self.get_device_list()
        except:
            return False
        return True

    def login(self, region: str, email: str, password: str):
        self._region = region
        self._email = email
        self._password = password

        self._auth_token = self._get_authentication_token()
        self._username = self._get_username_from_authentication_token()

    def get_device_list(self) -> dict:
        LOGGER.debug("Getting device list from Bambu Cloud")
        if self._region == "China":
            url = 'https://api.bambulab.cn/v1/iot-service/api/user/bind'
        else:
            url = 'https://api.bambulab.com/v1/iot-service/api/user/bind'
        headers = {'Authorization': 'Bearer ' + self._auth_token, 'User-Agent' : "OctoPrint Plugin"}
        with httpx.Client(http2=True) as client:
            response = client.get(url, headers=headers, timeout=10)
        if response.status_code >= 400:
            LOGGER.debug(f"Received error: {response.status_code}")
            raise ValueError(response.status_code)
        return response.json()['devices']

    # The slicer settings are of the following form:
    #
    # {
    #     "message": "success",
    #     "code": null,
    #     "error": null,
    #     "print": {
    #         "public": [
    #             {
    #                 "setting_id": "GP004",
    #                 "version": "01.09.00.15",
    #                 "name": "0.20mm Standard @BBL X1C",
    #                 "update_time": "2024-07-04 11:27:08",
    #                 "nickname": null
    #             },
    #             ...
    #         }
    #         "private": []
    #     },
    #     "printer": {
    #         "public": [
    #             {
    #                 "setting_id": "GM001",
    #                 "version": "01.09.00.15",
    #                 "name": "Bambu Lab X1 Carbon 0.4 nozzle",
    #                 "update_time": "2024-07-04 11:25:07",
    #                 "nickname": null
    #             },
    #             ...
    #         ],
    #         "private": []
    #     },
    #     "filament": {
    #         "public": [
    #             {
    #                 "setting_id": "GFSA01",
    #                 "version": "01.09.00.15",
    #                 "name": "Bambu PLA Matte @BBL X1C",
    #                 "update_time": "2024-07-04 11:29:21",
    #                 "nickname": null,
    #                 "filament_id": "GFA01"
    #             },
    #             ...
    #         ],
    #         "private": [
    #             {
    #                 "setting_id": "PFUS46ea5c221cabe5",
    #                 "version": "1.9.0.14",
    #                 "name": "Fillamentum PLA Extrafill @Bambu Lab X1 Carbon 0.4 nozzle",
    #                 "update_time": "2024-07-10 06:48:17",
    #                 "base_id": null,
    #                 "filament_id": "Pc628b24",
    #                 "filament_type": "PLA",
    #                 "filament_is_support": "0",
    #                 "nozzle_temperature": [
    #                     190,
    #                     240
    #                 ],
    #                 "nozzle_hrc": "3",
    #                 "filament_vendor": "Fillamentum"
    #             },
    #             ...
    #         ]
    #     },
    #     "settings": {}
    # }

    def get_slicer_settings(self) -> dict:
        LOGGER.debug("Getting slicer settings from Bambu Cloud")
        if self._region == "China":
            url = 'https://api.bambulab.cn/v1/iot-service/api/slicer/setting?version=undefined'
        else:
            url = 'https://api.bambulab.com/v1/iot-service/api/slicer/setting?version=undefined'
        headers = {'Authorization': 'Bearer ' + self._auth_token, 'User-Agent' : "OctoPrint Plugin"}
        with httpx.Client(http2=True) as client:
            response = client.get(url, headers=headers, timeout=10)
        if response.status_code >= 400:
            LOGGER.error(f"Slicer settings load failed: {response.status_code}")
            return None
        return response.json()

    # The task list is of the following form with a 'hits' array with typical 20 entries.
    #
    # "total": 531,
    # "hits": [
    #     {
    #     "id": 35237965,
    #     "designId": 0,
    #     "designTitle": "",
    #     "instanceId": 0,
    #     "modelId": "REDACTED",
    #     "title": "REDACTED",
    #     "cover": "REDACTED",
    #     "status": 4,
    #     "feedbackStatus": 0,
    #     "startTime": "2023-12-21T19:02:16Z",
    #     "endTime": "2023-12-21T19:02:35Z",
    #     "weight": 34.62,
    #     "length": 1161,
    #     "costTime": 10346,
    #     "profileId": 35276233,
    #     "plateIndex": 1,
    #     "plateName": "",
    #     "deviceId": "REDACTED",
    #     "amsDetailMapping": [
    #         {
    #         "ams": 4,
    #         "sourceColor": "F4D976FF",
    #         "targetColor": "F4D976FF",
    #         "filamentId": "GFL99",
    #         "filamentType": "PLA",
    #         "targetFilamentType": "",
    #         "weight": 34.62
    #         }
    #     ],
    #     "mode": "cloud_file",
    #     "isPublicProfile": false,
    #     "isPrintable": true,
    #     "deviceModel": "P1P",
    #     "deviceName": "Bambu P1P",
    #     "bedType": "textured_plate"
    #     },

    def get_tasklist(self) -> dict:
        if self._region == "China":
            url = 'https://api.bambulab.cn/v1/user-service/my/tasks'
        else:
            url = 'https://api.bambulab.com/v1/user-service/my/tasks'
        headers = {'Authorization': 'Bearer ' + self._auth_token, 'User-Agent' : "OctoPrint Plugin"}
        with httpx.Client(http2=True) as client:
            response = client.get(url, headers=headers, timeout=10)
        if response.status_code >= 400:
            LOGGER.debug(f"Received error: {response.status_code}")
            raise ValueError(response.status_code)
        return response.json()

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

    def get_tasklist_for_printer(self, deviceId: str) -> dict:
        LOGGER.debug(f"Getting task list from Bambu Cloud for Printer: {deviceId}")
        tasks = []
        data = self.get_tasklist()
        for task in data['hits']:
            if task['deviceId'] == deviceId:
                tasks.append(task)
        return tasks

    def get_device_type_from_device_product_name(self, device_product_name: str):
        if device_product_name == "X1 Carbon":
            return "X1C"
        return device_product_name.replace(" ", "")

    def download(self, url: str) -> bytearray:
        LOGGER.debug(f"Downloading cover image: {url}")
        with httpx.Client(http2=True) as client:
            response = client.get(url, timeout=10)
        if response.status_code >= 400:
            LOGGER.debug(f"Received error: {response.status_code}")
            raise ValueError(response.status_code)
        return response.content

    @property
    def username(self):
        return self._username

    @property
    def auth_token(self):
        return self._auth_token

    @property
    def cloud_mqtt_host(self):
        return "cn.mqtt.bambulab.com" if self._region == "China" else "us.mqtt.bambulab.com"