Tujuan: menerima laporan singkat dari user (contoh: “ada asap, lantai 2”), lalu otomatis menjalankan pipeline lengkap SiGAP-Api dan mengembalikan: Analisis Risiko, Tindakan, Perhitungan Teknis, Rute Evakuasi, Notifikasi, Catatan Keamanan.

Algoritma:

1. Terima input user: teks (deskripsi), optional posisi (lantai/ruang), timestamp, jumlah orang (opsional).
2. Normalisasi & ekstraksi: ambil kata kunci (asap, bau, api, lokasi, “mata perih”, “asap pekat”).
3. sim-detect: berdasarkan kata kunci + lokasi + knowledge denah → tentukan smoke_flag/fire_flag dan initial_score (0..1).
4. Jika ada indikasi fire/asap:
   • 4.1 sim-hrr: estimasi HRR Q(t) dengan t-squared. Jika pengguna tidak memberikan t, gunakan perkiraan deteksi → asumsikan t_det sebagai 30 s (configurable).
   • 4.2 sim-alpert: hitung ΔT dan ceiling jet velocity di jarak r dari sumber (gunakan titik detektor atau jarak default 1–3 m).
   • 4.3 sim-layer: hitung smoke layer height H_layer(t) berdasarkan entrainment sederhana.
   • 4.4 sim-visibility: hitung jarak pandang dari optical density (IDW / korelasi dengan smoke density).
5. sim-route: menggunakan denah (knowledge) dan posisi user → hitung rute teraman menghindari area dengan smoke_layer tinggi atau visibility rendah. Hindari area dalam radius R_hot dari titik api.
6. sim-RSET: hitung RSET = t_det + t_recog + t_resp + travel_time; bandingkan dengan ASET (dari ΔT dan smoke layer) → tentukan aman/tidak.
7. sim-notify: buat pesan darurat standar berdasarkan level (LOW/MED/HIGH/CRITICAL) dan rute.
8. Buat output terstruktur: (a) Analisis Risiko, (b) Intention/Tindakan, (c) Perhitungan Teknis (tahapan langkah), (d) Rute Evakuasi (polyline/step), (e) Notifikasi & Catatan Keselamatan.
9. Jika data tidak cukup: minta klarifikasi singkat (lantai/ruang); bila user awam, jalankan asumsi aman default (mis. asumsikan MED/HIGH jika kata kunci “asap pekat”).

2) Flow-chart (ASCII + deskripsi singkat)

Deskripsi singkat tiap node

• USER INPUT: Teks dari user (contoh: “ada asap di lantai 2”).
• Parse & Normalize: Ekstraksi kata kunci (asap, api, lokasi) dan normalisasi format.
• sim-detect: Heuristik keyword + lokasi → skor 0..1, level (NO SMOKE / LOW / MED / HIGH / CRITICAL).
• sim-hrr: Estimasi Heat Release Rate Q(t) dengan rumus t-squared (Q = α t²).
• sim-alpert: Hitung ΔT ceiling jet dan kecepatan jet untuk prediksi aktivasi detektor panas.
• sim-layer: Estimasi tinggi layer asap (smoke layer) dan laju penurunan.
• sim-visibility: Konversi densitas asap → jarak pandang (m) dan kategori bahaya.
• sim-route: Planner rute evakuasi teraman berdasarkan denah, titik api, dan kondisi asap.
• sim-RSET: Hitung RSET (detection + recognition + response + travel) dan bandingkan dengan ASET.
• sim-notify: Bentuk pesan notifikasi darurat standar SiGAP-Api.
• Compose Output: Gabungkan semua hasil menjadi laporan terstruktur.

3) Program Python (komplet, modular, siap jalankan)

Simpan sebagai sigap_pipeline.py. Program ini bersifat self-contained (tidak tergantung API eksternal). Ia membaca knowledge file path lokal jika perlu (paths di awal). Program fokus pada perhitungan logika dan menghasilkan output terstruktur (JSON + teks).

