CV-04
from fastapi import FastAPI, UploadFile, File
from fastapi.responses import HTMLResponse, JSONResponse
import cv2
import numpy as np
from pathlib import Path
from datetime import datetime
import base64
import shutil
# ============================================================
# SMART INCUBATOR VISION
# CV-04 — EGG COUNTER
# ESP32-S3 CAM + OV5640
# ============================================================
app = FastAPI(
title="Smart Incubator Vision CV-04",
version="0.4.0"
)
# ============================================================
# FOLDER
# ============================================================
BASE_DIR = Path(r"C:\smart-incubator")
IMAGE_DIR = BASE_DIR / "storage" / "images"
# ============================================================
# CV-04 PARAMETERS
# ============================================================
# Laptop tidak perlu memproses 5 MP penuh.
# OV5640 tetap boleh mengirim gambar resolusi tinggi,
# tetapi OpenCV akan mengecilkannya untuk proses deteksi.
PROCESS_MAX_WIDTH = 1280
# ------------------------------------------------------------
# FILTER UKURAN
# ------------------------------------------------------------
# Luas minimum objek dibanding luas gambar.
MIN_AREA_RATIO = 0.00015
# Luas maksimum objek.
MAX_AREA_RATIO = 0.18
# Sisi terkecil bounding box terhadap ukuran gambar.
MIN_SIDE_RATIO = 0.012
# Telur berbentuk oval.
# 1.0 = bulat sempurna.
# Telur biasanya > 1.
MAX_ASPECT_RATIO = 3.0
# ------------------------------------------------------------
# FILTER BENTUK
# ------------------------------------------------------------
# Solidity:
#
# area contour / area convex hull
#
# Telur relatif solid.
MIN_SOLIDITY = 0.72
# Extent:
#
# area contour / bounding rectangle
#
MIN_EXTENT = 0.35
# Circularity:
#
# 4*pi*area / perimeter^2
#
# Telur tidak harus bulat sempurna,
# sehingga threshold dibuat cukup rendah.
MIN_CIRCULARITY = 0.30
# ============================================================
# HSV THRESHOLD
# ============================================================
# ------------------------------------------------------------
# TELUR PUTIH / TERANG
#
# Saturasi rendah + brightness tinggi
# ------------------------------------------------------------
WHITE_S_MAX = 65
WHITE_V_MIN = 145
# ------------------------------------------------------------
# TELUR COKLAT
# ------------------------------------------------------------
BROWN_H_MIN = 0
BROWN_H_MAX = 35
BROWN_S_MIN = 35
BROWN_S_MAX = 200
BROWN_V_MIN = 55
# ============================================================
# ADAPTIVE THRESHOLD
# ============================================================
ADAPTIVE_BLOCK = 31
ADAPTIVE_C = 7
# ============================================================
# UTILITIES
# ============================================================
def ensure_dirs():
IMAGE_DIR.mkdir(
parents=True,
exist_ok=True
)
# ------------------------------------------------------------
def latest_image():
ensure_dirs()
files = list(
IMAGE_DIR.rglob("*")
)
files = [
p for p in files
if p.suffix.lower()
in [".jpg", ".jpeg", ".png"]
]
if not files:
return None
return max(
files,
key=lambda p: p.stat().st_mtime
)
# ------------------------------------------------------------
def read_latest():
path = latest_image()
if path is None:
return None, None
img = cv2.imread(
str(path)
)
if img is None:
return None, path
return img, path
# ------------------------------------------------------------
def resize_for_processing(img):
height, width = img.shape[:2]
if width <= PROCESS_MAX_WIDTH:
return img.copy(), 1.0
scale = (
PROCESS_MAX_WIDTH /
float(width)
)
new_width = int(
width * scale
)
new_height = int(
height * scale
)
resized = cv2.resize(
img,
(new_width, new_height),
interpolation=cv2.INTER_AREA
)
return resized, scale
# ------------------------------------------------------------
def encode_jpg(
img,
quality=88
):
success, buffer = cv2.imencode(
".jpg",
img,
[
cv2.IMWRITE_JPEG_QUALITY,
quality
]
)
if not success:
return ""
encoded = base64.b64encode(
buffer.tobytes()
).decode("ascii")
return (
"data:image/jpeg;base64,"
+ encoded
)
# ------------------------------------------------------------
def safe_float(value):
return round(
float(value),
4
)
# ============================================================
# BUILD MASKS
# ============================================================
def build_masks(img):
# -----------------------------------------
# Gaussian blur
# -----------------------------------------
blur = cv2.GaussianBlur(
img,
(5, 5),
0
)
# -----------------------------------------
# HSV
# -----------------------------------------
hsv = cv2.cvtColor(
blur,
cv2.COLOR_BGR2HSV
)
# -----------------------------------------
# GRAYSCALE
# -----------------------------------------
gray = cv2.cvtColor(
blur,
cv2.COLOR_BGR2GRAY
)
# ========================================================
# MASK 1
# WHITE EGG
# ========================================================
white_mask = cv2.inRange(
hsv,
np.array(
[
0,
0,
WHITE_V_MIN
],
dtype=np.uint8
),
np.array(
[
179,
WHITE_S_MAX,
255
],
dtype=np.uint8
)
)
# ========================================================
# MASK 2
# BROWN EGG
# ========================================================
brown_mask = cv2.inRange(
hsv,
np.array(
[
BROWN_H_MIN,
BROWN_S_MIN,
BROWN_V_MIN
],
dtype=np.uint8
),
np.array(
[
BROWN_H_MAX,
BROWN_S_MAX,
255
],
dtype=np.uint8
)
)
# ========================================================
# MASK 3
# OTSU
# ========================================================
_, otsu_mask = cv2.threshold(
gray,
0,
255,
cv2.THRESH_BINARY
+
cv2.THRESH_OTSU
)
# ========================================================
# MASK 4
# ADAPTIVE
# ========================================================
adaptive_mask = cv2.adaptiveThreshold(
gray,
255,
cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
cv2.THRESH_BINARY,
ADAPTIVE_BLOCK,
ADAPTIVE_C
)
masks = {
"white_hsv":
white_mask,
"brown_hsv":
brown_mask,
"otsu":
otsu_mask,
"adaptive":
adaptive_mask
}
# ========================================================
# MORPHOLOGY
# ========================================================
kernel_open = cv2.getStructuringElement(
cv2.MORPH_ELLIPSE,
(3, 3)
)
kernel_close = cv2.getStructuringElement(
cv2.MORPH_ELLIPSE,
(7, 7)
)
cleaned = {}
for name, mask in masks.items():
cleaned_mask = cv2.morphologyEx(
mask,
cv2.MORPH_OPEN,
kernel_open,
iterations=1
)
cleaned_mask = cv2.morphologyEx(
cleaned_mask,
cv2.MORPH_CLOSE,
kernel_close,
iterations=2
)
cleaned[name] = cleaned_mask
return cleaned, gray
# ============================================================
# CONTOUR FEATURES
# ============================================================
def contour_features(
contour,
image_area
):
area = cv2.contourArea(
contour
)
perimeter = cv2.arcLength(
contour,
True
)
if perimeter <= 0:
return None
x, y, w, h = cv2.boundingRect(
contour
)
if w <= 0 or h <= 0:
return None
# --------------------------------------------------------
# Convex hull
# --------------------------------------------------------
hull = cv2.convexHull(
contour
)
hull_area = cv2.contourArea(
hull
)
if hull_area > 0:
solidity = (
area /
hull_area
)
else:
solidity = 0.0
# --------------------------------------------------------
# Extent
# --------------------------------------------------------
extent = (
area /
float(w * h)
)
# --------------------------------------------------------
# Bounding aspect
# --------------------------------------------------------
aspect = (
max(w, h) /
float(min(w, h))
)
# --------------------------------------------------------
# Rotated rectangle
# --------------------------------------------------------
rect = cv2.minAreaRect(
contour
)
rw, rh = rect[1]
if rw > 0 and rh > 0:
rotated_aspect = (
max(rw, rh) /
min(rw, rh)
)
else:
rotated_aspect = aspect
# --------------------------------------------------------
# Circularity
# --------------------------------------------------------
circularity = (
4.0
*
np.pi
*
area
/
(perimeter * perimeter)
)
# --------------------------------------------------------
# Center
# --------------------------------------------------------
center_x = (
x +
w / 2.0
)
center_y = (
y +
h / 2.0
)
return {
"area":
area,
"area_ratio":
area /
float(image_area),
"perimeter":
perimeter,
"x":
x,
"y":
y,
"w":
w,
"h":
h,
"aspect":
aspect,
"rot_aspect":
rotated_aspect,
"solidity":
solidity,
"extent":
extent,
"circularity":
circularity,
"center_x":
center_x,
"center_y":
center_y
}
# ============================================================
# EGG SCORE
# ============================================================
def egg_score(features):
# -----------------------------------------
# Aspect score
# -----------------------------------------
aspect_score = max(
0.0,
1.0
-
max(
0.0,
features["rot_aspect"] - 1.0
)
/
2.0
)
# -----------------------------------------
# Solidity
# -----------------------------------------
solidity_score = min(
1.0,
features["solidity"]
/
0.95
)
# -----------------------------------------
# Circularity
# -----------------------------------------
circular_score = min(
1.0,
features["circularity"]
/
0.75
)
# -----------------------------------------
# Extent
# -----------------------------------------
extent_score = min(
1.0,
features["extent"]
/
0.80
)
# -----------------------------------------
# Combined score
# -----------------------------------------
score = (
0.25 * aspect_score
+
0.30 * solidity_score
+
0.25 * circular_score
+
0.20 * extent_score
)
return round(
score * 100.0,
1
)
# ============================================================
# EGG DETECTION
# ============================================================
def detect_eggs(img):
# ========================================================
# RESIZE
# ========================================================
processed, scale = resize_for_processing(
img
)
height, width = processed.shape[:2]
image_area = (
height *
width
)
# ========================================================
# BUILD MASKS
# ========================================================
masks, gray = build_masks(
processed
)
# ========================================================
# COLLECT CANDIDATES
# ========================================================
candidates = []
for mask_name, mask in masks.items():
contours, _ = cv2.findContours(
mask,
cv2.RETR_EXTERNAL,
cv2.CHAIN_APPROX_SIMPLE
)
for contour in contours:
features = contour_features(
contour,
image_area
)
if features is None:
continue
# ------------------------------------------------
# AREA
# ------------------------------------------------
if (
features["area_ratio"]
<
MIN_AREA_RATIO
):
continue
if (
features["area_ratio"]
>
MAX_AREA_RATIO
):
continue
# ------------------------------------------------
# MINIMUM OBJECT SIZE
# ------------------------------------------------
min_side_ratio = (
min(
features["w"],
features["h"]
)
/
float(
min(
width,
height
)
)
)
if (
min_side_ratio
<
MIN_SIDE_RATIO
):
continue
# ------------------------------------------------
# ASPECT
# ------------------------------------------------
if (
features["rot_aspect"]
>
MAX_ASPECT_RATIO
):
continue
# ------------------------------------------------
# SOLIDITY
# ------------------------------------------------
if (
features["solidity"]
<
MIN_SOLIDITY
):
continue
# ------------------------------------------------
# EXTENT
# ------------------------------------------------
if (
features["extent"]
<
MIN_EXTENT
):
continue
# ------------------------------------------------
# CIRCULARITY
# ------------------------------------------------
if (
features["circularity"]
<
MIN_CIRCULARITY
):
continue
# ------------------------------------------------
# SCORE
# ------------------------------------------------
features["mask"] = mask_name
features["score"] = egg_score(
features
)
candidates.append(
features
)
# ========================================================
# SORT BY SCORE
# ========================================================
candidates.sort(
key=lambda item:
item["score"],
reverse=True
)
# ========================================================
# REMOVE DUPLICATES
# ========================================================
selected = []
def intersection_over_union(
a,
b
):
ax1 = a["x"]
ay1 = a["y"]
ax2 = (
a["x"] +
a["w"]
)
ay2 = (
a["y"] +
a["h"]
)
bx1 = b["x"]
by1 = b["y"]
bx2 = (
b["x"] +
b["w"]
)
by2 = (
b["y"] +
b["h"]
)
ix1 = max(
ax1,
bx1
)
iy1 = max(
ay1,
by1
)
ix2 = min(
ax2,
bx2
)
iy2 = min(
ay2,
by2
)
iw = max(
0,
ix2 - ix1
)
ih = max(
0,
iy2 - iy1
)
intersection = (
iw *
ih
)
union = (
a["w"] *
a["h"]
+
b["w"] *
b["h"]
-
intersection
)
if union <= 0:
return 0.0
return (
intersection /
union
)
for candidate in candidates:
duplicate = False
for existing in selected:
center_distance = np.hypot(
candidate["center_x"]
-
existing["center_x"],
candidate["center_y"]
-
existing["center_y"]
)
reference_size = max(
min(
candidate["w"],
candidate["h"]
),
min(
existing["w"],
existing["h"]
)
)
if (
center_distance
<
0.45 *
reference_size
or
intersection_over_union(
candidate,
existing
)
>
0.35
):
duplicate = True
break
if not duplicate:
selected.append(
candidate
)
# ========================================================
# SORT BY POSITION
# ========================================================
selected.sort(
key=lambda item: (
item["center_y"],
item["center_x"]
)
)
# ========================================================
# DRAW RESULT
# ========================================================
result = processed.copy()
for index, egg in enumerate(
selected,
1
):
x = egg["x"]
y = egg["y"]
egg_width = egg["w"]
egg_height = egg["h"]
# bounding box
cv2.rectangle(
result,
(x, y),
(
x + egg_width,
y + egg_height
),
(0, 255, 0),
3
)
# center
center_x = int(
egg["center_x"]
)
center_y = int(
egg["center_y"]
)
cv2.circle(
result,
(
center_x,
center_y
),
5,
(0, 0, 255),
-1
)
# label
label = (
f"Egg {index} "
f"{egg['score']:.0f}%"
)
cv2.putText(
result,
label,
(
x,
max(
25,
y - 8
)
),
cv2.FONT_HERSHEY_SIMPLEX,
0.65,
(0, 255, 0),
2,
cv2.LINE_AA
)
return {
"processed":
processed,
"result":
result,
"masks":
masks,
"eggs":
selected,
"scale":
scale,
"threshold_info": {
"white_hsv":
(
f"S <= {WHITE_S_MAX}, "
f"V >= {WHITE_V_MIN}"
),
"brown_hsv":
(
f"H {BROWN_H_MIN}-"
f"{BROWN_H_MAX}, "
f"S {BROWN_S_MIN}-"
f"{BROWN_S_MAX}, "
f"V >= {BROWN_V_MIN}"
),
"otsu":
"automatic",
"adaptive":
(
f"block={ADAPTIVE_BLOCK}, "
f"C={ADAPTIVE_C}"
)
}
}
# ============================================================
# HOME
# ============================================================
@app.get(
"/",
response_class=HTMLResponse
)
def home():
return """
<html>
<head>
<title>
Smart Incubator Vision
</title>
</head>
<body
style="
font-family:Arial;
padding:30px;
"
>
<h1>
Smart Incubator Vision
</h1>
<p>
CV-04 Egg Counter
</p>
<p>
<a href="/vision">
Open Vision Dashboard
</a>
</p>
<p>
<a href="/health">
Health
</a>
</p>
</body>
</html>
"""
# ============================================================
# HEALTH
# ============================================================
@app.get(
"/health"
)
def health():
return {
"status":
"ok",
"module":
"CV-04 Egg Counter"
}
# ============================================================
# UPLOAD
# ============================================================
@app.post(
"/api/v1/vision/upload"
)
async def upload_image(
file: UploadFile = File(...)
):
ensure_dirs()
# --------------------------------------------------------
# CHECK FILE
# --------------------------------------------------------
if (
not file.content_type
or
not file.content_type.startswith(
"image/"
)
):
return JSONResponse(
status_code=400,
content={
"error":
"File harus berupa gambar."
}
)
# --------------------------------------------------------
# DATE FOLDER
# --------------------------------------------------------
date_dir = (
IMAGE_DIR
/
datetime.now().strftime(
"%Y-%m-%d"
)
)
date_dir.mkdir(
parents=True,
exist_ok=True
)
# --------------------------------------------------------
# EXTENSION
# --------------------------------------------------------
suffix = Path(
file.filename or "image.jpg"
).suffix.lower()
if suffix not in [
".jpg",
".jpeg",
".png"
]:
suffix = ".jpg"
# --------------------------------------------------------
# FILENAME
# --------------------------------------------------------
filename = (
datetime.now().strftime(
"%H%M%S_%f"
)
+
suffix
)
path = (
date_dir /
filename
)
# --------------------------------------------------------
# SAVE
# --------------------------------------------------------
with path.open(
"wb"
) as buffer:
shutil.copyfileobj(
file.file,
buffer
)
return {
"status":
"success",
"filename":
filename,
"path":
str(path),
"url":
"/vision"
}
# ============================================================
# CV-02 ANALYSIS
# ============================================================
@app.get(
"/api/v1/vision/analyze/latest"
)
def analyze_latest():
img, path = read_latest()
if img is None:
return JSONResponse(
status_code=404,
content={
"error":
"Belum ada gambar."
}
)
gray = cv2.cvtColor(
img,
cv2.COLOR_BGR2GRAY
)
brightness = float(
np.mean(gray)
)
contrast = float(
np.std(gray)
)
return {
"filename":
path.name,
"resolution": {
"width":
int(img.shape[1]),
"height":
int(img.shape[0])
},
"brightness":
round(
brightness,
2
),
"contrast":
round(
contrast,
2
)
}
# ============================================================
# CV-04 JSON
# ============================================================
@app.get(
"/api/v1/vision/preprocess/latest"
)
def preprocess_latest():
img, path = read_latest()
if img is None:
return JSONResponse(
status_code=404,
content={
"error":
"Belum ada gambar."
}
)
data = detect_eggs(
img
)
return {
"filename":
path.name,
"resolution": {
"width":
int(img.shape[1]),
"height":
int(img.shape[0])
},
"egg_count":
len(
data["eggs"]
),
"thresholds":
data[
"threshold_info"
],
"eggs": [
{
"id":
i + 1,
"x":
int(
egg["x"]
),
"y":
int(
egg["y"]
),
"width":
int(
egg["w"]
),
"height":
int(
egg["h"]
),
"area":
round(
float(
egg["area"]
),
1
),
"area_ratio":
safe_float(
egg[
"area_ratio"
]
),
"aspect_ratio":
safe_float(
egg[
"rot_aspect"
]
),
"solidity":
safe_float(
egg[
"solidity"
]
),
"circularity":
safe_float(
egg[
"circularity"
]
),
"confidence_heuristic":
egg[
"score"
],
"mask":
egg[
"mask"
]
}
for i, egg
in enumerate(
data["eggs"]
)
]
}
# ============================================================
# VISION DASHBOARD
# ============================================================
@app.get(
"/vision",
response_class=HTMLResponse
)
def vision_dashboard():
img, path = read_latest()
if img is None:
return """
<html>
<body
style="
font-family:Arial;
padding:30px
"
>
<h1>
CV-04 Egg Counter
</h1>
<p>
Belum ada gambar.
</p>
</body>
</html>
"""
# ========================================================
# BASIC IMAGE INFO
# ========================================================
gray = cv2.cvtColor(
img,
cv2.COLOR_BGR2GRAY
)
brightness = float(
np.mean(gray)
)
contrast = float(
np.std(gray)
)
# ========================================================
# DETECTION
# ========================================================
data = detect_eggs(
img
)
eggs = data[
"eggs"
]
# ========================================================
# MASK CARDS
# ========================================================
cards = []
for name, mask in data[
"masks"
].items():
cards.append(
f"""
<div>
<h3>
{name}
</h3>
<img
src="
{encode_jpg(mask)}
"
style="
width:100%;
max-width:500px;
border:1px solid #aaa
"
>
</div>
"""
)
# ========================================================
# EGG TABLE
# ========================================================
egg_rows = ""
for i, egg in enumerate(
eggs,
1
):
egg_rows += f"""
<tr>
<td>
{i}
</td>
<td>
{egg['mask']}
</td>
<td>
{egg['w']}
×
{egg['h']}
</td>
<td>
{egg['area']:.0f}
</td>
<td>
{egg['rot_aspect']:.2f}
</td>
<td>
{egg['solidity']:.2f}
</td>
<td>
{egg['circularity']:.2f}
</td>
<td>
{egg['score']:.1f}%
</td>
</tr>
"""
if not egg_rows:
egg_rows = """
<tr>
<td colspan="8">
Belum ada telur
terdeteksi
</td>
</tr>
"""
# ========================================================
# HTML
# ========================================================
return f"""
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta
http-equiv="refresh"
content="5"
>
<title>
CV-04 Egg Counter
</title>
<style>
body {{
font-family:
Arial;
background:
#f4f4f4;
margin:
0;
padding:
20px;
}}
.box {{
background:
white;
padding:
18px;
margin-bottom:
18px;
border-radius:
10px;
}}
.grid {{
display:
grid;
grid-template-columns:
repeat(
auto-fit,
minmax(
300px,
1fr
)
);
gap:
15px;
}}
table {{
width:
100%;
border-collapse:
collapse;
}}
th,
td {{
border:
1px solid #ccc;
padding:
7px;
text-align:
center;
}}
th {{
background:
#eee;
}}
.count {{
font-size:
42px;
font-weight:
bold;
}}
</style>
</head>
<body>
<!-- ============================================= -->
<!-- INFO -->
<!-- ============================================= -->
<div class="box">
<h1>
CV-04 — Egg Counter
</h1>
<p>
<b>File:</b>
{path.name}
</p>
<p>
<b>Resolution:</b>
{img.shape[1]}
×
{img.shape[0]}
</p>
<p>
<b>Brightness:</b>
{brightness:.2f}
</p>
<p>
<b>Contrast:</b>
{contrast:.2f}
</p>
</div>
<!-- ============================================= -->
<!-- COUNT -->
<!-- ============================================= -->
<div class="box">
<div>
DETECTED EGGS
</div>
<div class="count">
{len(eggs)}
</div>
<p>
Score bentuk adalah
<b>
confidence heuristik
</b>,
bukan probabilitas AI.
</p>
</div>
<!-- ============================================= -->
<!-- RESULT -->
<!-- ============================================= -->
<div class="box">
<h2>
Hasil Deteksi
</h2>
<img
src="
{encode_jpg(
data["result"]
)}
"
style="
width:100%;
max-width:1280px;
border:2px solid #333
"
>
</div>
<!-- ============================================= -->
<!-- MASK -->
<!-- ============================================= -->
<div class="box">
<h2>
Mask / Threshold
</h2>
<div class="grid">
{
''.join(cards)
}
</div>
</div>
<!-- ============================================= -->
<!-- PARAMETERS -->
<!-- ============================================= -->
<div class="box">
<h2>
Parameter CV-04
</h2>
<ul>
<li>
Area ratio:
{MIN_AREA_RATIO}
—
{MAX_AREA_RATIO}
</li>
<li>
Minimum side ratio:
{MIN_SIDE_RATIO}
</li>
<li>
Maximum aspect ratio:
{MAX_ASPECT_RATIO}
</li>
<li>
Minimum solidity:
{MIN_SOLIDITY}
</li>
<li>
Minimum extent:
{MIN_EXTENT}
</li>
<li>
Minimum circularity:
{MIN_CIRCULARITY}
</li>
<li>
White HSV:
{
data[
"threshold_info"
][
"white_hsv"
]
}
</li>
<li>
Brown HSV:
{
data[
"threshold_info"
][
"brown_hsv"
]
}
</li>
</ul>
</div>
<!-- ============================================= -->
<!-- DETAIL -->
<!-- ============================================= -->
<div class="box">
<h2>
Detail Kandidat Telur
</h2>
<table>
<tr>
<th>
ID
</th>
<th>
Mask
</th>
<th>
Ukuran
</th>
<th>
Area
</th>
<th>
Aspect
</th>
<th>
Solidity
</th>
<th>
Circularity
</th>
<th>
Score
</th>
</tr>
{egg_rows}
</table>
</div>
<p
style="
color:#666
"
>
Dashboard refresh
setiap 5 detik.
</p>
</body>
</html>
"""
# ============================================================
# END
# ============================================================
Post a Comment for "CV-04"