""" CP 707 Knowledge Base Sumber: Buku "Manajemen Broiler CP 707" oleh PT Charoen Pokphand Indonesia, Tbk. Edisi Juli 2023 Seluruh standar teknis dari buku panduan CP 707 untuk referensi on-premise AI Insight. TIDAK ada data dari sumber luar — hanya dari buku ini. Port murni Python 3.11+ dari cp707Knowledge.js (tanpa dependensi Django). """ from __future__ import annotations from typing import Any # ─── Topic-scoped metric registry ──────────────────────────────────────────── # Each topic grades ONLY its core metrics. Unknown is emitted only for the # topic's core metric(s) so the LLM cannot confuse standards with actuals. TOPIC_METRICS: dict[str, set[str]] = { "hitung_ayam": {"mortality"}, # cumulative mortality + daily mortality % of DOC "berat_ayam": {"bw", "adg"}, # body weight + ADG vs standard "fcr": {"fcr"}, # FCR vs standard only "eef": {"eef", "fcr"}, # EEF/IP + FCR (driver) "hitung_karung": {"feed_balance"}, # sack balance (in - poured - remaining) "iot_panel": {"environment"}, # temperature vs TET, humidity "dashboard": {"bw", "fcr", "mortality", "environment", "eef", "adg"}, "end_cycle": {"bw", "fcr", "mortality", "environment", "eef", "adg"}, } # EEF absolute scale bands (manager operational rules, NOT from book) # Used when book has no IP standard for the day (outside days 7-37). # >500 = UNREALISTIC/INVALID per user rule 2026-09-29 EEF_ABSOLUTE_BANDS: list[tuple[int, int, str, str]] = [ (501, 9999, "invalid", "EEF >500 tidak realistis — periksa kualitas data FCR/mortalitas/pakan"), (400, 500, "ok", "EEF {val} SANGAT BAIK (skala manajer, di luar buku CP 707)"), (350, 399, "ok", "EEF {val} BAIK (skala manajer, di luar buku CP 707)"), (300, 349, "ok", "EEF {val} STANDAR (skala manajer, di luar buku CP 707)"), (200, 299, "warning", "EEF {val} DI BAWAH RATA-RATA (skala manajer, di luar buku CP 707)"), (0, 199, "critical", "EEF {val} SANGAT RENDAH (skala manajer, di luar buku CP 707)"), ] # Daily mortality % of DOC thresholds (manager operational rules, NOT from book) # Book only has cumulative mortality standard. Daily uses its own thresholds: # >0.05% = warning, >=0.1% = critical. Divisor = DOC intake. DAILY_MORTALITY_DOC_THRESHOLDS = { "warning_pct": 0.05, "critical_pct": 0.10, } # ─── Standar Performa Mingguan CP 707 ─────────────────────────────────────── PERFORMANCE_STANDARD_WEEKLY: list[dict[str, Any]] = [ {"week": 1, "targetBW_g": 195, "adg_g": 34, "cumFeedConsumption_g": 164.5, "fcr": 0.844}, {"week": 2, "targetBW_g": 499, "adg_g": 50, "cumFeedConsumption_g": 530.5, "fcr": 1.063}, {"week": 3, "targetBW_g": 954, "adg_g": 80, "cumFeedConsumption_g": 1181.5, "fcr": 1.238}, {"week": 4, "targetBW_g": 1543, "adg_g": 88, "cumFeedConsumption_g": 2198.5, "fcr": 1.425}, {"week": 5, "targetBW_g": 2191, "adg_g": 94, "cumFeedConsumption_g": 3461, "fcr": 1.580}, ] # ─── Standar Performa Harian CP 707 (Lampiran 2) ──────────────────────────── PERFORMANCE_STANDARD_DAILY: list[dict[str, Any]] = [ {"day": 1, "bw_g": 57, "adg_g": 15, "mortalityCum_pct": 0.40, "feedDaily_g": 13, "feedCum_g": 13, "fcr": 0.228, "ip": None}, {"day": 2, "bw_g": 73, "adg_g": 16, "mortalityCum_pct": 0.50, "feedDaily_g": 17, "feedCum_g": 30, "fcr": 0.411, "ip": None}, {"day": 3, "bw_g": 90, "adg_g": 17, "mortalityCum_pct": 0.60, "feedDaily_g": 20.5, "feedCum_g": 50.5, "fcr": 0.561, "ip": None}, {"day": 4, "bw_g": 110, "adg_g": 20, "mortalityCum_pct": 0.70, "feedDaily_g": 23, "feedCum_g": 73.5, "fcr": 0.668, "ip": None}, {"day": 5, "bw_g": 134, "adg_g": 24, "mortalityCum_pct": 0.80, "feedDaily_g": 26, "feedCum_g": 99.5, "fcr": 0.743, "ip": None}, {"day": 6, "bw_g": 161, "adg_g": 27, "mortalityCum_pct": 0.90, "feedDaily_g": 31, "feedCum_g": 130.5, "fcr": 0.811, "ip": None}, {"day": 7, "bw_g": 195, "adg_g": 34, "mortalityCum_pct": 1.00, "feedDaily_g": 34, "feedCum_g": 164.5, "fcr": 0.844, "ip": 327}, {"day": 8, "bw_g": 231, "adg_g": 36, "mortalityCum_pct": 1.10, "feedDaily_g": 35, "feedCum_g": 199.5, "fcr": 0.864, "ip": 331}, {"day": 9, "bw_g": 269, "adg_g": 38, "mortalityCum_pct": 1.20, "feedDaily_g": 41, "feedCum_g": 240.5, "fcr": 0.894, "ip": 330}, {"day": 10, "bw_g": 311, "adg_g": 42, "mortalityCum_pct": 1.30, "feedDaily_g": 46, "feedCum_g": 286.5, "fcr": 0.921, "ip": 333}, {"day": 11, "bw_g": 355, "adg_g": 44, "mortalityCum_pct": 1.40, "feedDaily_g": 52, "feedCum_g": 338.5, "fcr": 0.954, "ip": 334}, {"day": 12, "bw_g": 401, "adg_g": 46, "mortalityCum_pct": 1.50, "feedDaily_g": 58, "feedCum_g": 396.5, "fcr": 0.989, "ip": 333}, {"day": 13, "bw_g": 449, "adg_g": 48, "mortalityCum_pct": 1.60, "feedDaily_g": 64, "feedCum_g": 460.5, "fcr": 1.026, "ip": 331}, {"day": 14, "bw_g": 499, "adg_g": 50, "mortalityCum_pct": 1.70, "feedDaily_g": 70, "feedCum_g": 530.5, "fcr": 1.063, "ip": 330}, {"day": 15, "bw_g": 552, "adg_g": 53, "mortalityCum_pct": 1.80, "feedDaily_g": 73, "feedCum_g": 603.5, "fcr": 1.093, "ip": 331}, {"day": 16, "bw_g": 609, "adg_g": 57, "mortalityCum_pct": 1.90, "feedDaily_g": 80, "feedCum_g": 683.5, "fcr": 1.122, "ip": 333}, {"day": 17, "bw_g": 669, "adg_g": 60, "mortalityCum_pct": 2.00, "feedDaily_g": 86, "feedCum_g": 769.5, "fcr": 1.150, "ip": 335}, {"day": 18, "bw_g": 734, "adg_g": 65, "mortalityCum_pct": 2.10, "feedDaily_g": 92, "feedCum_g": 861.5, "fcr": 1.174, "ip": 340}, {"day": 19, "bw_g": 802, "adg_g": 68, "mortalityCum_pct": 2.20, "feedDaily_g": 100, "feedCum_g": 961.5, "fcr": 1.199, "ip": 344}, {"day": 20, "bw_g": 874, "adg_g": 72, "mortalityCum_pct": 2.30, "feedDaily_g": 107, "feedCum_g": 1068.5, "fcr": 1.223, "ip": 349}, {"day": 21, "bw_g": 954, "adg_g": 80, "mortalityCum_pct": 2.40, "feedDaily_g": 113, "feedCum_g": 1181.5, "fcr": 1.238, "ip": 358}, {"day": 22, "bw_g": 1035, "adg_g": 81, "mortalityCum_pct": 2.52, "feedDaily_g": 128, "feedCum_g": 1309.5, "fcr": 1.265, "ip": 362}, {"day": 23, "bw_g": 1117, "adg_g": 82, "mortalityCum_pct": 2.64, "feedDaily_g": 133, "feedCum_g": 1442.5, "fcr": 1.291, "ip": 366}, {"day": 24, "bw_g": 1200, "adg_g": 83, "mortalityCum_pct": 2.76, "feedDaily_g": 139, "feedCum_g": 1581.5, "fcr": 1.318, "ip": 369}, {"day": 25, "bw_g": 1284, "adg_g": 84, "mortalityCum_pct": 2.88, "feedDaily_g": 145, "feedCum_g": 1726.5, "fcr": 1.345, "ip": 371}, {"day": 26, "bw_g": 1369, "adg_g": 85, "mortalityCum_pct": 3.00, "feedDaily_g": 151, "feedCum_g": 1877.5, "fcr": 1.371, "ip": 372}, {"day": 27, "bw_g": 1455, "adg_g": 86, "mortalityCum_pct": 3.12, "feedDaily_g": 157, "feedCum_g": 2034.5, "fcr": 1.398, "ip": 373}, {"day": 28, "bw_g": 1543, "adg_g": 88, "mortalityCum_pct": 3.25, "feedDaily_g": 164, "feedCum_g": 2198.5, "fcr": 1.425, "ip": 374}, {"day": 29, "bw_g": 1633, "adg_g": 90, "mortalityCum_pct": 3.38, "feedDaily_g": 173, "feedCum_g": 2371.5, "fcr": 1.452, "ip": 375}, {"day": 30, "bw_g": 1725, "adg_g": 92, "mortalityCum_pct": 3.52, "feedDaily_g": 176.5, "feedCum_g": 2548, "fcr": 1.477, "ip": 376}, {"day": 31, "bw_g": 1817, "adg_g": 92, "mortalityCum_pct": 3.66, "feedDaily_g": 178, "feedCum_g": 2726, "fcr": 1.500, "ip": 376}, {"day": 32, "bw_g": 1910, "adg_g": 93, "mortalityCum_pct": 3.80, "feedDaily_g": 180, "feedCum_g": 2906, "fcr": 1.521, "ip": 377}, {"day": 33, "bw_g": 2003, "adg_g": 93, "mortalityCum_pct": 3.95, "feedDaily_g": 183, "feedCum_g": 3089, "fcr": 1.542, "ip": 378}, {"day": 34, "bw_g": 2097, "adg_g": 94, "mortalityCum_pct": 4.10, "feedDaily_g": 185, "feedCum_g": 3274, "fcr": 1.561, "ip": 379}, {"day": 35, "bw_g": 2191, "adg_g": 94, "mortalityCum_pct": 4.25, "feedDaily_g": 187, "feedCum_g": 3461, "fcr": 1.580, "ip": 379}, {"day": 36, "bw_g": 2285, "adg_g": 94, "mortalityCum_pct": 4.45, "feedDaily_g": 190, "feedCum_g": 3651, "fcr": 1.598, "ip": 380}, {"day": 37, "bw_g": 2380, "adg_g": 95, "mortalityCum_pct": 4.65, "feedDaily_g": 193, "feedCum_g": 3844, "fcr": 1.615, "ip": 380}, ] # ─── Target Suhu Pemeliharaan per Umur (Buku CP 707) ──────────────────────── TEMPERATURE_STANDARD: list[dict[str, Any]] = [ {"ageDay_from": 0, "ageDay_to": 2, "temp_C": 32, "humidity_pct_min": 50, "humidity_pct_max": 70}, {"ageDay_from": 3, "ageDay_to": 4, "temp_C": 31, "humidity_pct_min": 50, "humidity_pct_max": 70}, {"ageDay_from": 5, "ageDay_to": 7, "temp_C": 30, "humidity_pct_min": 50, "humidity_pct_max": 70}, {"ageDay_from": 8, "ageDay_to": 14, "temp_C": 29, "humidity_pct_min": 50, "humidity_pct_max": 70}, {"ageDay_from": 15, "ageDay_to": 21, "temp_C": 28, "humidity_pct_min": 50, "humidity_pct_max": 70}, {"ageDay_from": 22, "ageDay_to": 28, "temp_C": 26, "humidity_pct_min": 50, "humidity_pct_max": 70}, {"ageDay_from": 29, "ageDay_to": 35, "temp_C": 23, "humidity_pct_min": 50, "humidity_pct_max": 70}, {"ageDay_from": 36, "ageDay_to": 99, "temp_C": 22, "humidity_pct_min": 50, "humidity_pct_max": 70}, ] # ─── Target Efektif Temperatur (TET) per Umur ─────────────────────────────── TARGET_EFFECTIVE_TEMPERATURE: list[dict[str, Any]] = [ {"ageDay_from": 0, "ageDay_to": 2, "tet_C": 32}, {"ageDay_from": 3, "ageDay_to": 4, "tet_C": 31}, {"ageDay_from": 5, "ageDay_to": 7, "tet_C": 30}, {"ageDay_from": 8, "ageDay_to": 14, "tet_C": 29}, {"ageDay_from": 15, "ageDay_to": 21, "tet_C": 27}, {"ageDay_from": 22, "ageDay_to": 28, "tet_C": 25}, {"ageDay_from": 29, "ageDay_to": 35, "tet_C": 22}, {"ageDay_from": 36, "ageDay_to": 99, "tet_C": 21}, ] # ─── Standar Kualitas Udara (Lampiran 3) ──────────────────────────────────── AIR_QUALITY_STANDARD: dict[str, Any] = { "ammonia": { "ideal_pct": "<10 ppm", "warning": 10, # >10 ppm: merusak permukaan paru-paru "critical": 25, # >25 ppm: pertumbuhan menurun "severe": 50, # >50 ppm: pertumbuhan menurun signifikan "note": ">20 ppm lebih rentan terhadap penyakit pernapasan", }, "co2": { "ideal": "<3000 ppm", "critical": 3500, # >3500 ppm: ascites dan kematian tinggi }, "co": { "ideal": "10 ppm", "warning": 50, # >50 ppm mempengaruhi kesehatan "critical": 100, # 100 ppm: meningkatkan angka kematian }, "humidity": { "ideal_after_brooding": "50-60%", "warning_high": 70, # >70% pada suhu >29°C berpengaruh pada pertumbuhan "warning_low": 50, # RH <50% selama brooding berpengaruh pada pertumbuhan }, } # ─── Standar Mortalitas CP 707 ─────────────────────────────────────────────── MORTALITY_THRESHOLDS: dict[str, Any] = { "normal_pct": 5, # <5% mortalitas dianggap normal "warning_pct": 7, # 5-7% perlu perhatian "critical_pct": 7, # >7% kritis # Standar kumulatif per umur (dari tabel harian lampiran 2) "byDayStandard": [ {"day": d["day"], "mortalityCumStd_pct": d["mortalityCum_pct"]} for d in PERFORMANCE_STANDARD_DAILY ], } # ─── Kepadatan Kandang (Buku CP 707) ───────────────────────────────────────── DENSITY_STANDARD: dict[str, Any] = { "openHouse": { "minKgPerM2": 12, "maxKgPerM2": 13, "description": "Kandang terbuka dengan ventilasi alami", }, "closedHouse": { "minKgPerM2": 24, "maxKgPerM2": 30, "description": "Kandang tertutup dapat mencapai 24-30 kg/m2", }, "byHarvestWeight": [ {"minBW_kg": 0.80, "maxBW_kg": 0.99, "density_ekorPerM2_min": 11.0, "density_ekorPerM2_max": 11.1}, {"minBW_kg": 1.00, "maxBW_kg": 1.19, "density_ekorPerM2_min": 10.0, "density_ekorPerM2_max": 10.5}, {"minBW_kg": 1.20, "maxBW_kg": 1.39, "density_ekorPerM2_min": 9.0, "density_ekorPerM2_max": 9.5}, {"minBW_kg": 1.40, "maxBW_kg": 1.59, "density_ekorPerM2_min": 8.0, "density_ekorPerM2_max": 8.5}, {"minBW_kg": 1.60, "maxBW_kg": 1.89, "density_ekorPerM2_min": 7.5, "density_ekorPerM2_max": 8.0}, {"minBW_kg": 1.90, "maxBW_kg": 99, "density_ekorPerM2_min": 7.0, "density_ekorPerM2_max": 7.5}, ], } # ─── Konsumsi Air per 1000 ekor per hari (suhu 21°C) ───────────────────────── WATER_CONSUMPTION_STANDARD: list[dict[str, Any]] = [ {"week": 1, "minLiter": 58, "maxLiter": 65}, {"week": 2, "minLiter": 102, "maxLiter": 115}, {"week": 3, "minLiter": 149, "maxLiter": 167}, {"week": 4, "minLiter": 192, "maxLiter": 216}, {"week": 5, "minLiter": 232, "maxLiter": 261}, {"week": 6, "minLiter": 274, "maxLiter": 308}, {"week": 7, "minLiter": 309, "maxLiter": 347}, {"week": 8, "minLiter": 342, "maxLiter": 385}, ] # Catatan: Di atas 21°C, kebutuhan air meningkat rata-rata 6.5% per kenaikan 1°C WATER_INCREASE_PER_DEGREE_ABOVE_21C_PCT = 6.5 # ─── Program Pencahayaan CP 707 ─────────────────────────────────────────────── LIGHTING_PROGRAM: list[dict[str, Any]] = [ {"ageDay_from": 0, "ageDay_to": 7, "onHours": 23, "darkFrom": "20:00", "darkTo": "21:00"}, {"ageDay_from": 8, "ageDay_to": 14, "onHours": 22, "darkFrom": "20:00", "darkTo": "22:00"}, {"ageDay_from": 15, "ageDay_to": 20, "onHours": 20, "darkFrom": "20:00", "darkTo": "23:00"}, {"ageDay_from": 21, "ageDay_to": 28, "onHours": 20, "darkFrom": "20:00", "darkTo": "24:00"}, {"ageDay_from": 29, "ageDay_to": 99, "onHours": 23, "darkFrom": "20:00", "darkTo": "22:00"}, ] LIGHTING_MIN_INTENSITY_LUX = 25 LIGHTING_BROODING_OPTIMAL_LUX = "40-60" LIGHTING_AFTER_15DAYS_LUX = 5 def _daily_row(day_age: Any) -> dict[str, Any] | None: try: day = int(day_age) except (TypeError, ValueError): return None for row in PERFORMANCE_STANDARD_DAILY: if row["day"] == day: return row return None def _as_float(value: Any) -> float | None: if value is None or isinstance(value, bool): return None try: n = float(value) except (TypeError, ValueError): return None if n != n: # NaN return None return n def _direction_from_delta(delta_pct: float, *, band: float = 5.0) -> str: if delta_pct > band: return "di_atas_standar" if delta_pct < -band: return "di_bawah_standar" return "sesuai_standar" def get_fcr_standard_by_day(day_age: Any) -> float | None: std = _daily_row(day_age) if std: return float(std["fcr"]) try: day = int(day_age) except (TypeError, ValueError): return None if day == 0: return None if day > 37: return 1.65 return None def get_bw_standard_by_day(day_age: Any) -> float | None: std = _daily_row(day_age) if std: return float(std["bw_g"]) try: day = int(day_age) except (TypeError, ValueError): return None if day == 0: return None if day > 37: return 2500.0 return None def get_temp_standard_by_day(day_age: Any) -> dict[str, Any]: try: day = int(day_age) except (TypeError, ValueError): return dict(TEMPERATURE_STANDARD[-1]) for row in TEMPERATURE_STANDARD: if row["ageDay_from"] <= day <= row["ageDay_to"]: return dict(row) return dict(TEMPERATURE_STANDARD[-1]) def get_tet_by_day(day_age: Any) -> float: try: day = int(day_age) except (TypeError, ValueError): return 21.0 for row in TARGET_EFFECTIVE_TEMPERATURE: if row["ageDay_from"] <= day <= row["ageDay_to"]: return float(row["tet_C"]) return 21.0 def get_ip_standard_by_day(day_age: Any) -> int | None: """ Standar IP (Indeks Performans / EEF) menurut Lampiran 2 buku CP 707. Kolom IP di tabel buku hanya terisi mulai hari ke-7 dan berhenti di hari ke-37 (nilai 380). Di luar rentang itu buku TIDAK menyatakan standar, jadi fungsi ini mengembalikan None — jangan menggantinya dengan tebakan. """ std = _daily_row(day_age) if std is None: return None ip = std.get("ip") return int(ip) if isinstance(ip, (int, float)) else None def get_ip_standard_range() -> dict[str, int] | None: """Rentang IP yang tercantum di buku, untuk konteks saat hari di luar tabel.""" values = [int(d["ip"]) for d in PERFORMANCE_STANDARD_DAILY if isinstance(d.get("ip"), (int, float))] if not values: return None return {"min": min(values), "max": max(values), "lastDay": 37} def get_mortality_cum_std_by_day(day_age: Any) -> float | None: std = _daily_row(day_age) if std: return float(std["mortalityCum_pct"]) try: day = int(day_age) except (TypeError, ValueError): return None if day == 0: return 0.40 if day > 37: return 4.65 + ((day - 37) * 0.1) return None def analyze_fcr(actual_fcr: Any, day_age: Any) -> dict[str, Any]: try: day_int = int(day_age) except (TypeError, ValueError): day_int = None if day_int == 0: return { "actual": _as_float(actual_fcr), "standard": None, "delta_pct": None, "direction": "unknown", "status": "ok", "message": "FCR belum dihitung pada hari ke-0 (DOC baru tiba / masa awal brooding).", } actual = _as_float(actual_fcr) # FCR 0 / negative is not a valid reading — treat as missing. if actual is not None and actual <= 0: actual = None std_fcr = get_fcr_standard_by_day(day_age) if actual is None or std_fcr is None: return { "actual": actual, "standard": std_fcr, "delta_pct": None, "direction": "unknown", "status": "unknown", "message": "Umur di luar standar tabel CP 707." if std_fcr is None else "FCR aktual tidak tersedia.", } delta_pct = round(((actual - std_fcr) / std_fcr) * 100, 1) direction = _direction_from_delta(delta_pct) if delta_pct > 10: status = "critical" message = ( f"FCR {actual:.2f} melebihi standar CP 707 ({std_fcr:.3f}) sebesar {delta_pct:.1f}% " f"pada umur hari ke-{day_age}. Periksa potensi pakan tercecer dan kualitas pakan." ) elif delta_pct > 5: status = "warning" message = ( f"FCR {actual:.2f} sedikit di atas standar CP 707 ({std_fcr:.3f}) " f"pada umur hari ke-{day_age}. Atur ketinggian piringan pakan dan evaluasi fines dalam pakan." ) elif delta_pct < -5: status = "ok" message = ( f"FCR {actual:.2f} lebih baik dari standar CP 707 ({std_fcr:.3f}) " f"pada umur hari ke-{day_age}. Pertahankan manajemen pakan." ) else: status = "ok" message = f"FCR {actual:.2f} sesuai standar CP 707 ({std_fcr:.3f}) pada umur hari ke-{day_age}." return { "actual": actual, "standard": std_fcr, "delta_pct": float(delta_pct), "direction": direction, "status": status, "message": message, } def analyze_bw(actual_bw_g: Any, day_age: Any, target_bw_g: Any = None) -> dict[str, Any]: try: day_int = int(day_age) except (TypeError, ValueError): day_int = None if day_int == 0: actual = _as_float(actual_bw_g) if actual is not None and actual <= 0: actual = None return { "actual": actual, "standard": actual, "delta_pct": None, "direction": "sesuai_standar" if actual is not None else "unknown", "status": "ok" if actual is not None else "unknown", "message": ( f"Bobot hari ke-0 adalah bobot awal DOC masuk ({actual:g}g) yang menjadi acuan standar baseline siklus." if actual is not None else "Bobot awal DOC masuk hari ke-0 tidak tersedia." ), } actual = _as_float(actual_bw_g) # 0 g is a missing/placeholder reading, not a real bird weight. if actual is not None and actual <= 0: actual = None std_bw = _as_float(target_bw_g) if std_bw is None or std_bw <= 0: std_bw = get_bw_standard_by_day(day_age) if actual is None or std_bw is None: return { "actual": actual, "standard": std_bw, "delta_pct": None, "direction": "unknown", "status": "unknown", "message": "Umur di luar standar tabel CP 707." if std_bw is None else "Bobot aktual tidak tersedia.", } delta_pct = round(((actual - std_bw) / std_bw) * 100, 1) direction = _direction_from_delta(delta_pct) if delta_pct < -15: status = "critical" message = ( f"Bobot badan aktual {actual:g}g jauh di bawah standar ({std_bw:g}g) " f"pada hari ke-{day_age}. Deviasi {abs(delta_pct):.1f}%. Lakukan grading segera." ) elif delta_pct < -5: status = "warning" message = ( f"Bobot badan aktual {actual:g}g sedikit di bawah standar ({std_bw:g}g) " f"pada hari ke-{day_age}. Deviasi {abs(delta_pct):.1f}%. Tingkatkan stimulasi pakan." ) elif delta_pct > 10: status = "ok" message = ( f"Bobot badan aktual {actual:g}g melampaui standar ({std_bw:g}g) " f"pada hari ke-{day_age}. Pertumbuhan sangat baik." ) else: status = "ok" message = f"Bobot badan aktual {actual:g}g sesuai standar ({std_bw:g}g) pada hari ke-{day_age}." return { "actual": actual, "standard": std_bw, "delta_pct": float(delta_pct), "direction": direction, "status": status, "message": message, } def analyze_mortality(mortality_rate_pct: Any, day_age: Any) -> dict[str, Any]: rate = _as_float(mortality_rate_pct) std_mortality = get_mortality_cum_std_by_day(day_age) if rate is None: return { "status": "unknown", "direction": "unknown", "severity_label": None, "actual_pct": None, "dilarang_kata": [], "stdMortality": std_mortality, "message": "Data mortalitas tidak tersedia.", } if rate > MORTALITY_THRESHOLDS["critical_pct"]: status = "critical" direction = "di_atas_standar" severity_label = "SANGAT TINGGI" dilarang = ["rendah", "normal", "baik", "aman", "terkendali", "sedikit"] message = ( f"Mortalitas kumulatif {rate:.2f}% adalah SANGAT TINGGI — melebihi batas kritis " f"CP 707 (>7%). Lakukan nekropsi darurat dan perketat biosekuriti." ) elif rate >= MORTALITY_THRESHOLDS["warning_pct"]: status = "warning" direction = "di_atas_standar" severity_label = "TINGGI" dilarang = ["rendah", "normal", "baik", "aman", "terkendali", "sedikit"] message = ( f"Mortalitas kumulatif {rate:.2f}% adalah TINGGI (warning). " "Standar CP 707 <5%." ) elif rate >= MORTALITY_THRESHOLDS["normal_pct"]: # 5% inclusive through just under warning threshold — still not "rendah" status = "warning" direction = "di_atas_standar" severity_label = "TINGGI" dilarang = ["rendah", "normal", "baik", "aman", "terkendali", "sedikit"] message = ( f"Mortalitas kumulatif {rate:.2f}% adalah TINGGI. " "Standar CP 707 <5%." ) elif std_mortality is not None and rate > std_mortality * 1.5: status = "warning" direction = "di_atas_standar" severity_label = "TINGGI" dilarang = ["rendah", "normal", "baik", "aman", "terkendali"] message = ( f"Mortalitas kumulatif {rate:.2f}% adalah TINGGI relatif standar harian CP 707 " f"({std_mortality:.2f}%) pada hari ke-{day_age}." ) else: status = "ok" direction = "sesuai_standar" if (std_mortality is None or rate <= std_mortality) else "di_atas_standar" severity_label = "normal/rendah" dilarang = ["SANGAT TINGGI", "kritis"] message = ( f"Mortalitas kumulatif {rate:.2f}% normal/rendah — dalam batas CP 707 (<5%)." ) return { "status": status, "direction": direction, "severity_label": severity_label, "actual_pct": rate, "dilarang_kata": dilarang, "stdMortality": std_mortality, "message": message, } def analyze_environment( temp_c: Any, humidity_pct: Any, day_age: Any, ) -> dict[str, Any]: """Temperature vs TET + humidity grading. Ammonia removed per 2026-09-28 decision: no ammonia data exists in IoT context; old code forced 0.0 making it look fake-safe. Advisory note added when humidity/temp indicate ventilation issues.""" temp = _as_float(temp_c) humidity = _as_float(humidity_pct) if temp is None: return { "status": "unknown", "direction": "unknown", "issues": [], "tempStd": get_temp_standard_by_day(day_age), "tet": get_tet_by_day(day_age), "message": "Data suhu tidak tersedia.", } if humidity is None: humidity = 65.0 temp_std = get_temp_standard_by_day(day_age) tet = get_tet_by_day(day_age) issues: list[str] = [] overall_status = "ok" if temp > temp_std["temp_C"] + 4: issues.append( f"Suhu {temp:.1f}°C jauh di atas target CP 707 ({temp_std['temp_C']}°C) " f"untuk umur hari ke-{day_age}. Nyalakan cooling pad/exhaust fan segera." ) overall_status = "critical" elif temp > temp_std["temp_C"] + 2: issues.append( f"Suhu {temp:.1f}°C di atas target CP 707 ({temp_std['temp_C']}°C) " f"untuk umur hari ke-{day_age}. Tingkatkan ventilasi." ) if overall_status != "critical": overall_status = "warning" elif temp < temp_std["temp_C"] - 3: issues.append( f"Suhu {temp:.1f}°C di bawah target CP 707 ({temp_std['temp_C']}°C). Nyalakan heater." ) if overall_status != "critical": overall_status = "warning" if humidity < temp_std["humidity_pct_min"]: issues.append( f"Kelembapan {humidity:.1f}% di bawah standar CP 707 " f"({temp_std['humidity_pct_min']}-{temp_std['humidity_pct_max']}%)." ) if overall_status != "critical": overall_status = "warning" elif humidity > temp_std["humidity_pct_max"]: issues.append( f"Kelembapan {humidity:.1f}% di atas standar CP 707 " f"({temp_std['humidity_pct_min']}-{temp_std['humidity_pct_max']}%). " "Periksa kebocoran nipple dan sekam basah." ) if overall_status != "critical": overall_status = "warning" # Advisory: if humidity/temp suggest ventilation problems, advise checking ammonia # (not a graded status; no ammonia data in context) if overall_status != "ok" and humidity > 75: issues.append( "Kondisi kelembapan tinggi + suhu off-target → periksa kadar amonia di kandang " "(batas aman CP 707 <10 ppm)." ) return { "status": overall_status, "direction": "sesuai_standar" if overall_status == "ok" else "di_atas_standar", "issues": issues, "tempStd": temp_std, "tet": tet, "message": ( " ".join(issues) if issues else f"Lingkungan kandang sesuai standar CP 707 untuk umur hari ke-{day_age}." ), } def analyze_adg(actual_adg: Any, day_age: Any) -> dict[str, Any]: """ADG (Average Daily Gain) vs daily standard from book.""" try: day_int = int(day_age) except (TypeError, ValueError): day_int = None if day_int == 0: return { "actual": _as_float(actual_adg), "standard": None, "delta_pct": None, "direction": "unknown", "status": "ok", "message": "ADG belum dihitung pada hari ke-0 (hari kedatangan DOC).", } actual = _as_float(actual_adg) if actual is not None and actual <= 0: actual = None std_row = _daily_row(day_age) std_adg = float(std_row["adg_g"]) if std_row else None if actual is None or std_adg is None: return { "actual": actual, "standard": std_adg, "delta_pct": None, "direction": "unknown", "status": "unknown", "message": ( "Umur di luar standar tabel CP 707." if std_adg is None else "ADG aktual tidak tersedia." ), } delta_pct = round(((actual - std_adg) / std_adg) * 100, 1) direction = _direction_from_delta(delta_pct) if delta_pct < -15: status = "critical" message = ( f"ADG {actual:.1f}g/hari jauh di bawah standar CP 707 ({std_adg}g/hari) " f"pada hari ke-{day_age}. Deviasi {abs(delta_pct):.1f}%. Periksa konsumsi pakan dan kesehatan ayam." ) elif delta_pct < -5: status = "warning" message = ( f"ADG {actual:.1f}g/hari sedikit di bawah standar CP 707 ({std_adg}g/hari) " f"pada hari ke-{day_age}. Deviasi {abs(delta_pct):.1f}%. Tingkatkan stimulasi pakan." ) elif delta_pct > 10: status = "ok" message = ( f"ADG {actual:.1f}g/hari melampaui standar CP 707 ({std_adg}g/hari) " f"pada hari ke-{day_age}. Pertumbuhan sangat baik." ) else: status = "ok" message = f"ADG {actual:.1f}g/hari sesuai standar CP 707 ({std_adg}g/hari) pada hari ke-{day_age}." return { "actual": actual, "standard": std_adg, "delta_pct": float(delta_pct), "direction": direction, "status": status, "message": message, } def _grade_eef_absolute_scale(eef_value: float) -> tuple[str, str]: """Grade EEF against manager absolute scale bands. Returns (status, message).""" for lo, hi, status, msg_template in EEF_ABSOLUTE_BANDS: if lo <= eef_value <= hi: if status == "invalid": return status, msg_template return status, msg_template.format(val=round(eef_value)) # Should not reach here; fallback return "unknown", f"EEF {round(eef_value)} tidak dapat digrading" def _compute_eef_trend(eef_series: list[dict[str, Any]] | None) -> dict[str, Any] | None: """ Compute EEF trend from daily series. Returns dict with: trend ("up" | "down" | "stable"), delta_3d (points), escalation ("sharp_drop_eef" | None). """ if not eef_series or not isinstance(eef_series, list): return None # Filter valid EEF points, sort by day valid = [] for p in eef_series: if not isinstance(p, dict): continue day = _as_float(p.get("hari") or p.get("day") or p.get("age")) eef = _as_float(p.get("eef") or p.get("ip") or p.get("indeks")) if day is not None and eef is not None and eef > 0: valid.append({"day": int(day), "eef": float(eef)}) if len(valid) < 3: return None valid.sort(key=lambda x: x["day"]) # Use last 3 days for trend last3 = valid[-3:] if len(last3) < 3: return None delta_3d = round(last3[-1]["eef"] - last3[0]["eef"], 1) if delta_3d >= 10: trend = "up" elif delta_3d <= -10: trend = "down" else: trend = "stable" escalation = None # Sharp drop: >20 points decline in 3 days if delta_3d <= -20: escalation = "sharp_drop_eef" return { "trend": trend, "delta_3d": delta_3d, "escalation": escalation, "series_used": last3, } def _compute_eef_from_components( fcr: float | None, mortality_cum_pct: float | None, day_age: int | None, bw_kg: float | None ) -> float | None: """ Compute EEF = (BW_kg * viability% * 100) / (FCR * age_days) viability% = 100 - cumulative_mortality% Returns None if any required component missing. """ if fcr is None or mortality_cum_pct is None or day_age is None or bw_kg is None: return None viability = 100.0 - float(mortality_cum_pct) if viability <= 0: return None try: eef = (bw_kg * viability * 100.0) / (fcr * day_age) return round(eef, 1) except ZeroDivisionError: return None def analyze_eef( eef_terakhir: Any, day_age: Any, fcr: Any = None, mortality_cum_pct: Any = None, bw_kg: Any = None, eef_seri: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """ EEF (IP) grading with dual ladder: 1. Book standard exists (days 7-37): delta vs book IP = primary status; trend = context 2. No book standard: manager absolute scale (<200/200-300/300-350/350-400/400-500, >500=invalid) 3. Trend always computed from eef_seri; sharp drop escalates ladder verdict one level. """ try: day_int = int(day_age) except (TypeError, ValueError): day_int = None if day_int == 0: return { "actual": None, "standard": None, "delta_pct": None, "direction": "unknown", "status": "ok", "message": "EEF/IP belum dihitung pada hari ke-0 (DOC baru tiba / masa awal brooding).", "trend": None, "source": "none", } eef_val = _as_float(eef_terakhir) if eef_val is not None and eef_val <= 0: eef_val = None # Try to compute EEF from components if not provided if eef_val is None: fcr_f = _as_float(fcr) mort_f = _as_float(mortality_cum_pct) bw_f = _as_float(bw_kg) day = _as_float(day_age) if fcr_f and mort_f and bw_f and day: eef_val = _compute_eef_from_components(fcr_f, mort_f, int(day), bw_f) # Get book IP standard for this day ip_std = get_ip_standard_by_day(day_age) if day_age else None trend_info = _compute_eef_trend(eef_seri) # Start with book standard if available if eef_val is not None and ip_std is not None: delta_pct = round(((eef_val - ip_std) / ip_std) * 100, 1) direction = _direction_from_delta(delta_pct) if delta_pct > 10: status = "ok" message = f"EEF/IP {eef_val:.0f} melampaui standar buku ({ip_std}) sebesar {delta_pct:.1f}% pada hari ke-{day_age}." elif delta_pct < -15: status = "critical" message = f"EEF/IP {eef_val:.0f} jauh di bawah standar buku ({ip_std}) sebesar {abs(delta_pct):.1f}% pada hari ke-{day_age}. Evaluasi FCR dan mortalitas." elif delta_pct < -5: status = "warning" message = f"EEF/IP {eef_val:.0f} di bawah standar buku ({ip_std}) sebesar {abs(delta_pct):.1f}% pada hari ke-{day_age}. Perbaiki efisiensi pakan." else: status = "ok" message = f"EEF/IP {eef_val:.0f} sesuai standar buku ({ip_std}) pada hari ke-{day_age}." # Apply trend escalation if trend_info and trend_info.get("escalation") == "sharp_drop_eef": if status == "ok": status = "warning" message += " Tren turun tajam (>20 poin/3 hari) — eskalasi ke warning." elif status == "warning": status = "critical" message += " Tren turun tajam (>20 poin/3 hari) — eskalasi ke critical." return { "actual": eef_val, "standard": ip_std, "delta_pct": delta_pct, "direction": direction, "status": status, "message": message, "trend": trend_info, "source": "book_standard", } # No book standard → manager absolute scale if eef_val is not None: status, base_message = _grade_eef_absolute_scale(eef_val) trend_context = "" if trend_info: trend_map = {"up": "naik", "down": "turun", "stable": "stabil"} trend_context = f" Tren {trend_map.get(trend_info['trend'], trend_info['trend'])} {trend_info['delta_3d']:.0f} poin/3 hari." if trend_info.get("escalation") == "sharp_drop_eef": # Escalate one level if status == "ok": status = "warning" trend_context += " Eskalasi sharp drop → warning." elif status == "warning": status = "critical" trend_context += " Eskalasi sharp drop → critical." message = base_message + trend_context return { "actual": eef_val, "standard": None, "delta_pct": None, "direction": "unknown", "status": status, "message": message, "trend": trend_info, "source": "manager_absolute_scale", } # No data at all return { "actual": None, "standard": ip_std, "delta_pct": None, "direction": "unknown", "status": "unknown", "message": "EEF/IP tidak tersedia (data FCR/mortalitas/bobot/umur tidak lengkap).", "trend": trend_info, "source": "none", } def analyze_daily_mortality_doc( mati_hari_ini: Any, doc_in_ekor: Any, ) -> dict[str, Any]: """ Daily mortality % of DOC: (mati_hari_ini / doc_in_ekor) * 100 Thresholds: >0.05% = warning, >=0.1% = critical (manager rules, not book). """ deaths = _as_float(mati_hari_ini) doc = _as_float(doc_in_ekor) if deaths is None or doc is None or doc <= 0: return { "status": "unknown", "direction": "unknown", "actual_pct": None, "message": "Data mortalitas harian atau DOC tidak tersedia.", } daily_pct = round((deaths / doc) * 100, 4) if daily_pct >= DAILY_MORTALITY_DOC_THRESHOLDS["critical_pct"]: status = "critical" message = ( f"Mortalitas harian {daily_pct:.4f}% dari DOC ({int(deaths)}/{int(doc)}) " f"≥ {DAILY_MORTALITY_DOC_THRESHOLDS['critical_pct']}% — CRITICAL. " "Segera cek penyebab kematian hari ini." ) elif daily_pct > DAILY_MORTALITY_DOC_THRESHOLDS["warning_pct"]: status = "warning" message = ( f"Mortalitas harian {daily_pct:.4f}% dari DOC ({int(deaths)}/{int(doc)}) " f"> {DAILY_MORTALITY_DOC_THRESHOLDS['warning_pct']}% — WARNING. " "Pantau dan cek kebersihan/biosekuriti." ) else: status = "ok" message = ( f"Mortalitas harian {daily_pct:.4f}% dari DOC ({int(deaths)}/{int(doc)}) " "dalam batas normal (≤0.05%)." ) return { "status": status, "direction": "di_atas_standar" if status != "ok" else "sesuai_standar", "actual_pct": daily_pct, "message": message, "deaths": deaths, "doc": doc, } def analyze_feed_balance( karung_masuk: Any, karung_tuang: Any, karung_sisa: Any, target_harian: Any = None, saldo_awal: Any = None, ) -> dict[str, Any]: """ Sack balance: (saldo_awal + masuk) - poured - remaining. Computes feed inventory status based on available stock vs daily usage. """ masuk = _as_float(karung_masuk) tuang = _as_float(karung_tuang) sisa = _as_float(karung_sisa) saldo_awal_num = _as_float(saldo_awal) if masuk is None and saldo_awal_num is None: return { "status": "unknown", "direction": "unknown", "message": "Data karung masuk tidak tersedia.", } total_masuk = (masuk or 0.0) + (saldo_awal_num or 0.0) balance = total_masuk - (tuang or 0.0) - (sisa or 0.0) iot_vs_manual = None if balance < 0: return { "status": "warning", "direction": "di_bawah_standar", "balance_karung": balance, "message": f"Saldo karung negatif ({balance:.1f} karung): total pakan dituang ({tuang or 0.0:.1f} karung) melebihi total stok masuk ({total_masuk:.1f} karung). Periksa mutasi karung.", } target = _as_float(target_harian) if target is not None and target > 0: if balance == 0: return { "status": "warning", "direction": "di_bawah_standar", "balance_karung": balance, "message": f"Saldo karung habis (0 karung). Perlu segera restock pakan untuk kebutuhan harian ({target:.1f} karung/hari).", } if balance < target: return { "status": "warning", "direction": "di_bawah_standar", "balance_karung": balance, "message": f"Saldo karung menipis ({balance:.1f} karung), di bawah kebutuhan rata-rata harian ({target:.1f} karung/hari). Perlu persiapan restock pakan.", } days_left = round(balance / target, 1) return { "status": "ok", "direction": "sesuai_standar", "balance_karung": balance, "iot_vs_manual": iot_vs_manual, "message": f"Stok pakan aman. Saldo karung tersisa {balance:.1f} karung (estimasi cukup untuk {days_left:.1f} hari dengan rata-rata konsumsi {target:.1f} karung/hari).", } detail_parts = [] if saldo_awal_num is not None: detail_parts.append(f"saldo awal {saldo_awal_num:.1f}") if masuk is not None: detail_parts.append(f"masuk {masuk:.1f}") masuk_str = " + ".join(detail_parts) if detail_parts else f"masuk {total_masuk:.1f}" return { "status": "ok", "direction": "sesuai_standar", "balance_karung": balance, "iot_vs_manual": iot_vs_manual, "message": f"Saldo karung: {masuk_str} - tuang {tuang or 0.0:.1f} - keluar {sisa or 0.0:.1f} = {balance:.1f} karung.", } def build_book_reference_context(day_age: Any) -> str: std_bw = get_bw_standard_by_day(day_age) std_fcr = get_fcr_standard_by_day(day_age) std_mortality = get_mortality_cum_std_by_day(day_age) temp_std = get_temp_standard_by_day(day_age) tet = get_tet_by_day(day_age) bw_target_str = ( "Mengikuti bobot awal DOC masuk (bukan standar baku tabel)" if day_age == 0 else (f"{std_bw} gram" if std_bw is not None else "N/A") ) return f""" === REFERENSI STANDAR BUKU CP 707 (SUMBER TUNGGAL - PT CHAROEN POKPHAND INDONESIA) === Umur ayam hari ke-{day_age}: - Target bobot badan: {bw_target_str} - Target FCR: {std_fcr if std_fcr is not None else 'N/A'} - Target mortalitas kumulatif normal: <5% (standar CP 707) - Target mortalitas kumulatif standar hari ke-{day_age}: {str(std_mortality) + '%' if std_mortality is not None else 'N/A'} - Target suhu kandang: {temp_std['temp_C']}°C - Target kelembapan: {temp_std['humidity_pct_min']}%-{temp_std['humidity_pct_max']}% - Target Efektif Temperatur (TET): {tet}°C - Batas amonia aman: <10 ppm (ideal), >25 ppm = pertumbuhan menurun (dari Lampiran 3 CP 707) - Batas CO2 aman: <3000 ppm (>3500 ppm = ascites & kematian tinggi) - Konsumsi air normal: 2-2.5x konsumsi pakan CATATAN PENTING: - Analisis HANYA berdasarkan standar buku CP 707 - Jika ada data yang tidak ada di buku CP 707, nyatakan "data tidak tersedia di buku CP 707" - Risiko yang disebutkan HARUS berdasarkan data aktual vs standar CP 707 === END REFERENSI === """ def build_cp707_standard_block(context_data: dict[str, Any] | None) -> str: """ Blok standar CP 707 berlabel satuan untuk hari yang sedang dilihat (setara buildCp707StandardBlock di aiInsights.js lama). """ if not isinstance(context_data, dict): return "" raw = None for key in ("hari_ke", "hari_terakhir", "currentDay", "dayAge"): if context_data.get(key) is not None: raw = context_data.get(key) break day_f = _as_float(raw) if day_f is None or day_f < 0: return "" day = int(day_f) ip = get_ip_standard_by_day(day) ip_range = get_ip_standard_range() target_bw_ctx = _resolve_target_bw(context_data, day) bw = target_bw_ctx if target_bw_ctx is not None else get_bw_standard_by_day(day) fcr = get_fcr_standard_by_day(day) mort = get_mortality_cum_std_by_day(day) bw_line = ( "- Bobot badan standar hari ke-0: Mengikuti bobot awal DOC masuk (bukan standar baku tabel)" if day == 0 else f"- Bobot badan standar hari ke-{day}: {str(bw) + ' gram' if bw is not None else 'tidak tercantum di buku'}" ) lines = [ bw_line, f"- FCR standar hari ke-{day}: {str(fcr) + ' (rasio, tanpa satuan)' if fcr is not None else 'tidak tercantum di buku'}", f"- Mortalitas kumulatif standar hari ke-{day}: {str(mort) + ' %' if mort is not None else 'tidak tercantum di buku'}", ] if mort is not None: hidup = round((100 - mort) * 100) / 100 lines.append( f"- Persen hidup standar hari ke-{day}: {hidup} % " "(turunan langsung dari mortalitas standar di atas — pakai angka ini, jangan menghitung sendiri)" ) else: lines.append(f"- Persen hidup standar hari ke-{day}: tidak tersedia") if ip is not None: lines.append(f"- IP/EEF standar hari ke-{day}: {ip} (indeks, TANPA satuan)") else: last_day = ip_range["lastDay"] if ip_range else 37 rng = f"{ip_range['min']}-{ip_range['max']}" if ip_range else "327-380" lines.append( f"- IP/EEF standar hari ke-{day}: TIDAK tercantum di buku. " f"Kolom IP pada Lampiran 2 hanya terisi hari ke-7 s/d ke-{last_day} dengan rentang {rng}. " "Jangan mengarang standar untuk hari ini." ) joined = "\n".join(lines) return f""" ═══════════════════════════════════════════════ STANDAR CP 707 UNTUK HARI INI (angka resmi dari Lampiran 2 buku) {joined} ATURAN WAJIB saat membandingkan dengan standar: - Pakai HANYA angka di blok ini sebagai standar. DILARANG mengambil angka standar dari tabel mentah di kutipan buku di bawah — kolomnya tidak berjudul, dan angka pakan (gram) sering tertukar menjadi standar IP/EEF. - Sebutkan satuan dengan benar: gram untuk bobot dan pakan, persen untuk mortalitas, dan IP/EEF adalah indeks TANPA satuan. - Jika standar untuk suatu metrik tidak tercantum, tulis "standar tidak tersedia di buku untuk hari ini" — jangan mengganti dengan angka lain. - Semua angka aktual harus berasal dari [Data Halaman (JSON)]. Dilarang menghitung sendiri atau mengarang angka yang tidak ada di sana. - Angka di blok ini adalah STANDAR saja (bukan nilai aktual kandang). Jika bobot/mortalitas aktual tidak ada di Data Halaman / GRADED FACTS, tulis "data tidak tersedia" — JANGAN menyalin angka standar sebagai aktual. ═══════════════════════════════════════════════""" def _nested_get(obj: Any, *path: str) -> Any: cur = obj for key in path: if not isinstance(cur, dict): return None cur = cur.get(key) return cur def _first_number(*candidates: Any) -> float | None: for value in candidates: n = _as_float(value) if n is not None: return n return None def _first_positive_number(*candidates: Any) -> float | None: """Like `_first_number`, but treat 0 / negative as missing (sensor placeholder).""" for value in candidates: n = _as_float(value) if n is not None and n > 0: return n return None def _resolve_day_age(context_pack: dict[str, Any]) -> int | None: day = _first_number( context_pack.get("hari_ke"), _nested_get(context_pack, "metadata", "currentDay"), _nested_get(context_pack, "cycle", "currentAgeDay"), context_pack.get("ageDay"), context_pack.get("umurHari"), context_pack.get("currentDay"), context_pack.get("dayAge"), ) if day is None or day < 0: return None return int(day) def _resolve_bw_grams(context_pack: dict[str, Any]) -> float | None: """Bobot dalam gram. Field `_kg` hanya untuk legacy payload (dikonversi ×1000). Values <= 0 are treated as unavailable (common IoT/placeholder zero). """ grams = _first_positive_number( context_pack.get("bobot_avg_gram"), context_pack.get("averageWeight"), context_pack.get("bobot_rata_rata_gram"), context_pack.get("bobot_iot_gram_terakhir"), context_pack.get("bobot_iot_gram"), _nested_get(context_pack, "weightStats", "averageWeight", "value"), _nested_get(context_pack, "weight", "current", "averageWeight"), context_pack.get("berat_rata_rata"), ) if grams is not None: return grams kg = _first_positive_number( context_pack.get("bobot_iot_kg_terakhir"), context_pack.get("bobot_iot_kg"), ) if kg is not None: return kg * 1000.0 return None def _resolve_fcr(context_pack: dict[str, Any]) -> float | None: """FCR ratio; 0 / negative counts as missing.""" return _first_positive_number( context_pack.get("fcr_terakhir"), context_pack.get("fcr"), _nested_get(context_pack, "fcr_eef", "fcr", "actual"), _nested_get(context_pack, "weight", "current", "fcr"), _nested_get(context_pack, "fcrSummary", "current"), ) def _resolve_adg(context_pack: dict[str, Any]) -> float | None: """ADG in g/hari from explicit fields.""" return _first_positive_number( context_pack.get("adg_gram"), context_pack.get("adg_g"), context_pack.get("average_daily_gain"), _nested_get(context_pack, "weightStats", "adg", "value"), _nested_get(context_pack, "weight", "current", "adg"), ) def _resolve_mortality(context_pack: dict[str, Any]) -> float | None: """Cumulative mortality % from explicit fields, % hidup, or kematian/DOC.""" direct = _first_number( context_pack.get("mortalitas_kumulatif_pct"), context_pack.get("mortalitas_kumulatif_persen"), context_pack.get("mortalitas_persen"), context_pack.get("mortalityRate"), context_pack.get("mortality_rate_pct"), _nested_get(context_pack, "chickenCounting", "dashboardSummary", "mortalityRate"), ) if direct is not None: return direct persen_hidup = _first_number(context_pack.get("persen_hidup")) if persen_hidup is not None: return round((100.0 - persen_hidup) * 100) / 100 death = _first_number( context_pack.get("kematian_kumulatif_ekor"), context_pack.get("mortalitas_kumulatif_ekor"), ) doc = _first_number( context_pack.get("doc_in_ekor"), context_pack.get("doc_in"), context_pack.get("initialPopulation"), ) if death is not None and doc is not None and doc > 0: return round((death / doc) * 10000) / 100 return None def _resolve_iot(context_pack: dict[str, Any]) -> dict[str, Any]: for key in ("iotPanel", "panel_iot", "telemetry", "iotData", "iot"): raw = context_pack.get(key) if isinstance(raw, dict): display = raw.get("display") if isinstance(display, dict): return display return raw # Flat InsightContext fields (rebuild) return context_pack def _resolve_target_bw(context_pack: dict[str, Any] | None, day_age: int | None) -> float | None: if not isinstance(context_pack, dict): return None direct = _first_positive_number( context_pack.get("target_bw_gram"), context_pack.get("targetWeight"), context_pack.get("bobot_target_gram"), context_pack.get("target_gram"), context_pack.get("target_weight"), ) if direct is not None: return direct tren = context_pack.get("tren_harian") if isinstance(tren, list) and day_age is not None: for row in tren: if isinstance(row, dict) and row.get("hari") == day_age: t = _first_positive_number(row.get("target_gram"), row.get("target_bw_gram"), row.get("target")) if t is not None: return t return None def analyze_with_cp707_standards(context_pack: dict[str, Any] | None, topic: str | None = None) -> dict[str, Any]: """Analisis komprehensif; toleran terhadap field InsightContext rebuild dan pack lama. Topic-scoped: only grades metrics in TOPIC_METRICS[topic]. Unknown emitted only for topic-core metrics so LLM cannot confuse standards as actuals. """ pack = context_pack if isinstance(context_pack, dict) else {} day_age = _resolve_day_age(pack) result: dict[str, Any] = {"dayAge": day_age, "analyses": {}} # Determine which metrics to grade for this topic metrics_to_grade = TOPIC_METRICS.get(topic, set()) if topic else {"bw", "fcr", "mortality", "environment"} # Body weight grading if "bw" in metrics_to_grade: actual_bw = _resolve_bw_grams(pack) target_bw = _resolve_target_bw(pack, day_age) if day_age is not None: result["analyses"]["bw"] = analyze_bw(actual_bw, day_age, target_bw) else: result["analyses"]["bw"] = {"status": "unknown", "message": "Umur tidak tersedia untuk grading bobot badan."} # FCR grading if "fcr" in metrics_to_grade: actual_fcr = _resolve_fcr(pack) if day_age is not None: result["analyses"]["fcr"] = analyze_fcr(actual_fcr, day_age) else: result["analyses"]["fcr"] = {"status": "unknown", "message": "Umur tidak tersedia untuk grading FCR."} # Cumulative mortality grading if "mortality" in metrics_to_grade: mortality_rate = _resolve_mortality(pack) result["analyses"]["mortality"] = analyze_mortality(mortality_rate, day_age) # ADG grading (berat_ayam, dashboard, end_cycle) if "adg" in metrics_to_grade: actual_adg = _resolve_adg(pack) if day_age is not None: result["analyses"]["adg"] = analyze_adg(actual_adg, day_age) else: result["analyses"]["adg"] = {"status": "unknown", "message": "Umur tidak tersedia untuk grading ADG."} # EEF grading (eef, dashboard, end_cycle) if "eef" in metrics_to_grade: eef_terakhir = _first_number(pack.get("eef_terakhir"), pack.get("ip_terakhir"), pack.get("indeks_prestasi")) fcr = _first_number(pack.get("fcr_terakhir"), pack.get("fcr")) mortality_cum_pct = _first_number(pack.get("mortalitas_kumulatif_persen"), pack.get("mortality_cum_pct")) bw_kg = _first_number(pack.get("bobot_avg_gram"), pack.get("averageWeight")) if bw_kg is not None: bw_kg = bw_kg / 1000.0 eef_seri = pack.get("eef_seri") if day_age is not None: result["analyses"]["eef"] = analyze_eef(eef_terakhir, day_age, fcr, mortality_cum_pct, bw_kg, eef_seri) else: result["analyses"]["eef"] = {"status": "unknown", "message": "Umur tidak tersedia untuk grading EEF/IP."} # Daily mortality % DOC grading (hitung_ayam, dashboard, end_cycle) if "mortality" in metrics_to_grade and topic in ("hitung_ayam", "dashboard", "end_cycle"): deaths_today = _first_number(pack.get("mortalitas_hari_ini_ekor"), pack.get("mati_hari_ini")) doc_in = _first_number(pack.get("doc_in_ekor"), pack.get("doc")) if deaths_today is not None and doc_in is not None and doc_in > 0: result["analyses"]["daily_mortality_doc"] = analyze_daily_mortality_doc(deaths_today, doc_in) # Feed balance grading (hitung_karung) if "feed_balance" in metrics_to_grade: saldo_awal = _first_number( pack.get("saldo_awal_karung"), pack.get("saldo_awal"), pack.get("initial_balance"), pack.get("feed_initial_balance"), ) karung_masuk = _first_number( pack.get("karung_masuk"), pack.get("karungMasuk_karung"), pack.get("total_karung_masuk_iot"), pack.get("total_karung_masuk_manual"), pack.get("sacks_in"), ) karung_tuang = _first_number( pack.get("karung_tuang"), pack.get("karungDituang_karung"), pack.get("total_karung_dituang_iot"), pack.get("total_karung_dituang_manual"), pack.get("sacks_poured"), ) karung_sisa = _first_number( pack.get("karung_sisa"), pack.get("karungKeluar_karung"), pack.get("total_karung_keluar_iot"), pack.get("total_karung_keluar_manual"), pack.get("sacks_remaining"), ) target_harian = _first_number( pack.get("target_harian"), pack.get("daily_target"), pack.get("rata_karung_per_hari"), pack.get("rataKarungDituangPerHari_karung"), ) if karung_masuk is not None or saldo_awal is not None: result["analyses"]["feed_balance"] = analyze_feed_balance( karung_masuk, karung_tuang, karung_sisa, target_harian, saldo_awal=saldo_awal ) else: result["analyses"]["feed_balance"] = {"status": "unknown", "message": "Data karung masuk tidak tersedia."} # Environment grading (iot_panel, dashboard, end_cycle) - NO ammonia if "environment" in metrics_to_grade: iot = _resolve_iot(pack) temp = _first_number( iot.get("suhu_rata_rata_C"), iot.get("avgTemp"), pack.get("suhu_rata_rata_C"), ) hum = _first_number( iot.get("kelembapan_persen"), iot.get("humidity"), pack.get("kelembapan_persen"), ) if temp is not None and day_age is not None: result["analyses"]["environment"] = analyze_environment( temp, hum if hum is not None else 65.0, day_age, ) return result