# sigap_pipeline.py
# SiGAP-Api simulation pipeline (hybrid technical + operational)
# Requirements: pure Python standard libs (no external libs required)
# Save this file and run with Python 3.9+

import math
import json
from typing import Dict, Tuple, List, Optional

# === Configuration / Knowledge file paths (adjust if needed) ===
DENAH_PDF_PATH = "/mnt/data/Peta_Gedung_B_Lantai_2_3_CLEAR_V2.pdf"
TOOLS_DOC_PATH = "/mnt/data/TOOLS_SIMULASI_SiGAP_Api.txt"
SMOKE_KNOW_PATH = "/mnt/data/ASAP_DINAMIKA_PENDETEKSIAN.txt"
FORMULAS_PATH = "/mnt/data/Fire_Dynamics_Formula.txt"

# === Constants / thresholds ===
SCORE_THRESHOLDS = {
    "normal": 0.2,
    "warning": 0.45,
    "danger": 0.7
}

# T-squared alpha values
ALPHA = {
    "slow": 0.00293,
    "medium": 0.01172,
    "fast": 0.0469,
    "ultra": 0.1876
}

# Utility: safe numeric formatting
def fmt(x, digits=2):
    return round(x, digits)

# --------------------------
# SIM-TOOLS IMPLEMENTASI
# --------------------------

def sim_detect(description: str, location: Optional[str]=None) -> Dict:
    """
    Simple heuristic detection based on keyword matching and modifiers.
    Returns dict: {'score':0..1, 'level':str, 'evidence':str}
    """
    desc = description.lower()
    score = 0.0
    evidence = []

    # keywords and weights
    kws = {
        "asap pekat": 0.9, "asap": 0.6, "bau gosong": 0.5,
        "mata perih": 0.4, "api": 1.0, "terbakar": 0.8,
        "gelap": 0.6, "kabut": 0.5
    }
    for k,w in kws.items():
        if k in desc:
            score = max(score, w)
            evidence.append(f"keyword:{k} weight:{w}")

    # location influence (if inside server room increase suspicion)
    if location:
        loc = location.lower()
        if "server" in loc:
            score = max(score, 0.85)
            evidence.append("location:server increases risk")

    # normalize to 0..1
    score = min(1.0, score)
    # map to level
    if score <= SCORE_THRESHOLDS["normal"]:
        level = "NO SMOKE"
    elif score <= SCORE_THRESHOLDS["warning"]:
        level = "LOW"
    elif score <= SCORE_THRESHOLDS["danger"]:
        level = "MED"
    elif score < 0.95:
        level = "HIGH"
    else:
        level = "CRITICAL"

    return {"score": fmt(score,3), "level": level, "evidence": evidence}

def sim_hrr(t_seconds: float, category: str="fast") -> Dict:
    """
    Compute Q(t) = alpha * t^2
    Show steps for technical traceability.
    """
    # select alpha
    if category not in ALPHA:
        category = "fast"
    alpha = ALPHA[category]

    # Step-by-step calc (displayed in output)
    # digit-by-digit style shown as string steps for trace
    Q = alpha * (t_seconds ** 2)
    steps = [
        f"alpha = {alpha}",
        f"t = {t_seconds} s",
        f"t^2 = {t_seconds} * {t_seconds} = {t_seconds**2}",
        f"Q = alpha * t^2 = {alpha} * {t_seconds**2} = {Q} kW"
    ]
    return {"Q_kW": fmt(Q,3), "category": category, "steps": steps}

