diff --git a/bambu_run/admin.py b/bambu_run/admin.py index 03d7989..4873f13 100644 --- a/bambu_run/admin.py +++ b/bambu_run/admin.py @@ -1,5 +1,5 @@ from django.contrib import admin -from .models import Printer, PrinterMetrics, Filament, FilamentType, FilamentSnapshot, PrintJob, FilamentUsage, BambuCloudTask +from .models import Printer, PrinterMetrics, Filament, FilamentType, FilamentSnapshot, PrintJob, FilamentUsage, BambuCloudTask, Hotend, HotendSnapshot @admin.register(Printer) @@ -107,6 +107,21 @@ class FilamentUsageAdmin(admin.ModelAdmin): readonly_fields = ('consumed_percent', 'consumed_grams') +@admin.register(Hotend) +class HotendAdmin(admin.ModelAdmin): + list_display = ('printer', 'serial_number', 'nozzle_type', 'is_toolhead', 'slot_number', 'used_time_seconds', 'wear_percent', 'last_seen_at') + list_filter = ('printer', 'is_toolhead', 'nozzle_type') + search_fields = ('serial_number',) + readonly_fields = ('last_seen_at', 'created_at') + + +@admin.register(HotendSnapshot) +class HotendSnapshotAdmin(admin.ModelAdmin): + list_display = ('printer_metric', 'hotend', 'raw_id', 'used_time_seconds', 'wear_percent', 'timestamp') + list_filter = ('hotend',) + readonly_fields = ('printer_metric', 'hotend', 'raw_id', 'used_time_seconds', 'wear_percent', 'stat', 'timestamp') + + @admin.register(BambuCloudTask) class BambuCloudTaskAdmin(admin.ModelAdmin): list_display = ('task_id', 'design_title', 'plate_title', 'device_serial', 'cloud_status', 'weight_grams', 'cloud_start_time', 'synced_at') diff --git a/bambu_run/management/commands/bambu_collector.py b/bambu_run/management/commands/bambu_collector.py index bdec54a..ce1bcdd 100644 --- a/bambu_run/management/commands/bambu_collector.py +++ b/bambu_run/management/commands/bambu_collector.py @@ -571,6 +571,38 @@ class Command(BaseCommand): match_method=match_method ) + def _update_hotends(self, printer, printer_metric, hotends_data): + from bambu_run.models import Hotend, HotendSnapshot + + for h in hotends_data: + if h.get("is_empty"): + continue + + hotend, _ = Hotend.objects.update_or_create( + printer=printer, + serial_number=h.get("serial_number"), + defaults={ + "raw_id": h.get("raw_id", 0), + "nozzle_type": h.get("nozzle_type", ""), + "diameter": self._to_decimal(h.get("diameter")), + "slot_number": h.get("slot_number"), + "is_toolhead": bool(h.get("is_toolhead")), + "last_filament_profile_id": h.get("fila_id", ""), + "last_color": h.get("color") or "", + "used_time_seconds": h.get("used_time_seconds", 0), + "wear_percent": h.get("wear_percent", 0), + }, + ) + + HotendSnapshot.objects.create( + printer_metric=printer_metric, + hotend=hotend, + raw_id=h.get("raw_id", 0), + used_time_seconds=h.get("used_time_seconds", 0), + wear_percent=h.get("wear_percent", 0), + stat=h.get("stat"), + ) + def _track_print_job(self, session, metric, snapshot): from bambu_run.models import PrintJob @@ -757,12 +789,17 @@ class Command(BaseCommand): external_spool=snapshot.get("external_spool", {}), lights_report=snapshot.get("lights_report", []), vortek_raw=snapshot.get("vortek_raw", {}), + nozzle_info=snapshot.get("hotends", []), ) filaments_data = snapshot.get('filaments', []) if filaments_data: self._create_filament_snapshots(metric, filaments_data, snapshot) + hotends_data = snapshot.get('hotends', []) + if hotends_data: + self._update_hotends(session.printer, metric, hotends_data) + self._track_print_job(session, metric, snapshot) session.success_count += 1 diff --git a/bambu_run/migrations/0007_hotend_hotendsnapshot.py b/bambu_run/migrations/0007_hotend_hotendsnapshot.py new file mode 100644 index 0000000..adf7a75 --- /dev/null +++ b/bambu_run/migrations/0007_hotend_hotendsnapshot.py @@ -0,0 +1,172 @@ +# Generated by Django 5.2.8 on 2026-06-20 14:07 + +import django.db.models.deletion +import django.utils.timezone +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("bambu_run", "0006_alter_filamentsnapshot_options_and_more"), + ] + + operations = [ + migrations.CreateModel( + name="Hotend", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("serial_number", models.CharField(db_index=True, max_length=100)), + ( + "nozzle_type", + models.CharField(blank=True, default="", max_length=50), + ), + ( + "diameter", + models.DecimalField( + blank=True, decimal_places=2, max_digits=3, null=True + ), + ), + ( + "raw_id", + models.PositiveSmallIntegerField( + help_text="Last-seen MQTT device.nozzle.info[].id" + ), + ), + ( + "slot_number", + models.PositiveSmallIntegerField( + blank=True, + help_text="Rack bay 1-6, derived from raw_id 16-21. Null if currently unknown (e.g. mounted on toolhead and id reports as the 0 sentinel).", + null=True, + ), + ), + ( + "is_toolhead", + models.BooleanField( + default=False, + help_text="True if currently mounted on the toolhead under normal polling (raw_id == 0).", + ), + ), + ( + "last_filament_profile_id", + models.CharField( + blank=True, + default="", + help_text="Bambu material profile id of the filament last loaded (MQTT fila_id, e.g. 'GFA01')", + max_length=20, + ), + ), + ( + "last_color", + models.CharField( + blank=True, + default="", + help_text="6-char hex of the filament last loaded (MQTT color_m, alpha stripped)", + max_length=6, + ), + ), + ("used_time_seconds", models.PositiveIntegerField(default=0)), + ( + "wear_percent", + models.DecimalField( + decimal_places=2, + default=0, + help_text="MQTT wear (0-128 scale) converted to a 0-100 percent", + max_digits=5, + ), + ), + ("last_seen_at", models.DateTimeField(auto_now=True)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ( + "printer", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="hotends", + to="bambu_run.printer", + ), + ), + ], + options={ + "verbose_name": "Hotend", + "verbose_name_plural": "Hotends", + "db_table": "infrastructure_hotend", + "ordering": ["printer", "-is_toolhead", "slot_number", "serial_number"], + "unique_together": {("printer", "serial_number")}, + }, + ), + migrations.CreateModel( + name="HotendSnapshot", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ("raw_id", models.PositiveSmallIntegerField()), + ("used_time_seconds", models.PositiveIntegerField(default=0)), + ( + "wear_percent", + models.DecimalField(decimal_places=2, default=0, max_digits=5), + ), + ( + "stat", + models.IntegerField( + blank=True, + help_text="Raw MQTT status code for this hotend", + null=True, + ), + ), + ( + "timestamp", + models.DateTimeField( + db_index=True, default=django.utils.timezone.now + ), + ), + ( + "hotend", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="snapshots", + to="bambu_run.hotend", + ), + ), + ( + "printer_metric", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="hotend_snapshots", + to="bambu_run.printermetrics", + ), + ), + ], + options={ + "verbose_name": "Hotend Snapshot", + "verbose_name_plural": "Hotend Snapshots", + "db_table": "infrastructure_hotend_snapshot", + "ordering": ["printer_metric", "hotend"], + "indexes": [ + models.Index( + fields=["printer_metric", "hotend"], + name="infrastruct_printer_b528aa_idx", + ), + models.Index( + fields=["hotend", "-timestamp"], + name="infrastruct_hotend__691f7e_idx", + ), + ], + }, + ), + ] diff --git a/bambu_run/migrations/0008_printermetrics_nozzle_info.py b/bambu_run/migrations/0008_printermetrics_nozzle_info.py new file mode 100644 index 0000000..4880437 --- /dev/null +++ b/bambu_run/migrations/0008_printermetrics_nozzle_info.py @@ -0,0 +1,22 @@ +# Generated by Django 5.2.8 on 2026-06-20 14:22 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("bambu_run", "0007_hotend_hotendsnapshot"), + ] + + operations = [ + migrations.AddField( + model_name="printermetrics", + name="nozzle_info", + field=models.JSONField( + blank=True, + default=list, + help_text="Parsed per-poll nozzle/hotend info list", + ), + ), + ] diff --git a/bambu_run/models.py b/bambu_run/models.py index fb27df4..85dbff3 100644 --- a/bambu_run/models.py +++ b/bambu_run/models.py @@ -242,6 +242,15 @@ class PrinterMetrics(models.Model): default=dict, blank=True, help_text="Raw print.device MQTT payload (Vortek rack groundwork)" ) + # Parsed device.nozzle.info[] from this poll, one dict per entry (mirrors + # HotendInfo.to_dict()). Includes induction-chip hotends *and* non-inductive + # nozzle positions (e.g. H2C's fixed left nozzle) that have no stable serial + # number to key a Hotend registry row on — kept here so the dashboard can show + # their readable type/diameter without claiming an identity/history we don't have. + nozzle_info = models.JSONField( + default=list, blank=True, help_text="Parsed per-poll nozzle/hotend info list" + ) + class Meta: db_table = "infrastructure_printer_metrics" verbose_name = "Printer Metric" @@ -740,3 +749,108 @@ class FilamentUsage(models.Model): self.consumed_grams = int( self.filament.initial_weight_grams * (self.consumed_percent / 100.0) ) + + +class Hotend(models.Model): + """Registry of individual Vortek hotends, keyed by serial number. + + A Vortek rack holds up to 6 swappable hotends (bays, MQTT `id` 16-21) plus + 1 mounted on the toolhead at a time (MQTT `id` 0). `raw_id` reflects whichever + address was last seen on the wire for this hotend; `slot_number` is only set + when that address falls in the 16-21 rack-bay range — confirmed by watching + a "Read All" MQTT capture reassign a toolhead-mounted hotend's id from 0 to + its true bay id. + """ + + printer = models.ForeignKey( + 'Printer', on_delete=models.CASCADE, related_name='hotends' + ) + serial_number = models.CharField(max_length=100, db_index=True) + + nozzle_type = models.CharField(max_length=50, blank=True, default="") + diameter = models.DecimalField( + max_digits=3, decimal_places=2, null=True, blank=True + ) + + raw_id = models.PositiveSmallIntegerField( + help_text="Last-seen MQTT device.nozzle.info[].id" + ) + slot_number = models.PositiveSmallIntegerField( + null=True, blank=True, + help_text="Rack bay 1-6, derived from raw_id 16-21. Null if currently unknown (e.g. mounted on toolhead and id reports as the 0 sentinel)." + ) + is_toolhead = models.BooleanField( + default=False, + help_text="True if currently mounted on the toolhead under normal polling (raw_id == 0)." + ) + + last_filament_profile_id = models.CharField( + max_length=20, blank=True, default="", + help_text="Bambu material profile id of the filament last loaded (MQTT fila_id, e.g. 'GFA01')" + ) + last_color = models.CharField( + max_length=6, blank=True, default="", + help_text="6-char hex of the filament last loaded (MQTT color_m, alpha stripped)" + ) + + used_time_seconds = models.PositiveIntegerField(default=0) + wear_percent = models.DecimalField( + max_digits=5, decimal_places=2, default=0, + help_text="MQTT wear (0-128 scale) converted to a 0-100 percent" + ) + + last_seen_at = models.DateTimeField(auto_now=True) + created_at = models.DateTimeField(auto_now_add=True) + + class Meta: + db_table = "infrastructure_hotend" + verbose_name = "Hotend" + verbose_name_plural = "Hotends" + ordering = ['printer', '-is_toolhead', 'slot_number', 'serial_number'] + unique_together = [['printer', 'serial_number']] + + def __str__(self): + location = "Toolhead" if self.is_toolhead else ( + f"Slot {self.slot_number}" if self.slot_number else "Rack" + ) + return f"{self.serial_number} ({location})" + + @property + def used_time_display(self) -> str: + hours, remainder = divmod(self.used_time_seconds, 3600) + minutes = remainder // 60 + return f"{hours}h {minutes}m" if hours else f"{minutes}m" + + +class HotendSnapshot(models.Model): + """Point-in-time reading of a Hotend, one row per collector poll.""" + + printer_metric = models.ForeignKey( + 'PrinterMetrics', on_delete=models.CASCADE, + related_name='hotend_snapshots' + ) + hotend = models.ForeignKey( + 'Hotend', on_delete=models.CASCADE, + related_name='snapshots' + ) + + raw_id = models.PositiveSmallIntegerField() + used_time_seconds = models.PositiveIntegerField(default=0) + wear_percent = models.DecimalField(max_digits=5, decimal_places=2, default=0) + stat = models.IntegerField( + null=True, blank=True, help_text="Raw MQTT status code for this hotend" + ) + timestamp = models.DateTimeField(default=timezone.now, db_index=True) + + class Meta: + db_table = "infrastructure_hotend_snapshot" + verbose_name = "Hotend Snapshot" + verbose_name_plural = "Hotend Snapshots" + ordering = ['printer_metric', 'hotend'] + indexes = [ + models.Index(fields=['printer_metric', 'hotend']), + models.Index(fields=['hotend', '-timestamp']), + ] + + def __str__(self): + return f"{self.hotend.serial_number} @ {self.timestamp.strftime('%Y-%m-%d %H:%M:%S')}" diff --git a/bambu_run/mqtt_client.py b/bambu_run/mqtt_client.py index 0ddc815..e8474d3 100644 --- a/bambu_run/mqtt_client.py +++ b/bambu_run/mqtt_client.py @@ -296,6 +296,73 @@ class AMSState: return loaded +@dataclass +class HotendInfo: + """A single hotend reported in `device.nozzle.info[]` (Vortek rack). + + `raw_id` semantics (confirmed by watching a live "Read All" MQTT capture): + 0 = currently mounted on the (swappable) toolhead — the sentinel hides the + true bay address until "Read All" resolves it; 1 = the fixed left nozzle on + dual-nozzle printers (no RFID chip, always reports sn="N/A"); 16-21 = rack + bay address, slot_number = raw_id - 15 (1-6). + """ + raw_id: int = 0 + serial_number: str = "" + nozzle_type: str = "" + diameter: float = 0.4 + fila_id: str = "" + color: Optional[str] = None + used_time_seconds: int = 0 + wear_percent: float = 0.0 + stat: int = 0 + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "HotendInfo": + from .utils import strip_color_padding + + return cls( + raw_id=int(data.get("id", 0)), + serial_number=data.get("sn", ""), + nozzle_type=data.get("type", ""), + diameter=float(data.get("diameter", 0.4)), + fila_id=data.get("fila_id", ""), + color=strip_color_padding(data.get("color_m")), + used_time_seconds=int(data.get("p_t", 0)), + wear_percent=round(float(data.get("wear", 0.0)) / 128.0 * 100, 2), + stat=int(data.get("stat", 0)), + ) + + @property + def is_toolhead(self) -> bool: + return self.raw_id == 0 + + @property + def is_empty(self) -> bool: + return self.serial_number in ("", "N/A") + + @property + def slot_number(self) -> Optional[int]: + if 16 <= self.raw_id <= 21: + return self.raw_id - 15 + return None + + def to_dict(self) -> Dict[str, Any]: + return { + "raw_id": self.raw_id, + "serial_number": self.serial_number, + "nozzle_type": self.nozzle_type, + "diameter": self.diameter, + "fila_id": self.fila_id, + "color": self.color, + "used_time_seconds": self.used_time_seconds, + "wear_percent": self.wear_percent, + "stat": self.stat, + "is_toolhead": self.is_toolhead, + "is_empty": self.is_empty, + "slot_number": self.slot_number, + } + + @dataclass class PrinterState: """Complete printer state parsed from MQTT data""" @@ -388,6 +455,9 @@ class PrinterState: # External spool (virtual tray) vt_tray: Optional[Dict[str, Any]] = None + # Vortek hotend rack (device.nozzle.info[]) + hotends: List[HotendInfo] = field(default_factory=list) + # Raw data for any additional fields _raw_data: Dict[str, Any] = field(default_factory=dict, repr=False) @@ -431,6 +501,12 @@ class PrinterState: nozzle_target_temp_left = float((t >> 16) & 0xFFFF) nozzle_temp_left = float(t & 0xFFFF) + # Vortek hotend rack: device.nozzle.info[] — one entry per hotend. + hotends = [ + HotendInfo.from_dict(h) + for h in (device.get("nozzle") or {}).get("info") or [] + ] + return cls( timestamp=timestamp, sequence_id=str(print_data.get("sequence_id", "")), @@ -487,6 +563,7 @@ class PrinterState: gcode_file_prepare_percent=str(print_data.get("gcode_file_prepare_percent", "")), lifecycle=print_data.get("lifecycle", ""), vt_tray=print_data.get("vt_tray"), + hotends=hotends, _raw_data=data, ) @@ -533,6 +610,7 @@ class PrinterState: # hotends + 1 fixed left nozzle) isn't fully modeled yet — stash everything # here so no data is lost once the real Vortek MQTT schema is confirmed. "vortek_raw": self._raw_data.get("print", {}).get("device", {}), + "hotends": [h.to_dict() for h in self.hotends], "hms": self.hms, "stg_cur": self.stg_cur, "lights_report": self.lights_report, diff --git a/bambu_run/templates/bambu_run/printer_dashboard.html b/bambu_run/templates/bambu_run/printer_dashboard.html index 0040219..a171f74 100644 --- a/bambu_run/templates/bambu_run/printer_dashboard.html +++ b/bambu_run/templates/bambu_run/printer_dashboard.html @@ -282,6 +282,64 @@ + + {% if stats.hotends or stats.nozzle_positions %} +
+
+
+
+
Hotends
+
+
+
+ {% for hotend in stats.hotends %} +
+
+
+
+
+ {% if hotend.is_toolhead %}Toolhead{% elif hotend.slot_number %}Slot {{ hotend.slot_number }}{% else %}Rack{% endif %} +
+ {% if hotend.is_toolhead %}Toolhead{% endif %} +
+

