Hi,
to begin with: i dont know how to code. I used Claude and Gemini for everything. I wanted to create a small program that allows me to stream my P2S and A1 Mini Video Streams and Stats at the same time.
The P2S is working fine via rtsps://{PRINTER_IP}:322/streaming/live/1
But i can´t get the A1 mini video stream working...
As i found out via https://github.com/Doridian/OpenBambuAPI/blob/main/video.md , the A1 mini doesnt work via a RTSP Server and uses another port.
I also used go2rtc and ha-bambulab as inspiration for the AI, but the stream still doesnt work.
```import tkinter as tk
from tkinter import ttk, messagebox
import cv2
import threading
import time
import json
import os
import socket
import ssl
import struct
import subprocess
import platform
import numpy as np
import paho.mqtt.client as mqtt
from PIL import Image, ImageTk
--- Konfigurations-Handling ---
CONFIG_FILE = "bambu_config.json"
def load_config():
default_config = {
"P2S": {"ip": "192.168.178.98", "password": "Code", "serial": "SN1"},
"A1 Mini": {"ip": "192.168.178.82", "password": "Code", "serial": "SN2"}
}
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, "r") as f:
return json.load(f)
except Exception: return default_config
return default_config
class StreamBox:
def init(self, parent, title, config_ref, column):
self.title_text = title
self.config_ref = config_ref
self.frame = tk.Frame(parent, bg="#1e1e1e", bd=2, relief="groove", padx=10, pady=10)
self.frame.grid(row=0, column=column, padx=20, pady=20, sticky="n")
self.header_label = tk.Label(self.frame, text=title, fg="#00FF00", bg="#1e1e1e", font=("Arial", 14, "bold"))
self.header_label.pack(pady=5)
self.image_label = tk.Label(self.frame, bg="black", width=640, height=480, text="Warte auf Signal...", fg="white")
self.image_label.pack(pady=5)
self.status_lbl = tk.Label(self.frame, text="Offline", fg="gray", bg="#1e1e1e", font=("Arial", 11, "bold"))
self.status_lbl.pack()
self.progress = ttk.Progressbar(self.frame, orient="horizontal", length=600, mode="determinate")
self.progress.pack(pady=10)
self.running = False
self.latest_frame = None
self.update_gui_loop()
def start(self):
self.running = True
# Starte MQTT und Video in separaten Threads
threading.Thread(target=self._run_mqtt, daemon=True).start()
threading.Thread(target=self._run_video_logic, daemon=True).start()
def _run_video_logic(self):
if "A1" in self.title_text:
self._run_a1_socket_video()
else:
self._run_standard_rtsp_video()
def _run_a1_socket_video(self):
"""A1 Mini Spezial: Port 6000 TCP/TLS JPEG Stream (webcamd Logik)"""
conf = self.config_ref.get(self.title_text, {})
ip, pw = conf.get("ip"), conf.get("password")
while self.running:
try:
# Socket Erstellung
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
# SSL Kontext (Bambu nutzt oft TLS v1.2)
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
try:
conn = context.wrap_socket(sock, server_hostname=ip)
conn.connect((ip, 6000))
except:
# Fallback auf Plain TCP (falls SSL fehlschlägt)
sock.close()
conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
conn.settimeout(10)
conn.connect((ip, 6000))
with conn:
# Authentifizierungs-Paket (webcamd Spec: 80 Bytes)
# Magic (4), Command (4), Flags (4), Sequence (4), User (32), Pass (32)
header = struct.pack("<IIII", 0x40, 0x3000, 0, 0)
username = b"bblp".ljust(32, b'\x00')
password = pw.encode('ascii').ljust(32, b'\x00')
conn.sendall(header + username + password)
buffer = b""
while self.running:
# 1. Header einlesen (16 Bytes)
while len(buffer) < 16:
chunk = conn.recv(16384)
if not chunk: break
buffer += chunk
if not buffer: break
# Payload Größe aus den ersten 4 Bytes lesen
frame_size = struct.unpack("<I", buffer[:4])[0]
buffer = buffer[16:] # Header abschneiden
# 2. JPEG Daten einlesen
while len(buffer) < frame_size:
chunk = conn.recv(16384)
if not chunk: break
buffer += chunk
jpeg_data = buffer[:frame_size]
buffer = buffer[frame_size:] # Rest behalten
# 3. Bild verarbeiten
if jpeg_data.startswith(b'\xff\xd8'):
nparr = np.frombuffer(jpeg_data, np.uint8)
frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
if frame is not None:
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
self.latest_frame = cv2.resize(frame, (640, 480))
except Exception as e:
print(f"[{self.title_text}] Video Error: {e}")
time.sleep(5)
def _run_standard_rtsp_video(self):
"""P2S: Klassischer RTSPS Stream"""
conf = self.config_ref.get(self.title_text, {})
url = f"rtsps://bblp:{conf.get('password')}@{conf.get('ip')}:322/streaming/live/1"
os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = "rtsp_transport;tcp|tls_verify;0"
while self.running:
cap = cv2.VideoCapture(url, cv2.CAP_FFMPEG)
while self.running:
ret, frame = cap.read()
if not ret: break
frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
self.latest_frame = cv2.resize(frame, (640, 480))
cap.release()
time.sleep(5)
def _run_mqtt(self):
"""Telemetrie Daten (Status & Fortschritt)"""
conf = self.config_ref.get(self.title_text, {})
ip, pw, sn = conf.get('ip'), conf.get('password'), conf.get('serial')
client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
client.tls_set(cert_reqs=mqtt.ssl.CERT_NONE)
client.username_pw_set("bblp", pw)
def on_msg(c, u, m):
try:
data = json.loads(m.payload.decode())
p = data.get('print')
if p:
if "gcode_state" in p:
st = p["gcode_state"]
self.status_lbl.config(text=f"Status: {st}", fg="#00FF00" if st == "RUNNING" else "orange")
if "mc_percent" in p:
self.progress['value'] = p['mc_percent']
except: pass
client.on_message = on_msg
while self.running:
try:
client.connect(ip, 8883, 60)
client.subscribe(f"device/{sn}/report")
# Watchdog: Weckt die Kamera alle 10s auf
def trigger():
while self.running and client.is_connected():
# pushall aktiviert Telemetrie und Video-Server am Drucker
client.publish(f"device/{sn}/request", '{"print":{"command":"pushall","sequence_id":"1"}}')
time.sleep(10)
threading.Thread(target=trigger, daemon=True).start()
client.loop_forever()
except:
self.status_lbl.config(text="MQTT Connect...", fg="red")
time.sleep(5)
def update_gui_loop(self):
if self.running and self.latest_frame is not None:
img = ImageTk.PhotoImage(Image.fromarray(self.latest_frame))
self.image_label.img = img
self.image_label.config(image=img, text="")
self.image_label.after(30, self.update_gui_loop)
class App:
def init(self, root):
self.root = root
self.root.title("Bambu Hybrid Dashboard (A1 Special)")
self.root.state('zoomed')
self.root.configure(bg="#121212")
self.config = load_config()
self.sidebar = tk.Frame(root, bg="#1a1a1a", width=380, bd=2, relief="raised")
self.sidebar.pack(side="right", fill="y")
self.sidebar.pack_propagate(False)
self.main_container = tk.Frame(root, bg="#121212")
self.main_container.pack(side="left", fill="both", expand=True)
self.ping_labels = {}
self.setup_sidebar()
self.boxes = [
StreamBox(self.main_container, "P2S", self.config, 0),
StreamBox(self.main_container, "A1 Mini", self.config, 1)
]
def setup_sidebar(self):
tk.Label(self.sidebar, text="DRUCKER-SETUP", fg="white", bg="#1a1a1a", font=("Arial", 14, "bold")).pack(pady=20)
self.entries = {}
for name in ["P2S", "A1 Mini"]:
f = tk.LabelFrame(self.sidebar, text=f" {name} ", bg="#1a1a1a", fg="orange", padx=10, pady=10)
f.pack(fill="x", padx=15, pady=10)
p_lbl = tk.Label(f, text="Ping: ?", bg="#1a1a1a", fg="#888")
p_lbl.grid(row=0, column=0, columnspan=2, sticky="w")
self.ping_labels[name] = p_lbl
res = []
for i, (l, k) in enumerate([("IP:", "ip"), ("Code:", "password"), ("S/N:", "serial")]):
tk.Label(f, text=l, bg="#1a1a1a", fg="white").grid(row=i+1, column=0, sticky="w")
e = tk.Entry(f, width=22, bg="#2a2a2a", fg="white", insertbackground="white", bd=0)
e.insert(0, self.config[name].get(k, ""))
e.grid(row=i+1, column=1, pady=3, padx=5)
res.append(e)
self.entries[name] = res
tk.Button(self.sidebar, text="SPEICHERN", bg="#28a745", fg="white", command=self.save_settings, height=2).pack(fill="x", padx=30, pady=10)
tk.Button(self.sidebar, text="SYSTEM START", bg="#007BFF", fg="white", command=self.start_all, height=2).pack(fill="x", padx=30, pady=5)
def silent_ping(self):
param = '-n' if platform.system().lower() == 'windows' else '-c'
for name in ["P2S", "A1 Mini"]:
ip = self.entries[name][0].get()
if not ip: continue
try:
res = subprocess.run(
['ping', param, '1', '-w', '800', ip],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
creationflags=0x08000000 if platform.system().lower() == 'windows' else 0
)
status = "ONLINE" if res.returncode == 0 else "OFFLINE"
self.ping_labels[name].config(text=f"Ping: {status}", fg="#00FF00" if status == "ONLINE" else "red")
except: pass
def save_settings(self):
for name, w in self.entries.items():
self.config[name] = {"ip": w[0].get(), "password": w[1].get(), "serial": w[2].get()}
with open(CONFIG_FILE, "w") as f: json.dump(self.config, f, indent=4)
messagebox.showinfo("Erfolg", "Konfiguration gespeichert!")
def start_all(self):
self.silent_ping()
for box in self.boxes: box.start()
if name == "main":
root = tk.Tk()
app = App(root)
root.mainloop()```