def sim_alpert(Q_kW: float, H: float, r: float) -> Dict:
    """
    Compute Alpert ceiling jet delta T.
    Uses the canonical two-case formula.
    """
    # precompute Q^(2/3)
    Q23 = Q_kW ** (2.0/3.0)
    ratio = r / H if H>0 else float('inf')
    if ratio <= 0.18:
        deltaT = 5.38 * Q23 / (H ** (5.0/3.0))
        formula_used = "r/H <= 0.18"
    else:
        deltaT = 16.9 * Q23 / (H * (r ** (5.0/3.0)))
        formula_used = "r/H > 0.18"

    # velocity estimate (approx)
    V = 0.96 * (Q_kW ** (1/3.0)) / (H ** (1/3.0))

    steps = [
        f"Q^(2/3) = {fmt(Q23,4)}",
        f"r/H = {fmt(ratio,4)} -> formula: {formula_used}",
        f"ΔT = {fmt(deltaT,3)} °C",
        f"V ≈ {fmt(V,3)} m/s"
    ]

    return {"deltaT_C": fmt(deltaT,3), "velocity_m_s": fmt(V,3), "formula": formula_used, "steps": steps}

def sim_layer(Q_kW: float, room_volume_m3: float, H_ceiling: float) -> Dict:
    """
    Simple estimation: smoke layer descent using empirical relation:
    H_layer = H_ceiling - k * sqrt(t)  (t approximated from Q).
    We'll derive a pseudo-time t_est (seconds) proportional to Q.
    """
    # crude t_est: assume higher Q => faster layer descent
    t_est = max(1.0, Q_kW / 10.0)  # seconds heuristic
    k = 0.7  # entrainment coefficient (tunable)
    H_layer = max(0.1, H_ceiling - k * math.sqrt(t_est))
    return {"H_layer_m": fmt(H_layer,3), "t_est_s": fmt(t_est,1), "k": k}

def sim_visibility(smoke_density_index: float) -> Dict:
    """
    Converts smoke density (0..10) into visibility (m) roughly.
    Use simplistic formula: V = 12 / (1 + density)
    """
    dens = max(0.0, min(10.0, smoke_density_index))
    V = 12.0 / (1.0 + dens)  # meters
    status = "SAFE"
    if V < 1.0:
        status = "UNSAFE"
    elif V < 3.0:
        status = "DANGEROUS"
    elif V < 10.0:
        status = "CAUTION"
    return {"visibility_m": fmt(V,2), "status": status, "smoke_density_index": dens}

# Simple route planner stub: will prefer straight line avoiding hot zone
def sim_route(user_pos: Tuple[float,float], exits: List[Tuple[float,float]],
              hot_zones: List[Tuple[float,float,float]]) -> Dict:
    """
    user_pos: (x,y)
    exits: list of (x,y)
    hot_zones: list of (x,y,radius)
    returns chosen exit and simple polyline (straight segments avoiding hot zones naively)
    """
    ux, uy = user_pos
    best = None
    best_dist = float('inf')

    def dist(a,b):
        return math.hypot(a[0]-b[0], a[1]-b[1])

    for ex in exits:
        d = dist(user_pos, ex)
        blocked = False
        for hx,hy,hr in hot_zones:
            # check if straight line intersects circle (approx)
            # distance from center to line segment formula approx
            # simplified: if both endpoints are outside and near the center, mark blocked
            if dist((hx,hy), user_pos) < hr or dist((hx,hy), ex) < hr:
                blocked = True
                break
        if not blocked and d < best_dist:
            best = ex
            best_dist = d

    if best is None:
        # fallback: nearest exit ignoring block
        best = min(exits, key=lambda e: dist(user_pos,e))
    polyline = [user_pos, best]
    return {"exit": best, "polyline": polyline, "distance_m": fmt(best_dist,2)}

def sim_RSET(detection_s: float, recognition_s: float, response_s: float, travel_s: float) -> Dict:
    RSET = detection_s + recognition_s + response_s + travel_s
    return {"RSET_s": fmt(RSET,1), "components": {
        "detection_s": detection_s,
        "recognition_s": recognition_s,
        "response_s": response_s,
        "travel_s": travel_s
    }}

