File size: 6,945 Bytes
77cd16c
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
import io
import streamlit as st
import cv2
import numpy as np
import tensorflow as tf
import mediapipe as mp
import tempfile
import os
import time
from tensorflow.keras.models import load_model # type: ignore
from gtts import gTTS
from playsound import playsound

# ==============================
# KONFIGURASI DASAR
# ==============================
BASE_DIR = os.getcwd()
MODEL_PATH = os.path.join(BASE_DIR, "models", "sign_model.h5")
LABEL_CLASSES_PATH = os.path.join(BASE_DIR, "models", "label_classes.npy")
TEMP_AUDIO_FILE = os.path.join(tempfile.gettempdir(), "temp_prediction.mp3")

# ==============================
# LOAD MODEL
# ==============================
@st.cache_resource
def load_all_models():
    model = load_model(MODEL_PATH)
    mobilenet_model = tf.keras.applications.MobileNetV2(
        input_shape=(224, 224, 3),
        include_top=False,
        weights='imagenet',
        pooling='avg'
    )
    actions = np.load(LABEL_CLASSES_PATH)
    return model, mobilenet_model, actions

model, mobilenet_model, actions = load_all_models()

# ==============================
# MEDIAPIPE SETUP
# ==============================
mp_holistic = mp.solutions.holistic
mp_drawing = mp.solutions.drawing_utils
holistic = mp_holistic.Holistic(min_detection_confidence=0.5, min_tracking_confidence=0.5)

# ==============================
# FUNGSI PENDUKUNG
# ==============================
def mediapipe_detection(image, model):
    image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB)
    image.flags.writeable = False
    results = model.process(image)
    image.flags.writeable = True
    return cv2.cvtColor(image, cv2.COLOR_RGB2BGR), results

def draw_styled_landmarks(image, results):
    if results.pose_landmarks:
        mp_drawing.draw_landmarks(image, results.pose_landmarks, mp_holistic.POSE_CONNECTIONS)
    if results.left_hand_landmarks:
        mp_drawing.draw_landmarks(image, results.left_hand_landmarks, mp_holistic.HAND_CONNECTIONS)
    if results.right_hand_landmarks:
        mp_drawing.draw_landmarks(image, results.right_hand_landmarks, mp_holistic.HAND_CONNECTIONS)

def extract_landmarks(results):
    pose = np.array([[res.x, res.y, res.z, res.visibility] for res in results.pose_landmarks.landmark]).flatten() if results.pose_landmarks else np.zeros(33*4)
    lh = np.array([[res.x, res.y, res.z] for res in results.left_hand_landmarks.landmark]).flatten() if results.left_hand_landmarks else np.zeros(21*3)
    rh = np.array([[res.x, res.y, res.z] for res in results.right_hand_landmarks.landmark]).flatten() if results.right_hand_landmarks else np.zeros(21*3)
    return np.concatenate([pose, lh, rh])

def get_bbox(results, shape):
    xs, ys = [], []
    for lm_set in [results.pose_landmarks, results.left_hand_landmarks, results.right_hand_landmarks]:
        if lm_set:
            for lm in lm_set.landmark:
                xs.append(int(lm.x * shape[1]))
                ys.append(int(lm.y * shape[0]))
    if not xs or not ys:
        return None
    return max(0, min(xs)), max(0, min(ys)), min(shape[1], max(xs)), min(shape[0], max(ys))

def create_canvas_crop(img, bbox):
    CANVAS_SIZE = 600
    if bbox is None:
        return np.ones((CANVAS_SIZE, CANVAS_SIZE, 3), dtype=np.uint8) * 255
    x1, y1, x2, y2 = bbox
    roi = img[y1:y2, x1:x2]
    if roi.size == 0:
        return np.ones((CANVAS_SIZE, CANVAS_SIZE, 3), dtype=np.uint8) * 255
    h, w = roi.shape[:2]
    scale = min((CANVAS_SIZE*0.9)/w, (CANVAS_SIZE*0.9)/h)
    new_w, new_h = int(w*scale), int(h*scale)
    resized = cv2.resize(roi, (new_w, new_h))
    canvas = np.ones((CANVAS_SIZE, CANVAS_SIZE, 3), dtype=np.uint8) * 255
    x_offset, y_offset = (CANVAS_SIZE-new_w)//2, (CANVAS_SIZE-new_h)//2
    canvas[y_offset:y_offset+new_h, x_offset:x_offset+new_w] = resized
    return canvas

def text_to_speech(text, lang ='id'):
    mp3_fp = io.BytesIO()
    tts = gTTS(text=text, lang=lang)
    tts.write_to_fp(mp3_fp)
    mp3_fp.seek(0)
    return mp3_fp.read()
# ==============================
# STREAMLIT UI
# ==============================
st.title("🤟 Real-time Sign Language Translator")
st.markdown("Aplikasi ini menerjemahkan bahasa isyarat ke teks dan suara secara real-time menggunakan kamera.")

col1, col2 = st.columns(2)
with col1:
    start_button = st.button("Mulai Deteksi")
with col2:
    stop_button = st.button("Hentikan")

FRAME_WINDOW = st.image([])
sentence_placeholder = st.empty()

# ==============================
# LOOP STREAMING
# ==============================
sequence = []
sentence = []
threshold = 0.9
last_prediction_time = 0
COOLDOWN = 2

if start_button:
    cap = cv2.VideoCapture(0)
    st.info("Kamera aktif. Tekan 'Hentikan' untuk berhenti.")
    while cap.isOpened() and not stop_button:
        ret, frame = cap.read()
        if not ret:
            st.warning("Tidak dapat membaca frame dari kamera.")
            break

        image, results = mediapipe_detection(frame, holistic)
        draw_styled_landmarks(image, results)

        keypoints = extract_landmarks(results)
        if np.any(keypoints != 0):
            bbox = get_bbox(results, frame.shape)
            canvas_crop = create_canvas_crop(frame, bbox)

            img_rgb = cv2.cvtColor(canvas_crop, cv2.COLOR_BGR2RGB)
            resized_img = cv2.resize(img_rgb, (224, 224))
            preprocessed_img = tf.keras.applications.mobilenet_v2.preprocess_input(resized_img)
            mobilenet_features = mobilenet_model.predict(np.expand_dims(preprocessed_img, axis=0), verbose=0).flatten()

            fused_features = np.concatenate([mobilenet_features, keypoints])
            sequence.append(fused_features)
            sequence = sequence[-30:]

            if len(sequence) == 30:
                current_time = time.time()
                if current_time - last_prediction_time > COOLDOWN:
                    res = model.predict(np.expand_dims(sequence, axis=0), verbose=0)[0]
                    if res[np.argmax(res)] > threshold:
                        predicted_label = actions[np.argmax(res)]
                        if len(sentence) == 0 or predicted_label != sentence[-1]:
                            sentence.append(predicted_label)
                            last_prediction_time = current_time
                            try:
                                text_to_speech(predicted_label)
                            except Exception as e:
                                st.warning(f"Voice output error: {e}")

        if len(sentence) > 5:
            sentence = sentence[-5:]

        # Tampilkan hasil dan frame
        FRAME_WINDOW.image(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
        sentence_placeholder.markdown(f"### 🗣️ Prediksi: {' '.join(sentence)}")

    cap.release()
    st.success("Deteksi dihentikan.")