2 Commits

7 changed files with 45 additions and 505 deletions

View File

@@ -1,77 +0,0 @@
"""Add Printer.category so printer queries can be scoped away from other devices.
`infrastructure_device` is shared with host projects. In a standalone Bambu-Run
deployment bambu_run owns the table and the column must be created here. In a
host project like RAE the table was created by that project's own app and
already carries a `category` column, so creating it again would fail.
`AddFieldIfMissing` introspects the table and only emits DDL when needed; the
model state is updated either way.
"""
import django.db.models.manager
from django.db import migrations, models
class AddFieldIfMissing(migrations.AddField):
"""AddField that is a no-op at the database level if the column exists."""
def database_forwards(self, app_label, schema_editor, from_state, to_state):
model = to_state.apps.get_model(app_label, self.model_name)
with schema_editor.connection.cursor() as cursor:
existing = {
column.name
for column in schema_editor.connection.introspection.get_table_description(
cursor, model._meta.db_table
)
}
if self.name in existing:
return
super().database_forwards(app_label, schema_editor, from_state, to_state)
def database_backwards(self, app_label, schema_editor, from_state, to_state):
"""Reverse the model state only, never the column.
In a host project the column belongs to that project's own app — dropping
it on reverse would break the host's device model. Leaving an unused
column behind in a standalone rollback is the harmless side of this trade.
"""
return
class Migration(migrations.Migration):
dependencies = [
("bambu_run", "0008_printermetrics_nozzle_info"),
]
operations = [
AddFieldIfMissing(
model_name="printer",
name="category",
field=models.CharField(
default="threed_printer",
help_text=(
"Device category. Always 'threed_printer' for printers — present "
"because host projects may share this table with other device types."
),
max_length=50,
),
),
migrations.AlterModelOptions(
name="printer",
options={
"base_manager_name": "all_objects",
"default_manager_name": "objects",
"ordering": ["name"],
"verbose_name": "Printer",
"verbose_name_plural": "Printers",
},
),
migrations.AlterModelManagers(
name="printer",
managers=[
("all_objects", django.db.models.manager.Manager()),
("objects", django.db.models.manager.Manager()),
],
),
]

View File

@@ -33,33 +33,11 @@ def ams_type_from_info(info_code) -> str:
return AMS_INFO_TO_TYPE.get(code[-4:], "") or AMS_INFO_TO_TYPE.get(code, "") return AMS_INFO_TO_TYPE.get(code[-4:], "") or AMS_INFO_TO_TYPE.get(code, "")
class PrinterManager(models.Manager):
"""Default manager — scopes every query to actual 3D printers.
`Printer` shares the `infrastructure_device` table with a host project's other
device rows (RAE stores its NAS, routers and cameras there too). Without this
scoping, `Printer.objects.filter(is_active=True).first()` can return a NAS.
"""
def get_queryset(self):
return super().get_queryset().filter(category=Printer.CATEGORY_3D_PRINTER)
class Printer(models.Model): class Printer(models.Model):
"""Represents a Bambu Lab 3D printer device""" """Represents a Bambu Lab 3D printer device"""
CATEGORY_3D_PRINTER = "threed_printer"
name = models.CharField(max_length=200, help_text="Friendly device name") name = models.CharField(max_length=200, help_text="Friendly device name")
model = models.CharField(max_length=100, help_text="Device model (e.g., X1C, P1S)") model = models.CharField(max_length=100, help_text="Device model (e.g., X1C, P1S)")
category = models.CharField(
max_length=50,
default=CATEGORY_3D_PRINTER,
help_text=(
"Device category. Always 'threed_printer' for printers — present because "
"host projects may share this table with other device types."
),
)
manufacturer = models.CharField( manufacturer = models.CharField(
max_length=100, default="Bambu Lab", help_text="e.g., Bambu Lab" max_length=100, default="Bambu Lab", help_text="e.g., Bambu Lab"
) )
@@ -74,19 +52,11 @@ class Printer(models.Model):
first_seen = models.DateTimeField(auto_now_add=True) first_seen = models.DateTimeField(auto_now_add=True)
last_updated = models.DateTimeField(auto_now=True) last_updated = models.DateTimeField(auto_now=True)
# `all_objects` is declared first so it serves as the base manager for related
# descriptors (PrinterMetrics.device etc.) — those must never filter, or rows
# attached to a mis-categorised device become unreachable.
all_objects = models.Manager()
objects = PrinterManager()
class Meta: class Meta:
db_table = "infrastructure_device" db_table = "infrastructure_device"
verbose_name = "Printer" verbose_name = "Printer"
verbose_name_plural = "Printers" verbose_name_plural = "Printers"
ordering = ["name"] ordering = ["name"]
base_manager_name = "all_objects"
default_manager_name = "objects"
def __str__(self): def __str__(self):
return f"{self.name} ({self.model})" return f"{self.name} ({self.model})"