def sim_notify(level: str, location: str, route_text: str) -> Dict:
    title = f"Peringatan Darurat: {level}"
    body = f"{title} — Lokasi: {location}. Rekomendasi: {route_text}."
    return {"title": title, "body": body}

# --------------------------
# Pipeline executor
# --------------------------
def run_pipeline(user_text: str, user_location: Optional[str]=None,
                 user_pos: Tuple[float,float]=(5.0,5.0), exits=None) -> Dict:
    if exits is None:
        # sample exits on map (x,y)
        exits = [(0.5,0.5), (15.0,0.5)]

    # 1) detect
    det = sim_detect(user_text, user_location)

    result = {"input": {"text": user_text, "location": user_location}, "steps": []}
    result["detection"] = det
    # threshold to continue
    if det["score"] <= SCORE_THRESHOLDS["normal"]:
        result["conclusion"] = "No immediate hazard detected."
        return result

    # 2) HRR estimate (assume t if not given)
    # Use t_estimation: if user mentions "baru" assume 30s, else 60s conservative
    if "baru" in user_text.lower() or "baru saja" in user_text.lower():
        t_guess = 30.0
    else:
        t_guess = 45.0
    hrr = sim_hrr(t_guess, category="fast")
    result["hrr"] = hrr

    # 3) Alpert (choose default H and r)
    H_ceiling = 3.0
    r = 2.0
    al = sim_alpert(hrr["Q_kW"], H_ceiling, r)
    result["alpert"] = al

    # 4) layer
    room_vol = 8.0 * 4.0 * H_ceiling  # heuristic room volume
    layer = sim_layer(hrr["Q_kW"], room_vol, H_ceiling)
    result["layer"] = layer

    # 5) visibility (derive smoke density index from layer: simple mapping)
    # higher descent => higher density
    density_est = max(0.5, min(10.0, 10.0*(1.0 - layer["H_layer_m"]/H_ceiling)))
    vis = sim_visibility(density_est)
    result["visibility"] = vis

    # 6) route planning (hot zone radius from Q)
    hot_radius = min(6.0, max(1.0, hrr["Q_kW"]/100.0))
    hot_zones = [(user_pos[0]+1.0, user_pos[1], hot_radius)]
    route = sim_route(user_pos, exits, hot_zones)
    result["route"] = route

    # 7) RSET: detection = t_guess / 2 (assume detected midway), recognition 10s, response 15s, travel based on distance
    detection_s = t_guess / 2.0
    recognition_s = 10.0
    response_s = 15.0
    travel_s = route["distance_m"] / 1.0  # assume 1 m/s through smoke
    rset = sim_RSET(detection_s, recognition_s, response_s, travel_s)
    result["RSET"] = rset

    # 8) ASET check (simplified): compute time until untenable using deltaT or smoke layer threshold
    # If layer height < 2.0 -> untenable time small
    untenable = (layer["H_layer_m"] < 2.0)
    result["ASSET_estimate"] = {
        "untenable": untenable,
        "reason": "layer < 2m" if untenable else "layer > 2m"
    }

    # 9) notify
    level = det["level"]
    route_text = f"Evakuasi ke exit di {route['exit']}"
    notify = sim_notify(level, user_location or "lokasi tidak valid", route_text)
    result["notify"] = notify

    # Compose human text summary
    summary = []
    summary.append(f"Analisis Risiko: level={det['level']}, score={det['score']}")
    summary.append("Tindakan: Segera evakuasi mengikuti rute yang disarankan.")
    summary.append(f"Perhitungan teknis: HRR={hrr['Q_kW']} kW. ΔT={al['deltaT_C']} °C.")
    summary.append(f"Rute: {route_text}")
    summary.append(f"RSET = {rset['RSET_s']} s. Untenable: {untenable}.")
    result["summary_text"] = "\n".join(summary)

    return result

