diff --git a/bambu_run/mcp/jobs.py b/bambu_run/mcp/jobs.py new file mode 100644 index 0000000..da14696 --- /dev/null +++ b/bambu_run/mcp/jobs.py @@ -0,0 +1,74 @@ +"""list_print_jobs, get_print_job, get_print_statistics -- the +read_print_history capability tools.""" +from __future__ import annotations + +import statistics +from datetime import timedelta + +from django.db.models import Avg, Count, ExpressionWrapper, F, FloatField, Q, Sum +from django.utils import timezone + +from bambu_run.mcp.shapes import display_name, normalize_consumed, resolve_printer +from bambu_run.models import BambuCloudTask, FilamentUsage, PrintJob + + +def list_print_jobs( + status: str | None = None, + days: int = 30, + name_contains: str | None = None, + printer: str | int | None = None, + limit: int = 20, +) -> dict: + """Print job history, most recent first. + + status filters on FINISH/FAILED/CANCELLED, or "running" for the + in-flight job (final_status is null while printing). filament_grams + prefers Bambu Cloud's reported weight over the noisier AMS percentage + estimate -- check filament_grams_source ("cloud" vs None, meaning no + cloud data) before trusting precision. cost is always null until + filament purchase prices are recorded. limit has a hard ceiling of 100, + independent of days -- a busy printer's older jobs simply won't be + returned by one call; narrow days or increase limit deliberately. + """ + limit = min(limit, 100) + qs = PrintJob.objects.select_related("cloud_task", "device") + if printer is not None: + qs = qs.filter(device__in=resolve_printer(printer)) + since = timezone.now() - timedelta(days=days) + qs = qs.filter(start_time__gte=since) + if status == "running": + qs = qs.filter(final_status__isnull=True) + elif status: + qs = qs.filter(final_status=status) + if name_contains: + qs = qs.filter(project_name__icontains=name_contains) + + jobs = list(qs.order_by("-start_time")[:limit]) + return {"jobs": [_job_summary(j) for j in jobs]} + + +def _job_summary(job): + cloud = job.cloud_task + est_min = int(cloud.cost_time_seconds / 60) if cloud and cloud.cost_time_seconds else None + grams = None + grams_source = None + if cloud and cloud.weight_grams is not None: + grams = float(cloud.weight_grams) + grams_source = "cloud" + return { + "id": job.id, + "display_name": display_name( + cloud.design_title if cloud else "", cloud.plate_title if cloud else "", job.project_name + ), + "status": job.final_status, + "printer": job.device.name, + "started_at": job.start_time.isoformat(), + "ended_at": job.end_time.isoformat() if job.end_time else None, + "duration_min": job.duration_minutes, + "estimated_duration_min": est_min, + "layers_total": job.total_layers, + "filament_grams": grams, + "filament_grams_source": grams_source, + "cost": None, + "has_cover_image": bool(cloud and cloud.cover_url), + } diff --git a/tests/test_mcp_jobs.py b/tests/test_mcp_jobs.py new file mode 100644 index 0000000..17c958c --- /dev/null +++ b/tests/test_mcp_jobs.py @@ -0,0 +1,71 @@ +import pytest +from datetime import timedelta +from django.utils import timezone + +from bambu_run.models import Printer, PrintJob, BambuCloudTask +from bambu_run.mcp.jobs import list_print_jobs + + +@pytest.mark.django_db +def test_list_print_jobs_orders_most_recent_first(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + now = timezone.now() + older = PrintJob.objects.create(device=p, project_name="Old", start_time=now - timedelta(days=1), final_status="FINISH") + newer = PrintJob.objects.create(device=p, project_name="New", start_time=now, final_status="FINISH") + + result = list_print_jobs() + + assert [j["id"] for j in result["jobs"]] == [newer.id, older.id] + + +@pytest.mark.django_db +def test_list_print_jobs_filters_by_status_running(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + running = PrintJob.objects.create(device=p, project_name="Live", start_time=timezone.now(), final_status=None) + PrintJob.objects.create(device=p, project_name="Done", start_time=timezone.now(), final_status="FINISH") + + result = list_print_jobs(status="running") + + assert [j["id"] for j in result["jobs"]] == [running.id] + + +@pytest.mark.django_db +def test_list_print_jobs_limit_hard_ceiling_100(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + now = timezone.now() + PrintJob.objects.bulk_create([ + PrintJob(device=p, project_name=f"Job {i}", start_time=now - timedelta(hours=i), final_status="FINISH") + for i in range(150) + ]) + + result = list_print_jobs(limit=1000) + + assert len(result["jobs"]) == 100 + + +@pytest.mark.django_db +def test_list_print_jobs_prefers_cloud_weight_and_tags_source(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + cloud = BambuCloudTask.objects.create(task_id=1, design_title="Cool Bracket", weight_grams=70.10) + PrintJob.objects.create( + device=p, project_name="raw_name", start_time=timezone.now(), + final_status="FINISH", cloud_task=cloud, + ) + + result = list_print_jobs() + + job = result["jobs"][0] + assert job["display_name"] == "Cool Bracket" + assert job["filament_grams"] == 70.10 + assert job["filament_grams_source"] == "cloud" + assert job["cost"] is None + + +@pytest.mark.django_db +def test_list_print_jobs_display_name_falls_back_to_project_name(): + p = Printer.objects.create(name="RNL-H2C", model="H2C", is_active=True) + PrintJob.objects.create(device=p, project_name="raw_project", start_time=timezone.now(), final_status="FINISH") + + result = list_print_jobs() + + assert result["jobs"][0]["display_name"] == "raw_project"