SN {{ hotend.serial_number }}

+

{{ hotend.nozzle_type }}{% if hotend.diameter %} · {{ hotend.diameter }}mm{% endif %}

+ {% if hotend.last_filament_profile_id %}

Last: {{ hotend.last_filament_profile_id }}

{% endif %} +
+ Used time + {{ hotend.used_time_display }} +
+
+ Wear + {{ hotend.wear_percent|floatformat:0 }}% +
+
+
+
+
+
+
+ {% endfor %} + {% for nozzle in stats.nozzle_positions %} +
+
+
+
+
{% if nozzle.is_toolhead %}Toolhead{% else %}Fixed Nozzle{% endif %}
+
+

{{ nozzle.nozzle_type }}{% if nozzle.diameter %} · {{ nozzle.diameter }}mm{% endif %}

+

No induction chip data

+
+
+
+ {% endfor %} +
+
+
+
+
+ {% endif %} + {% if not is_basic_user %}
diff --git a/bambu_run/views.py b/bambu_run/views.py index 15c5c0d..414d5ba 100644 --- a/bambu_run/views.py +++ b/bambu_run/views.py @@ -11,7 +11,7 @@ import json import zoneinfo from .conf import app_settings -from .models import Printer, PrinterMetrics, Filament, FilamentColor, FilamentType, FilamentSnapshot, PrintJob, FilamentUsage +from .models import Printer, PrinterMetrics, Filament, FilamentColor, FilamentType, FilamentSnapshot, PrintJob, FilamentUsage, Hotend from .forms import FilamentForm, FilamentColorForm, FilamentTypeForm _METRICS_API_FIELDS = [ @@ -231,6 +231,19 @@ class PrinterDashboardView(LoginRequiredMixin, TemplateView): "filaments": filaments_list, "ams_units": ams_units_list, "ams_groups": ams_groups, + "hotends": list( + Hotend.objects.filter(printer=printer_device) + .order_by('-is_toolhead', 'slot_number', 'serial_number') + ), + # Nozzle positions with no induction chip (no stable serial number to + # key a Hotend registry row on, e.g. H2C's fixed left nozzle) — shown + # read-only from the latest poll, not persisted/historical. Entries with + # no readable type/diameter at all (i.e. genuinely nothing there) are + # dropped rather than shown as an empty placeholder. + "nozzle_positions": [ + h for h in (latest_metric.nozzle_info or []) + if h.get('is_empty') and (h.get('nozzle_type') or h.get('diameter')) + ], "external_spool": latest_metric.external_spool or {}, "timestamp": latest_metric.timestamp.astimezone(tz).strftime("%Y-%m-%d %H:%M:%S"), } diff --git a/tests/test_hotend_collection.py b/tests/test_hotend_collection.py new file mode 100644 index 0000000..28ffda5 --- /dev/null +++ b/tests/test_hotend_collection.py @@ -0,0 +1,121 @@ +import pytest + +from bambu_run.management.commands.bambu_collector import Command, DeviceSession, resolve_printer_device +from bambu_run.models import Hotend, HotendSnapshot, PrinterMetrics + + +class FakeClient: + """Stub in place of BambuPrinter — returns canned snapshots, no real MQTT.""" + + def __init__(self, snapshots): + self._snapshots = snapshots + self._index = 0 + self._client = None + + def get_snapshot(self): + snap = self._snapshots[min(self._index, len(self._snapshots) - 1)] + self._index += 1 + return snap + + +def make_session(device_id, name, snapshots): + printer = resolve_printer_device(device_id, {"name": name, "dev_product_name": "H2C"}) + return DeviceSession(device_id=device_id, client=FakeClient(snapshots), printer=printer) + + +def hotends_snapshot(used_time=11472, wear=100.0): + return { + "gcode_state": "IDLE", + "hotends": [ + { + "raw_id": 21, "serial_number": "20D06A5B2918952", "nozzle_type": "HS01", + "diameter": 0.4, "fila_id": "GFA01", "color": "FFFFFF", + "used_time_seconds": used_time, "wear_percent": wear, "stat": 0, + "is_toolhead": False, "is_empty": False, "slot_number": 6, + }, + { + "raw_id": 1, "serial_number": "N/A", "nozzle_type": "HS01", + "diameter": 0.4, "fila_id": "", "color": None, + "used_time_seconds": 0, "wear_percent": 0.0, "stat": 0, + "is_toolhead": False, "is_empty": True, "slot_number": None, + }, + { + "raw_id": 0, "serial_number": "20D06A5C0426280", "nozzle_type": "HS01", + "diameter": 0.4, "fila_id": "GFA00", "color": "FEC600", + "used_time_seconds": 93490, "wear_percent": 100.0, "stat": 0, + "is_toolhead": True, "is_empty": False, "slot_number": None, + }, + ], + } + + +@pytest.mark.django_db +def test_first_poll_creates_one_hotend_per_non_empty_entry(): + session = make_session("SERIAL-A", "Printer A", [hotends_snapshot()]) + + cmd = Command() + cmd.verbose = False + cmd._collect_printer_data(session) + + hotends = Hotend.objects.filter(printer=session.printer) + assert hotends.count() == 2 # empty bay (sn="N/A") skipped + + rack = hotends.get(serial_number="20D06A5B2918952") + assert rack.raw_id == 21 + assert rack.slot_number == 6 + assert rack.is_toolhead is False + assert rack.used_time_seconds == 11472 + assert rack.wear_percent == 100.0 + assert rack.nozzle_type == "HS01" + assert rack.last_filament_profile_id == "GFA01" + assert rack.last_color == "FFFFFF" + + toolhead = hotends.get(serial_number="20D06A5C0426280") + assert toolhead.is_toolhead is True + assert toolhead.slot_number is None + + +@pytest.mark.django_db +def test_first_poll_creates_one_snapshot_per_non_empty_hotend(): + session = make_session("SERIAL-A", "Printer A", [hotends_snapshot()]) + + cmd = Command() + cmd.verbose = False + cmd._collect_printer_data(session) + + metric = PrinterMetrics.objects.get(device=session.printer) + assert HotendSnapshot.objects.filter(printer_metric=metric).count() == 2 + + +@pytest.mark.django_db +def test_collector_persists_raw_nozzle_info_including_non_inductive_entries(): + session = make_session("SERIAL-A", "Printer A", [hotends_snapshot()]) + + cmd = Command() + cmd.verbose = False + cmd._collect_printer_data(session) + + metric = PrinterMetrics.objects.get(device=session.printer) + assert len(metric.nozzle_info) == 3 # all entries, including the empty/non-inductive one + serials = {h["serial_number"] for h in metric.nozzle_info} + assert serials == {"20D06A5B2918952", "N/A", "20D06A5C0426280"} + + +@pytest.mark.django_db +def test_second_poll_updates_existing_hotend_instead_of_duplicating(): + session = make_session( + "SERIAL-A", "Printer A", + [hotends_snapshot(used_time=11472, wear=100.0), hotends_snapshot(used_time=11500, wear=100.0)], + ) + + cmd = Command() + cmd.verbose = False + cmd._collect_printer_data(session) + cmd._collect_printer_data(session) + + hotends = Hotend.objects.filter(printer=session.printer, serial_number="20D06A5B2918952") + assert hotends.count() == 1 + assert hotends.first().used_time_seconds == 11500 + + snapshots = HotendSnapshot.objects.filter(hotend=hotends.first()) + assert snapshots.count() == 2 diff --git a/tests/test_hotend_dashboard.py b/tests/test_hotend_dashboard.py new file mode 100644 index 0000000..3043721 --- /dev/null +++ b/tests/test_hotend_dashboard.py @@ -0,0 +1,128 @@ +import pytest +from django.urls import reverse +from django.utils import timezone + +from bambu_run.models import Printer, PrinterMetrics, Hotend + + +@pytest.fixture +def logged_in_client(client, django_user_model): + user = django_user_model.objects.create_user(username="tester", password="pw") + client.force_login(user) + return client + + +@pytest.mark.django_db +def test_dashboard_context_includes_hotends_toolhead_first(logged_in_client): + printer = Printer.objects.create(name="Printer A", model="H2C", is_active=True) + PrinterMetrics.objects.create(device=printer, timestamp=timezone.now()) + + Hotend.objects.create( + printer=printer, serial_number="RACK-SN", raw_id=16, slot_number=1, + is_toolhead=False, nozzle_type="HS01", used_time_seconds=3600, wear_percent=50, + ) + Hotend.objects.create( + printer=printer, serial_number="TOOLHEAD-SN", raw_id=0, slot_number=None, + is_toolhead=True, nozzle_type="HS01", used_time_seconds=7200, wear_percent=80, + ) + + resp = logged_in_client.get( + reverse("bambu_run:printer_dashboard", kwargs={"pk": printer.pk}) + ) + + hotends = resp.context["stats"]["hotends"] + assert len(hotends) == 2 + assert hotends[0].serial_number == "TOOLHEAD-SN" + assert hotends[1].serial_number == "RACK-SN" + + +@pytest.mark.django_db +def test_dashboard_context_includes_non_inductive_nozzle_positions(logged_in_client): + printer = Printer.objects.create(name="Printer A", model="H2C", is_active=True) + PrinterMetrics.objects.create( + device=printer, timestamp=timezone.now(), + nozzle_info=[ + { + "raw_id": 1, "serial_number": "N/A", "nozzle_type": "HS01", "diameter": 0.4, + "fila_id": "", "color": None, "used_time_seconds": 0, "wear_percent": 0.0, + "stat": 0, "is_toolhead": False, "is_empty": True, "slot_number": None, + }, + ], + ) + + resp = logged_in_client.get( + reverse("bambu_run:printer_dashboard", kwargs={"pk": printer.pk}) + ) + + positions = resp.context["stats"]["nozzle_positions"] + assert len(positions) == 1 + assert positions[0]["nozzle_type"] == "HS01" + + +@pytest.mark.django_db +def test_dashboard_omits_nozzle_positions_with_no_readable_data(logged_in_client): + printer = Printer.objects.create(name="Printer A", model="H2C", is_active=True) + PrinterMetrics.objects.create( + device=printer, timestamp=timezone.now(), + nozzle_info=[ + { + "raw_id": 1, "serial_number": "N/A", "nozzle_type": "", "diameter": 0, + "fila_id": "", "color": None, "used_time_seconds": 0, "wear_percent": 0.0, + "stat": 0, "is_toolhead": False, "is_empty": True, "slot_number": None, + }, + ], + ) + + resp = logged_in_client.get( + reverse("bambu_run:printer_dashboard", kwargs={"pk": printer.pk}) + ) + + assert resp.context["stats"]["nozzle_positions"] == [] + assert "
Hotends
" not in resp.content.decode() + + +@pytest.mark.django_db +def test_dashboard_renders_nozzle_position_without_serial_or_wear(logged_in_client): + printer = Printer.objects.create(name="Printer A", model="H2C", is_active=True) + PrinterMetrics.objects.create( + device=printer, timestamp=timezone.now(), + nozzle_info=[ + { + "raw_id": 1, "serial_number": "N/A", "nozzle_type": "HS01", "diameter": 0.4, + "fila_id": "", "color": None, "used_time_seconds": 0, "wear_percent": 0.0, + "stat": 0, "is_toolhead": False, "is_empty": True, "slot_number": None, + }, + ], + ) + + resp = logged_in_client.get( + reverse("bambu_run:printer_dashboard", kwargs={"pk": printer.pk}) + ) + + html = resp.content.decode() + assert "Hotends" in html + assert "HS01" in html + assert "SN: N/A" not in html + assert "SN N/A" not in html + + +@pytest.mark.django_db +def test_dashboard_renders_hotends_card(logged_in_client): + printer = Printer.objects.create(name="Printer A", model="H2C", is_active=True) + PrinterMetrics.objects.create(device=printer, timestamp=timezone.now()) + + Hotend.objects.create( + printer=printer, serial_number="RACK-SN", raw_id=18, slot_number=3, + is_toolhead=False, nozzle_type="HS01", diameter=0.4, + used_time_seconds=3661, wear_percent=50, last_filament_profile_id="GFA01", + last_color="DE4343", + ) + + resp = logged_in_client.get( + reverse("bambu_run:printer_dashboard", kwargs={"pk": printer.pk}) + ) + + html = resp.content.decode() + assert "Hotends" in html + assert "RACK-SN" in html + assert "Slot 3" in html diff --git a/tests/test_hotend_parsing.py b/tests/test_hotend_parsing.py new file mode 100644 index 0000000..e837385 --- /dev/null +++ b/tests/test_hotend_parsing.py @@ -0,0 +1,100 @@ +from bambu_run.mqtt_client import PrinterState + + +def real_nozzle_payload(): + """Real captured device.nozzle payload from a live H2C with a Vortek rack + (1x AMS, 1x AMS 2 Pro, 1x AMS HT physically connected — unrelated here). + SN/used-time cross-checked against the user's Bambu Studio Hotends Info table.""" + return { + "exist": 3997699, + "src_id": 17, + "tar_id": 17, + "state": 0, + "info": [ + {"id": 21, "sn": "20D06A5B2918952", "type": "HS01", "diameter": 0.4, + "fila_id": "GFA01", "color_m": "FFFFFFFF", "p_t": 11472, "wear": 128.0, "stat": 0, "tm": 350}, + {"id": 1, "sn": "N/A", "type": "HS01", "diameter": 0.4, + "fila_id": "", "color_m": "00000000", "p_t": 0, "wear": 0.0, "stat": 0, "tm": 0}, + {"id": 16, "sn": "20D06A5B2919219", "type": "HS01", "diameter": 0.4, + "fila_id": "GFA01", "color_m": "A3D8E1FF", "p_t": 105386, "wear": 128.0, "stat": 0, "tm": 350}, + {"id": 20, "sn": "20D06A590610257", "type": "HS01", "diameter": 0.4, + "fila_id": "GFG01", "color_m": "00000000", "p_t": 81506, "wear": 128.0, "stat": 0, "tm": 350}, + {"id": 18, "sn": "20D06A591506263", "type": "HS01", "diameter": 0.4, + "fila_id": "GFA01", "color_m": "DE4343FF", "p_t": 30962, "wear": 128.0, "stat": 0, "tm": 350}, + {"id": 0, "sn": "20D06A5C0426280", "type": "HS01", "diameter": 0.4, + "fila_id": "GFA00", "color_m": "FEC600FF", "p_t": 93490, "wear": 128.0, "stat": 0, "tm": 350}, + {"id": 19, "sn": "20D06A5C0207881", "type": "HS01", "diameter": 0.4, + "fila_id": "GFA01", "color_m": "DE4343FF", "p_t": 1430, "wear": 128.0, "stat": 0, "tm": 350}, + ], + } + + +def make_data(nozzle_payload): + return {"print": {"gcode_state": "IDLE", "device": {"nozzle": nozzle_payload}}} + + +def test_snapshot_includes_one_hotend_per_nozzle_info_entry(): + state = PrinterState.from_mqtt_data(make_data(real_nozzle_payload())) + snapshot = state.get_snapshot() + + assert len(snapshot["hotends"]) == 7 + + +def test_hotend_fields_extracted_correctly(): + state = PrinterState.from_mqtt_data(make_data(real_nozzle_payload())) + snapshot = state.get_snapshot() + + by_sn = {h["serial_number"]: h for h in snapshot["hotends"]} + h = by_sn["20D06A5B2919219"] + + assert h["raw_id"] == 16 + assert h["nozzle_type"] == "HS01" + assert h["diameter"] == 0.4 + assert h["fila_id"] == "GFA01" + assert h["color"] == "A3D8E1" # alpha stripped + assert h["used_time_seconds"] == 105386 + assert h["wear_percent"] == 100.0 # 128/128*100 + assert h["is_empty"] is False + + +def test_id_zero_is_toolhead_and_resolves_slot_number(): + state = PrinterState.from_mqtt_data(make_data(real_nozzle_payload())) + snapshot = state.get_snapshot() + + by_sn = {h["serial_number"]: h for h in snapshot["hotends"]} + toolhead = by_sn["20D06A5C0426280"] + + assert toolhead["raw_id"] == 0 + assert toolhead["is_toolhead"] is True + assert toolhead["slot_number"] is None # true bay address hidden while id==0 sentinel + + +def test_rack_bay_ids_resolve_to_slot_numbers_one_through_six(): + state = PrinterState.from_mqtt_data(make_data(real_nozzle_payload())) + snapshot = state.get_snapshot() + + by_sn = {h["serial_number"]: h for h in snapshot["hotends"]} + + assert by_sn["20D06A5B2919219"]["slot_number"] == 1 # raw_id 16 + assert by_sn["20D06A591506263"]["slot_number"] == 3 # raw_id 18 + assert by_sn["20D06A5C0207881"]["slot_number"] == 4 # raw_id 19 + assert by_sn["20D06A590610257"]["slot_number"] == 5 # raw_id 20 + assert by_sn["20D06A5B2918952"]["slot_number"] == 6 # raw_id 21 + + +def test_empty_bay_with_na_serial_is_flagged_empty(): + state = PrinterState.from_mqtt_data(make_data(real_nozzle_payload())) + snapshot = state.get_snapshot() + + by_sn = {h["serial_number"]: h for h in snapshot["hotends"]} + empty = by_sn["N/A"] + + assert empty["is_empty"] is True + assert empty["is_toolhead"] is False + + +def test_snapshot_hotends_empty_list_when_no_nozzle_payload(): + state = PrinterState.from_mqtt_data({"print": {"gcode_state": "IDLE"}}) + snapshot = state.get_snapshot() + + assert snapshot["hotends"] == []