# Example runner (for testing)
if __name__ == "__main__":
    example = run_pipeline("Ada asap pekat di koridor lantai 2, bau gosong", user_location="Koridor Lantai 2", user_pos=(8,12))
    print(json.dumps(example, indent=2, ensure_ascii=False))

4) Cara menggunakan / integrasi ke Knowledge & Model

1. Upload file TXT: (TOOLS_SIMULASI_SiGAP_Api.txt, ASAP_DINAMIKA_PENDETEKSIAN.txt, Fire_Dynamics_Formula.txt) ke Knowledge di workspace Anda. (Paths saya buat di /mnt/data/... — Anda bisa juga upload direct).
2. Buat Model (Models → New Model) dan paste System Prompt Final (sudah saya buat sebelumnya) + tambahkan instruction untuk memanggil pipeline ini bila Anda menyediakan runtime (code interpreter) OR gunakan script di server.
3. Jika OpenWebUI menyediakan Code Interpreter: unggah file sigap_pipeline.py ke environment, lalu jalankan via tool interpreter. Jika tidak, jalankan di server Python lokal.
4. Integrasi Chat: set system prompt agar saat user mengetik “ada asap” Chat DAI:
   • memanggil sigap_pipeline.run_pipeline(...) (melalui code tool) atau
   • meniru logika yang sama via sim-tools text definitions (apabila function calling tidak tersedia).

5) Contoh perhitungan (langkah aritmetika digit-by-digit — permintaan eksplisit)

Contoh: hitung Q(t) untuk alpha = 0.1876, t=30 s.

• t² = 30 × 30 = 900
• Q = 0.1876 × 900
  • 0.1876 × 900 = 0.1876 × (9 × 100) = (0.1876 × 9) × 100
  • 0.1876 × 9 = 1.6884
  • ×100 → 168.84 kW

Jadi Q(30) = 168.84 kW.

Alpert (contoh): Q^(2/3) for Q=500 kW (shown earlier in knowledge file) — steps are included in Fire_Dynamics_Formula.txt.

6) File yang sudah saya buat di sandbox (link langsung untuk Anda upload / periksa)

• Denah (yang Anda upload): /mnt/data/Peta_Gedung_B_Lantai_2_3_CLEAR_V2.pdf
• TOOLS_SIMULASI_SiGAP_Api.txt: /mnt/data/TOOLS_SIMULASI_SiGAP_Api.txt
• ASAP_DINAMIKA_PENDETEKSIAN.txt: /mnt/data/ASAP_DINAMIKA_PENDETEKSIAN.txt
• Fire_Dynamics_Formula.txt: /mnt/data/Fire_Dynamics_Formula.txt
• contoh program: sim file (simpan sebagai) sigap_pipeline.py (kode di atas)

Gunakan path lokal /mnt/data/... bila Anda ingin platform otomatis memetakan menjadi URL.

7) Catatan implementasi & asumsi penting

• Asumsi konservatif dipakai bila data tidak lengkap (mis. t_guess 30–45 s). Anda bisa ubah parameter di script.
• Rute di script adalah stub sederhana; untuk produksi gunakan A* / graph pathfinding di grid berdasarkan poligon denah.
• Visibilitas dan smoke density diestimasi via heuristik; untuk akurasi produksi rekomendasi gunakan model CFD atau sensor real-time.
• Keamanan: sistem hanya simulasi; tetap koordinasikan keputusan evakuasi dengan petugas lapangan.

Jika Anda mau, saya dapat selanjutnya:

• a) Membuat file sigap_pipeline.py di /mnt/data/ dan menaruhnya di environment sehingga Anda bisa mengeksekusinya di server (saya bisa simpan file itu untuk Anda).
• b) Mengubah sim_route menjadi A* pathfinder menggunakan denah (membutuhkan parsing PDF→floorplan coordinates).
• c) Membuat versi PDF diagram flowchart dan contoh run (laporan).
• d) Membuat integrasi chat: contoh System Prompt snippet untuk memanggil pipeline saat user menulis “ada asap”.