View File

@@ -14,13 +14,9 @@ from .conf import app_settings
from .models import Printer, PrinterMetrics, Filament, FilamentColor, FilamentType, FilamentSnapshot, PrintJob, FilamentUsage, Hotend from .models import Printer, PrinterMetrics, Filament, FilamentColor, FilamentType, FilamentSnapshot, PrintJob, FilamentUsage, Hotend
from .forms import FilamentForm, FilamentColorForm, FilamentTypeForm from .forms import FilamentForm, FilamentColorForm, FilamentTypeForm
# Every field the chart serializers read must be listed here. A field that is
# accessed but missing triggers a deferred-field load — one extra SELECT per row,
# which turns a single-query page into thousands.
_METRICS_API_FIELDS = [ _METRICS_API_FIELDS = [
'id', 'device_id', 'timestamp', 'id', 'device_id', 'timestamp',
'nozzle_temp', 'nozzle_target_temp', 'nozzle_temp', 'nozzle_target_temp',
'nozzle_temp_left', 'nozzle_target_temp_left',
'bed_temp', 'bed_target_temp', 'bed_temp', 'bed_target_temp',
'print_percent', 'cooling_fan_speed', 'heatbreak_fan_speed', 'print_percent', 'cooling_fan_speed', 'heatbreak_fan_speed',
'wifi_signal_dbm', 'ams_humidity_raw', 'ams_temp', 'wifi_signal_dbm', 'ams_humidity_raw', 'ams_temp',
@@ -29,9 +25,6 @@ _METRICS_API_FIELDS = [
'external_spool', 'external_spool',
] ]
_MAX_CHART_POINTS = 3000 _MAX_CHART_POINTS = 3000
# Fallback window for requests that don't specify a full date range. Without it a
# bare API call scans the entire metrics table.
_DEFAULT_WINDOW = timedelta(hours=24)
def resolve_printer_from_request(pk): def resolve_printer_from_request(pk):
@@ -39,47 +32,12 @@ def resolve_printer_from_request(pk):
`pk` given (URL kwarg) -> that exact printer, 404 if missing/inactive. `pk` given (URL kwarg) -> that exact printer, 404 if missing/inactive.
`pk` omitted -> first active printer (today's single-printer default behavior). `pk` omitted -> first active printer (today's single-printer default behavior).
Both paths go through `Printer.objects`, which is category-scoped, so a
non-printer row sharing `infrastructure_device` (a NAS, a router) can never be
resolved as "the printer" — even when no active printer exists.
""" """
if pk is not None: if pk is not None:
return get_object_or_404(Printer, pk=pk, is_active=True) return get_object_or_404(Printer, pk=pk, is_active=True)
return Printer.objects.filter(is_active=True).first() return Printer.objects.filter(is_active=True).first()
def sample_metrics(metrics_list, max_points=None):
"""Evenly thin a metrics list to at most `max_points`, always keeping the last
reading — the stat cards are built from it."""
max_points = max_points or _MAX_CHART_POINTS
total = len(metrics_list)
if total <= max_points:
return metrics_list
step = (total // max_points) + 1
sampled = metrics_list[::step]
if sampled[-1] is not metrics_list[-1]:
sampled.append(metrics_list[-1])
return sampled
def fetch_snapshots_by_metric(metrics_list):
"""Load filament snapshots for exactly the metrics we're serializing.
Beats `prefetch_related` on the unsampled queryset, which pulls a snapshot row
for every metric in the window (~25k rows for 24h) including the ones sampling
just discarded.
"""
if not metrics_list:
return {}
snapshots_by_metric = {}
for snap in FilamentSnapshot.objects.filter(
printer_metric_id__in=[m.id for m in metrics_list]
):
snapshots_by_metric.setdefault(snap.printer_metric_id, []).append(snap)
return snapshots_by_metric
class PrinterDashboardView(LoginRequiredMixin, TemplateView): class PrinterDashboardView(LoginRequiredMixin, TemplateView):
template_name = "bambu_run/printer_dashboard.html" template_name = "bambu_run/printer_dashboard.html"
@@ -114,28 +72,14 @@ class PrinterDashboardView(LoginRequiredMixin, TemplateView):
# Get date range (overridable by subclasses) # Get date range (overridable by subclasses)
start_dt, end_dt = self._get_date_range(self.request) start_dt, end_dt = self._get_date_range(self.request)
query = PrinterMetrics.objects.filter( metrics = PrinterMetrics.objects.filter(
device=printer_device, timestamp__gte=start_dt device=printer_device, timestamp__gte=start_dt
) )
if end_dt: if end_dt:
query = query.filter(timestamp__lte=end_dt) metrics = metrics.filter(timestamp__lte=end_dt)
metrics = metrics.prefetch_related('filament_snapshots').order_by("timestamp")
# Chart series only need the columns the serializer below reads, and only latest_metric = metrics.last()
# as many points as a chart can render. Fetching every column (including
# the large JSON blobs) for every row is what made this page slow.
metrics = sample_metrics(
list(query.only(*_METRICS_API_FIELDS).order_by("timestamp"))
)
snapshots_by_metric = fetch_snapshots_by_metric(metrics)
# The stat cards read far more fields than the charts do, so the latest
# reading is fetched separately as a full instance rather than deferring
# (a deferred field on a sampled row costs an extra query per access).
latest_metric = (
query.prefetch_related('filament_snapshots__filament')
.order_by("-timestamp")
.first()
)
printer_data_json = { printer_data_json = {
"timestamps": [ "timestamps": [
@@ -189,17 +133,14 @@ class PrinterDashboardView(LoginRequiredMixin, TemplateView):
"total_layer_num": [ "total_layer_num": [
m.total_layer_num if m.total_layer_num else 0 for m in metrics m.total_layer_num if m.total_layer_num else 0 for m in metrics
], ],
"filament_timeline": self._prepare_filament_timeline( "filament_timeline": self._prepare_filament_timeline(metrics),
metrics, snapshots_by_metric
),
} }
stats = {} stats = {}
if latest_metric: if latest_metric:
filaments_list = [] filaments_list = []
try: try:
# `.all()` (not `.select_related()`) so the prefetch cache is used filament_snapshots = latest_metric.filament_snapshots.select_related('filament').all()
filament_snapshots = latest_metric.filament_snapshots.all()
for snapshot in filament_snapshots: for snapshot in filament_snapshots:
filament_dict = { filament_dict = {
'tray_id': snapshot.tray_id, 'tray_id': snapshot.tray_id,
@@ -322,18 +263,18 @@ class PrinterDashboardView(LoginRequiredMixin, TemplateView):
"timestamp": latest_metric.timestamp.astimezone(tz).strftime("%Y-%m-%d %H:%M:%S"), "timestamp": latest_metric.timestamp.astimezone(tz).strftime("%Y-%m-%d %H:%M:%S"),
} }
project_markers = self._calculate_project_markers(metrics, tz, printer_device) project_markers = self._calculate_project_markers(list(metrics), tz)
printer_data_json["project_markers"] = project_markers printer_data_json["project_markers"] = project_markers
context["printer_device"] = printer_device context["printer_device"] = printer_device
context["device_name"] = printer_device.name context["device_name"] = printer_device.name
context["stats"] = stats context["stats"] = stats
context["metrics_count"] = len(metrics) context["metrics_count"] = metrics.count()
context["printer_data_json"] = json.dumps(printer_data_json) context["printer_data_json"] = json.dumps(printer_data_json)
return context return context
def _calculate_project_markers(self, metrics, timezone_info, device): def _calculate_project_markers(self, metrics, timezone_info):
"""Calculate where print jobs start and end, using cloud design_title when available.""" """Calculate where print jobs start and end, using cloud design_title when available."""
if not metrics: if not metrics:
return [] return []
@@ -341,6 +282,7 @@ class PrinterDashboardView(LoginRequiredMixin, TemplateView):
# Build a lookup: subtask_name -> display_name from PrintJobs in this time window # Build a lookup: subtask_name -> display_name from PrintJobs in this time window
window_start = metrics[0].timestamp window_start = metrics[0].timestamp
window_end = metrics[-1].timestamp window_end = metrics[-1].timestamp
device = metrics[0].device
jobs_qs = PrintJob.objects.filter( jobs_qs = PrintJob.objects.filter(
device=device, device=device,
start_time__gte=window_start - timedelta(minutes=5), start_time__gte=window_start - timedelta(minutes=5),
@@ -386,17 +328,18 @@ class PrinterDashboardView(LoginRequiredMixin, TemplateView):
return markers return markers
def _prepare_filament_timeline(self, metrics, snapshots_by_metric): def _prepare_filament_timeline(self, metrics):
"""Prepare filament data organized by unique filament configurations. """Prepare filament data organized by unique filament configurations."""
Snapshots are passed in pre-grouped by metric id; reading them off each
metric instance instead would issue one query per point.
"""
filament_data = {} filament_data = {}
total_points = len(metrics) total_points = len(metrics)
for idx, metric in enumerate(metrics): for idx, metric in enumerate(metrics):
for snapshot in snapshots_by_metric.get(metric.id, []): try:
snapshots = metric.filament_snapshots.all()
except Exception:
snapshots = []
for snapshot in snapshots:
tray_id = snapshot.tray_id tray_id = snapshot.tray_id
ams_unit_id = snapshot.ams_unit_id ams_unit_id = snapshot.ams_unit_id
ams_type = snapshot.ams_type or '' ams_type = snapshot.ams_type or ''
@@ -472,25 +415,38 @@ class PrinterDataAPIView(LoginRequiredMixin, View):
.only(*_METRICS_API_FIELDS) .only(*_METRICS_API_FIELDS)
) )
# Both bounds are always applied. A missing bound falls back to a 24h if start_date and start_time and end_date and end_time:
# window rather than being left open — an unbounded range would scan start_dt = datetime.strptime(f"{start_date} {start_time}", "%Y-%m-%d %H:%M").replace(tzinfo=tz)
# every metric ever recorded. end_dt = datetime.strptime(f"{end_date} {end_time}", "%Y-%m-%d %H:%M").replace(tzinfo=tz)
def _parse(date_str, time_str): query = query.filter(timestamp__gte=start_dt, timestamp__lte=end_dt)
return datetime.strptime( range_seconds = (end_dt - start_dt).total_seconds()
f"{date_str} {time_str}", "%Y-%m-%d %H:%M" expected_count = max(1, int(range_seconds / 30))
).replace(tzinfo=tz) elif start_date and start_time:
start_dt = datetime.strptime(f"{start_date} {start_time}", "%Y-%m-%d %H:%M").replace(tzinfo=tz)
query = query.filter(timestamp__gte=start_dt)
expected_count = _MAX_CHART_POINTS
elif end_date and end_time:
end_dt = datetime.strptime(f"{end_date} {end_time}", "%Y-%m-%d %H:%M").replace(tzinfo=tz)
query = query.filter(timestamp__lte=end_dt)
expected_count = _MAX_CHART_POINTS
else:
expected_count = _MAX_CHART_POINTS
end_dt = _parse(end_date, end_time) if end_date else timezone.now() step = max(1, expected_count // _MAX_CHART_POINTS)
start_dt = _parse(start_date, start_time) if start_date else end_dt - _DEFAULT_WINDOW
query = query.filter(timestamp__gte=start_dt, timestamp__lte=end_dt)
# Stage B: single DB round-trip, downsample in Python # Stage B: single DB round-trip, downsample in Python
metrics_list = sample_metrics(list(query.order_by("timestamp"))) metrics_list = list(query.order_by("timestamp"))
if step > 1:
metrics_list = metrics_list[::step]
total_points = len(metrics_list) total_points = len(metrics_list)
# Stage C: targeted snapshot fetch (only sampled IDs) # Stage C: targeted snapshot fetch (only sampled IDs)
snapshots_by_metric = fetch_snapshots_by_metric(metrics_list) snapshots_by_metric: dict = {}
if metrics_list:
sampled_ids = [m.id for m in metrics_list]
for snap in FilamentSnapshot.objects.filter(printer_metric_id__in=sampled_ids):
snapshots_by_metric.setdefault(snap.printer_metric_id, []).append(snap)
# Stage D: single-pass serialization # Stage D: single-pass serialization
timestamps = [] timestamps = []

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "bambu-run" name = "bambu-run"
version = "0.1.10" version = "0.1.8"
description = "Django reusable app for Bambu Lab 3D printer monitoring and filament inventory management" description = "Django reusable app for Bambu Lab 3D printer monitoring and filament inventory management"
readme = "README.md" readme = "README.md"
license = {text = "MIT"} license = {text = "MIT"}

View File

@@ -57,13 +57,8 @@ def test_filament_timeline_keeps_same_tray_id_units_separate(logged_in_client):
type="PLA", sub_type="PLA Basic", color="FF0000", remain_percent=50, type="PLA", sub_type="PLA Basic", color="FF0000", remain_percent=50,
) )
from bambu_run.views import fetch_snapshots_by_metric
view = PrinterDashboardView() view = PrinterDashboardView()
metrics = list(PrinterMetrics.objects.filter(pk=metric.pk)) timeline = view._prepare_filament_timeline(PrinterMetrics.objects.filter(pk=metric.pk))
timeline = view._prepare_filament_timeline(
metrics, fetch_snapshots_by_metric(metrics)
)
assert len(timeline) == 2 assert len(timeline) == 2

View File

@@ -1,71 +0,0 @@
"""Printer shares the `infrastructure_device` table with non-printer devices
(NAS, routers, ...) in host projects like RAE. Printer queries must never
resolve one of those rows.
"""
import pytest
from django.urls import reverse
from bambu_run.models import Printer
from bambu_run.views import resolve_printer_from_request
@pytest.fixture
def logged_in_client(client, django_user_model):
user = django_user_model.objects.create_user(username="scoping", password="pw")
client.force_login(user)
return client
@pytest.fixture
def nas():
"""A non-printer device row sharing the table, sorting before any printer."""
return Printer.all_objects.create(
name="A NAS", model="DS920+", category="nas", is_active=True
)
@pytest.mark.django_db
def test_default_manager_excludes_non_printers(nas):
printer = Printer.objects.create(name="Z Printer", model="H2C", is_active=True)
assert list(Printer.objects.all()) == [printer]
assert nas in Printer.all_objects.all()
@pytest.mark.django_db
def test_new_printers_default_to_the_printer_category():
printer = Printer.objects.create(name="Fresh", model="H2C")
assert printer.category == Printer.CATEGORY_3D_PRINTER
assert printer in Printer.objects.all()
@pytest.mark.django_db
def test_resolve_printer_skips_an_active_nas(nas):
"""The exact production failure: NAS sorts first and is active, printer is not."""
printer = Printer.objects.create(name="Z Printer", model="H2C", is_active=False)
assert resolve_printer_from_request(None) is None, "inactive printer must not resolve"
printer.is_active = True
printer.save()
assert resolve_printer_from_request(None) == printer
@pytest.mark.django_db
def test_resolve_printer_by_pk_rejects_a_non_printer(nas):
from django.http import Http404
with pytest.raises(Http404):
resolve_printer_from_request(nas.pk)
@pytest.mark.django_db
def test_dashboard_does_not_fall_back_to_a_nas(logged_in_client, nas):
resp = logged_in_client.get(reverse("bambu_run:printer_dashboard"))
assert resp.status_code == 200
assert "error" in resp.context
assert resp.context.get("printer_device") is None
assert list(resp.context["all_printers"]) == []

View File

@@ -1,233 +0,0 @@
"""Guards against the query-count and payload regressions that made the printer
pages slow: deferred-field N+1s, unbounded date ranges, and unsampled chart data.
"""
import json
from datetime import timedelta
import pytest
from django.urls import reverse
from django.utils import timezone
from bambu_run.models import Printer, PrinterMetrics, FilamentSnapshot
from bambu_run.views import _MAX_CHART_POINTS
@pytest.fixture
def logged_in_client(client, django_user_model):
user = django_user_model.objects.create_user(username="perf", password="pw")
client.force_login(user)
return client
@pytest.fixture
def printer():
return Printer.objects.create(name="Perf Printer", model="H2C", is_active=True)
def _make_metrics(printer, count, *, snapshots_per_metric=2, spacing_seconds=30):
"""Create `count` metrics ending now, each with some filament snapshots."""
now = timezone.now()
metrics = PrinterMetrics.objects.bulk_create(
[
PrinterMetrics(
device=printer,
timestamp=now - timedelta(seconds=spacing_seconds * (count - i)),
nozzle_temp=200 + i % 5,
nozzle_target_temp=220,
nozzle_temp_left=180 + i % 3,
nozzle_target_temp_left=190,
bed_temp=60,
bed_target_temp=60,
print_percent=i % 100,
gcode_state="RUNNING",
print_type="local",
subtask_name="job",
)
for i in range(count)
]
)
FilamentSnapshot.objects.bulk_create(
[
FilamentSnapshot(
printer_metric=m,
tray_id=str(tray),
type="PLA",
sub_type="Bambu",
color="FF0000FF",
remain_percent=80,
)
for m in metrics
for tray in range(snapshots_per_metric)
]
)
return metrics
# --- Root cause 1: deferred-field N+1 in the API -----------------------------
def _count_queries(client, url, params=None):
from django.db import connection
from django.test.utils import CaptureQueriesContext
with CaptureQueriesContext(connection) as ctx:
resp = client.get(url, params or {})
assert resp.status_code == 200
return len(ctx)
@pytest.mark.django_db
def test_api_query_count_is_independent_of_row_count(logged_in_client, printer):
"""Every field the serializer reads must be in .only(), or Django emits one
extra SELECT per row per missing field — making query count scale with data."""
today = timezone.localtime().date()
url = reverse("bambu_run:printer_api")
params = {
"start_date": str(today - timedelta(days=1)),
"end_date": str(today),
"start_time": "00:00",
"end_time": "23:59",
}
_make_metrics(printer, 10)
few = _count_queries(logged_in_client, url, params)
_make_metrics(printer, 190)
many = _count_queries(logged_in_client, url, params)
assert few == many, f"query count scales with rows: {few} -> {many}"
@pytest.mark.django_db
def test_api_returns_dual_nozzle_values(logged_in_client, printer):
"""The left-nozzle fields must survive the .only() narrowing."""
_make_metrics(printer, 5)
today = timezone.localtime().date()
resp = logged_in_client.get(
reverse("bambu_run:printer_api"),
{
"start_date": str(today - timedelta(days=1)),
"end_date": str(today),
"start_time": "00:00",
"end_time": "23:59",
},
)
data = resp.json()
assert any(v is not None for v in data["nozzle_temp_left"])
assert any(v is not None for v in data["nozzle_target_temp_left"])
# --- Root cause: unbounded query when date params are missing ----------------
@pytest.mark.django_db
def test_api_without_params_is_time_bounded(logged_in_client, printer):
"""A bare API call must not scan the whole table — it defaults to 24h."""
_make_metrics(printer, 10, spacing_seconds=30) # inside 24h
old = PrinterMetrics.objects.create(
device=printer, timestamp=timezone.now() - timedelta(days=30), nozzle_temp=100
)
resp = logged_in_client.get(reverse("bambu_run:printer_api"))
data = resp.json()
assert len(data["timestamps"]) == 10
assert old.timestamp.isoformat() not in data["timestamps_iso"]
@pytest.mark.django_db
def test_api_with_only_start_date_is_time_bounded(logged_in_client, printer):
"""Partial params must not drop the upper bound and scan forever."""
_make_metrics(printer, 5)
today = timezone.localtime().date()
resp = logged_in_client.get(
reverse("bambu_run:printer_api"), {"start_date": str(today - timedelta(days=1))}
)
assert resp.status_code == 200
assert len(resp.json()["timestamps"]) == 5
@pytest.mark.django_db
def test_api_downsamples_above_max_chart_points(logged_in_client, printer, monkeypatch):
monkeypatch.setattr("bambu_run.views._MAX_CHART_POINTS", 10)
_make_metrics(printer, 40, snapshots_per_metric=1, spacing_seconds=30)
today = timezone.localtime().date()
resp = logged_in_client.get(
reverse("bambu_run:printer_api"),
{
"start_date": str(today - timedelta(days=1)),
"end_date": str(today),
"start_time": "00:00",
"end_time": "23:59",
},
)
assert 0 < len(resp.json()["timestamps"]) <= 10
# --- Root cause 2: the dashboard render -------------------------------------
@pytest.mark.django_db
def test_dashboard_query_count_is_independent_of_row_count(logged_in_client, printer):
url = reverse("bambu_run:printer_dashboard")
_make_metrics(printer, 10)
few = _count_queries(logged_in_client, url)
_make_metrics(printer, 190)
many = _count_queries(logged_in_client, url)
assert few == many, f"query count scales with rows: {few} -> {many}"
@pytest.mark.django_db
def test_dashboard_downsamples_chart_payload(logged_in_client, printer, monkeypatch):
"""The dashboard inlines its JSON into the HTML, so it must sample like the API."""
monkeypatch.setattr("bambu_run.views._MAX_CHART_POINTS", 10)
_make_metrics(printer, 60, snapshots_per_metric=1)
resp = logged_in_client.get(reverse("bambu_run:printer_dashboard"))
payload = json.loads(resp.context["printer_data_json"])
assert 0 < len(payload["timestamps"]) <= 10
@pytest.mark.django_db
def test_dashboard_stats_use_the_newest_metric(logged_in_client, printer):
"""Sampling must never drop the latest reading — the stat cards depend on it."""
import zoneinfo
from bambu_run.conf import app_settings
_make_metrics(printer, 20)
newest = PrinterMetrics.objects.create(
device=printer, timestamp=timezone.now(), nozzle_temp=242, gcode_state="RUNNING"
)
resp = logged_in_client.get(reverse("bambu_run:printer_dashboard"))
assert resp.context["stats"]["nozzle_temp"] == pytest.approx(242)
assert resp.context["stats"]["timestamp"] == newest.timestamp.astimezone(
zoneinfo.ZoneInfo(app_settings.TIMEZONE)
).strftime("%Y-%m-%d %H:%M:%S")
@pytest.mark.django_db
def test_dashboard_filament_timeline_aligns_with_timestamps(logged_in_client, printer):
"""remain_data must stay index-aligned with timestamps after sampling."""
_make_metrics(printer, 30, snapshots_per_metric=2)
resp = logged_in_client.get(reverse("bambu_run:printer_dashboard"))
payload = json.loads(resp.context["printer_data_json"])
n = len(payload["timestamps"])
assert payload["filament_timeline"]
for series in payload["filament_timeline"].values():
assert len(series["remain_data"]) == n