Spaces:
Sleeping
Sleeping
File size: 4,379 Bytes
9541352 297eef6 9541352 297eef6 9541352 ae00b72 9541352 297eef6 9541352 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 |
import logging, sys, asyncio, random, json, requests, os
from pathlib import Path
from dotenv import load_dotenv
# ============================================================
# LOGGER SETUP
# ============================================================
logger = logging.getLogger("ai_engine_dummy")
logger.setLevel(logging.INFO)
# Handler untuk tampil di console
handler = logging.StreamHandler(sys.stdout)
formatter = logging.Formatter("[%(asctime)s] %(levelname)s → %(message)s", "%H:%M:%S")
handler.setFormatter(formatter)
logger.addHandler(handler)
# ============================================================
# HELPER FUNCTION (opsional untuk simulasi delay)
# ============================================================
async def async_sleep_random(min_s=0.2, max_s=0.8):
"""
Helper untuk simulasi waktu inferensi secara acak.
"""
durasi = random.uniform(min_s, max_s)
await asyncio.sleep(durasi)
# ==============================================================
# VALIDATION FUNCTION
# ==============================================================
def validate_input(required_parts_fields, station_id, cameras, parts, webhook_url):
errors = []
# Validate station_id
if not station_id or not str(station_id).strip():
errors.append("station_id is missing or empty")
# Validate parts
for field in required_parts_fields:
if field not in parts or not str(parts[field]).strip():
errors.append(f"parts.{field} is missing or empty")
# Validate cameras list
if not isinstance(cameras, list) or len(cameras) == 0:
errors.append("cameras must be a non-empty list")
else:
for index, cam in enumerate(cameras):
if "camera_id" not in cam or not str(cam["camera_id"]).strip():
errors.append(f"camera[{index}].camera_id missing or empty")
if "image_base64" not in cam or not str(cam["image_base64"]).strip():
errors.append(f"camera[{index}].image_base64 missing or empty")
# if "rtsp_url" not in cam or not str(cam["rtsp_url"]).strip():
# errors.append(f"camera[{index}].rtsp_url missing or empty")
# Validate webhook
if not webhook_url or not webhook_url.startswith("http"):
errors.append("webhook_url is invalid or missing")
return errors
# ============================================================
# HELPER
# ============================================================
def _metadata():
"""
load file metadata.json into json data
"""
path = Path("metadata/product.json")
if not path.exists():
return {"status": "error", "message": "metadata.json not found"}
with open(path, "r") as f:
data = json.load(f)
return data
def _color_map():
path = Path("metadata/defect.json")
if not path.exists():
return []
with open(path, "r") as f:
return json.load(f)
def model_by_id_metadata(part_id):
"""
part_id = "4" # id part / product
get model_path from metadata by part_id
"""
metadata = _metadata()
id_part = part_id
item = next((x for x in metadata if x["id"] == id_part), None)
model_path = item['model_path']
return model_path
def model_by_pin_metadata(pin_api):
"""
pin_api = "JI4ACL-GCSBYHBK03" # PIN API from part / product
get model_path from metadata by pin_api
"""
metadata = _metadata()
pin_api = pin_api
item = next((x for x in metadata if x["pin_api"] == pin_api), None)
model_path = item['model_path']
return model_path
def color_defect(defect_name):
"""
Return BGR tuple color for OpenCV bounding box.
"""
defect_name = defect_name.lower()
data = _color_map()
item = next((x for x in data if x["class"].lower() == defect_name), None)
if item is None:
return (0, 255, 0) # Default GREEN jika tidak ditemukan
hex_color = item["color"].lstrip('#')
r = int(hex_color[0:2], 16)
g = int(hex_color[2:4], 16)
b = int(hex_color[4:6], 16)
return (b, g, r) # Convert to BGR
def load_json(path):
with open(path, "r") as f:
return json.load(f)
def start_detection(base_url, payload):
return requests.post(f"{base_url}/start-detection", json=payload)
def load_config():
load_dotenv()
return os.getenv("BASE_URL"), os.getenv("WEBHOOK") |