First upload

1DCNN_Transformer_L-dim256_train8_1405_checkpoint.weights.h5

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:0b6db79af3b518150cea916c743d0d9e4cc8eb8e58aea4bec3b52dd714c6d2a3
+size 33827128

Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.12
+
+WORKDIR /code
+
+COPY * /code
+
+RUN pip install --no-cache-dir -r /app/requirement.txt
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

app.py

@@ -0,0 +1,40 @@
+from fastapi import FastAPI, HTTPException
+import random
+from pydantic import BaseModel
+from typing import List
+from utils import decode_image
+from mediapipe_preprocess import mediapipe_process
+from model_predict import model_predict
+import numpy as np
+
+class Item(BaseModel):
+    text: str
+    number: int | None = None
+
+class RecordFrames(BaseModel):
+    images: List[str]
+
+app = FastAPI()
+
+@app.get("/")
+def return_hello():
+    return {"text":"Hello from server"}
+
+@app.get("/random")
+def return_random():
+    return {"random number":int(random.random()*1000)}
+
+@app.post("/receive")
+def return_received(item: Item):
+    return {"text":item.text,
+            "number":item.number}
+
+@app.post("/predict")
+def return_predict(record: RecordFrames):
+    if not record.images:
+        raise HTTPException(status_code=400, detail="No images provided")
+    
+    frames = [np.array(decode_image(img)) for img in record.images] 
+    keypoints = mediapipe_process(frames)
+    label = model_predict(keypoints)
+    return {"label":label}

label_list.json

@@ -0,0 +1 @@
+{"0": "an", "1": "ban", "2": "ban dem", "3": "ban ngay", "4": "bo", "5": "cam on", "6": "choi", "7": "cuoi", "8": "di", "9": "di hoc", "10": "khoc", "11": "lam viec", "12": "me", "13": "moi ngay", "14": "sach", "15": "toi", "16": "viet", "17": "xem", "18": "xin chao", "19": "xin loi"}

mediapipe_preprocess.py

@@ -0,0 +1,118 @@
+import mediapipe as mp
+import numpy as np
+import cv2
+import copy
+
+mp_holistic = mp.solutions.holistic 
+mp_drawing = mp.solutions.drawing_utils
+width, height = 640, 480
+
+model = mp_holistic.Holistic(min_detection_confidence=0.5, min_tracking_confidence=0.5)
+
+def mediapipe_detection(image):
+    # từ image, model dự đoán trả về kết quả (định dạng mặc định)
+    results = model.process(image)
+    return results
+
+def extract_keypoint(results,last):
+    res = []
+    if results.pose_landmarks:
+        for p in results.pose_landmarks.landmark:
+            res.append(np.array([p.x,p.y,p.z,p.visibility]))
+    else:
+        for _ in range(33):
+            res.append(np.array([0,0,0,0]))
+    #--------------
+    if results.left_hand_landmarks:
+        for p in results.left_hand_landmarks.landmark:
+            res.append(np.array([p.x,p.y,p.z]))
+    elif last!= None and last.left_hand_landmarks:
+        for p in last.left_hand_landmarks.landmark:
+            res.append(np.array([p.x,p.y,p.z]))
+    else:
+        for _ in range(21):
+            res.append(np.array([0,0,0]))
+    #---------------
+    if results.right_hand_landmarks:
+        for p in results.right_hand_landmarks.landmark:
+            res.append(np.array([p.x,p.y,p.z]))
+    elif last!=None and last.right_hand_landmarks:
+        for p in last.right_hand_landmarks.landmark:
+            res.append(np.array([p.x,p.y,p.z]))
+    else:
+        for _ in range(21):
+            res.append(np.array([0,0,0]))
+    return res
+
+def normalize_keypoint(res,img=None):
+    #normalize keypoint
+    x1,y1,x2,y2 = res[11][0]*width,res[11][1]*height,res[12][0]*width,res[12][1]*height
+    try:
+        cv2.circle(img,(int(x1),int(y1)),4,(0,255,255),-1)
+        cv2.circle(img,(int(x2),int(y2)),4,(0,255,255),-1)
+    except:
+        # print("No img found")
+        pass
+    dis = np.sqrt((x1-x2)**2+(y1-y2)**2)
+    x_cen = (res[11][0]+res[12][0])/2
+    y_cen = (res[11][1]+res[12][1])/2
+    vector = [0.5-x_cen,0.5-y_cen]
+    scale = (200*width/640)/dis
+    for i in range(len(res)):
+        if res[i][0]==0 and res[i][1]==0:
+            continue
+        res[i][0] = vector[0]+res[i][0]
+        res[i][1] = vector[1]+res[i][1]
+        res[i][0] = 0.5+(res[i][0]-0.5)*scale
+        res[i][1] = 0.5+(res[i][1]-0.5)*scale
+    return res
+
+def update_mpresult(res,results,last):
+    c = 0
+    if results.pose_landmarks:
+        for p in results.pose_landmarks.landmark:
+            p.x = res[c][0]
+            p.y = res[c][1]
+            if(c==20 and p.y>1.1 and last): last.right_hand_landmarks = None
+            elif(c==19 and p.y>1.1 and last): last.left_hand_landmarks = None
+            c+=1
+    else:
+        for _ in range(33):
+            c+=1
+    if results.left_hand_landmarks:
+        for p in results.left_hand_landmarks.landmark:
+            p.x = res[c][0]
+            p.y = res[c][1]
+            c+=1
+    else:
+        if last!=None and last.left_hand_landmarks: results.left_hand_landmarks = copy.deepcopy(last.left_hand_landmarks)
+        for _ in range(21):
+            c+=1
+    if results.right_hand_landmarks:
+        for p in results.right_hand_landmarks.landmark:
+            p.x = res[c][0]
+            p.y = res[c][1]
+            c+=1
+    else:
+        if last!=None and last.right_hand_landmarks: results.right_hand_landmarks = copy.deepcopy(last.right_hand_landmarks)
+        for _ in range(21):
+            c+=1
+    return results
+
+def extract_keypoints_flatten(result, last, img=None):
+    #đây là hàm chính thức
+    res = extract_keypoint(result, last)
+    res = normalize_keypoint(res,img)
+    update_mpresult(res,result,last)
+    return np.concatenate([x for x in res])
+
+def mediapipe_process(frames):
+    """Main function to call, process a batch of frames into numpy array for prediction"""
+    sequence = []
+    last = None
+    for frame in frames:
+        results = mediapipe_detection(frame)
+        keypoints = extract_keypoints_flatten(results, last)
+        last = copy.deepcopy(results)
+        sequence.append(keypoints)
+    return np.array(sequence)

model.py

@@ -0,0 +1,301 @@
+# --------- Define auxiliary classes ---------
+
+import os
+import keras
+import tensorflow as tf
+
+@keras.saving.register_keras_serializable(package="1DCNN_Transformer")
+class ECA(tf.keras.layers.Layer):
+    def __init__(self, kernel_size=5, **kwargs):
+        super().__init__(**kwargs)
+        self.supports_masking = True
+        self.kernel_size = kernel_size
+        self.conv = tf.keras.layers.Conv1D(1, kernel_size=kernel_size, strides=1, padding="same", use_bias=False)
+
+    def call(self, inputs, mask=None):
+        nn = tf.keras.layers.GlobalAveragePooling1D()(inputs, mask=mask)
+        nn = tf.expand_dims(nn, -1)
+        nn = self.conv(nn)
+        nn = tf.squeeze(nn, -1)
+        nn = tf.nn.sigmoid(nn)
+        nn = nn[:,None,:]
+        return inputs * nn
+
+    def get_config(self):
+        base_config = super().get_config()
+        config = {
+            # "supports_masking" : keras.saving.serialize_keras_object(self.supports_masking),
+            "kernel_size" : keras.saving.serialize_keras_object(self.kernel_size)
+        }
+        return {**base_config, **config}
+
+    @classmethod
+    def from_config(cls,config):
+        kernel_size_config = config.pop("kernel_size")
+        kernel_size = keras.saving.deserialize_keras_object(kernel_size_config)
+        return cls(kernel_size, **config)
+
+@keras.saving.register_keras_serializable(package="1DCNN_Transformer")
+class LateDropout(tf.keras.layers.Layer):
+    def __init__(self, rate, noise_shape=None, start_step=0, **kwargs):
+        super().__init__(**kwargs)
+        self.supports_masking = True
+        self.rate = rate
+        self.noise_shape = noise_shape
+        self.start_step = start_step
+        self.dropout = tf.keras.layers.Dropout(rate, noise_shape=noise_shape)
+
+    def build(self, input_shape):
+        super().build(input_shape)
+        agg = tf.VariableAggregation.ONLY_FIRST_REPLICA
+        self._train_counter = tf.Variable(0, dtype="int64", aggregation=agg, trainable=False)
+
+    def call(self, inputs, training=False):
+        x = tf.cond(self._train_counter < self.start_step, lambda:inputs, lambda:self.dropout(inputs, training=training))
+        if training:
+            self._train_counter.assign_add(1)
+        return x
+
+    def get_config(self):
+        base_config = super().get_config()
+        config = {
+            # "supports_masking" : keras.saving.serialize_keras_object(self.supports_masking),
+            "rate" : keras.saving.serialize_keras_object(self.rate),
+            "start_step" : keras.saving.serialize_keras_object(self.start_step),
+            "noise_shape" : keras.saving.serialize_keras_object(self.noise_shape),
+        }
+        return {**base_config, **config}
+
+    @classmethod
+    def from_config(cls,config):
+        rate_config = config.pop("rate")
+        rate = keras.saving.deserialize_keras_object(rate_config)
+        start_step_config = config.pop("start_step")
+        start_step = keras.saving.deserialize_keras_object(start_step_config)
+        noise_shape_config = config.pop("noise_shape")
+        noise_shape = keras.saving.deserialize_keras_object(noise_shape_config)
+        return cls(rate, noise_shape, start_step,  **config)
+
+@keras.saving.register_keras_serializable(package="1DCNN_Transformer")
+class CausalDWConv1D(tf.keras.layers.Layer):
+    def __init__(self,
+        kernel_size=17,
+        dilation_rate=1,
+        use_bias=False,
+        depthwise_initializer='glorot_uniform',
+        name='', **kwargs):
+        super().__init__(name=name,**kwargs)
+        self.kernel_size = kernel_size
+        self.dilation_rate = dilation_rate
+        self.use_bias = use_bias
+        self.depthwise_initializer=depthwise_initializer
+        self.lname=name
+
+        self.causal_pad = tf.keras.layers.ZeroPadding1D((dilation_rate*(kernel_size-1),0),name=name + '_pad')
+        self.dw_conv = tf.keras.layers.DepthwiseConv1D(
+                            kernel_size,
+                            strides=1,
+                            dilation_rate=dilation_rate,
+                            padding='valid',
+                            use_bias=use_bias,
+                            depthwise_initializer=depthwise_initializer,
+                            name=name + '_dwconv')
+        self.supports_masking = True
+
+    def call(self, inputs):
+        x = self.causal_pad(inputs)
+        x = self.dw_conv(x)
+        return x
+
+    def get_config(self):
+        base_config = super().get_config()
+        config = {
+            "kernel_size" : keras.saving.serialize_keras_object(self.kernel_size),
+            "dilation_rate" : keras.saving.serialize_keras_object(self.dilation_rate),
+            "use_bias" : keras.saving.serialize_keras_object(self.use_bias),
+            "depthwise_initializer" : keras.saving.serialize_keras_object(self.depthwise_initializer),
+            "name" : keras.saving.serialize_keras_object(self.lname),
+        }
+        return {**base_config, **config}
+
+    @classmethod
+    def from_config(cls,config):
+        kernel_size_config = config.pop("kernel_size")
+        kernel_size = keras.saving.deserialize_keras_object(kernel_size_config)
+        dilation_rate_config = config.pop("dilation_rate")
+        dilation_rate = keras.saving.deserialize_keras_object(dilation_rate_config)
+        bias_config = config.pop("use_bias")
+        bias = keras.saving.deserialize_keras_object(bias_config)
+        depthwise_config = config.pop("depthwise_initializer")
+        depthwise = keras.saving.deserialize_keras_object(depthwise_config)
+        name_config = config.pop("name")
+        name = keras.saving.deserialize_keras_object(name_config)
+
+        return cls(kernel_size,dilation_rate,bias,depthwise,name, **config)
+
+def Conv1DBlock(channel_size,
+          kernel_size,
+          dilation_rate=1,
+          drop_rate=0.0,
+          expand_ratio=2,
+          se_ratio=0.25,
+          activation='swish',
+          name=None):
+    '''
+    efficient conv1d block, @hoyso48
+    '''
+    if name is None:
+        name = str(tf.keras.backend.get_uid("mbblock"))
+    # Expansion phase
+    def apply(inputs):
+        channels_in = tf.keras.backend.int_shape(inputs)[-1]
+        channels_expand = channels_in * expand_ratio
+
+        skip = inputs
+
+        x = tf.keras.layers.Dense(
+            channels_expand,
+            use_bias=True,
+            activation=activation,
+            name=name + '_expand_conv')(inputs)
+
+        # Depthwise Convolution
+        x = CausalDWConv1D(kernel_size,
+            dilation_rate=dilation_rate,
+            use_bias=False,
+            name=name + '_dwconv')(x)
+
+        x = tf.keras.layers.BatchNormalization(momentum=0.95, name=name + '_bn')(x)
+
+        x  = ECA()(x)
+
+        x = tf.keras.layers.Dense(
+            channel_size,
+            use_bias=True,
+            name=name + '_project_conv')(x)
+
+        if drop_rate > 0:
+            x = tf.keras.layers.Dropout(drop_rate, noise_shape=(None,1,1), name=name + '_drop')(x)
+
+        if (channels_in == channel_size):
+            x = tf.keras.layers.add([x, skip], name=name + '_add')
+        return x
+
+    return apply
+
+
+@keras.saving.register_keras_serializable(package="1DCNN_Transformer")
+class MultiHeadSelfAttention(tf.keras.layers.Layer):
+    def __init__(self, dim=256, num_heads=4, dropout=0, **kwargs):
+        super().__init__(**kwargs)
+        self.dim = dim
+        self.scale = self.dim ** -0.5
+        self.num_heads = num_heads
+        self.dropout = dropout
+        self.qkv = tf.keras.layers.Dense(3 * dim, use_bias=False)
+        self.drop1 = tf.keras.layers.Dropout(dropout)
+        self.proj = tf.keras.layers.Dense(dim, use_bias=False)
+        self.supports_masking = True
+
+    def call(self, inputs, mask=None):
+        qkv = self.qkv(inputs)
+        qkv = tf.keras.layers.Permute((2, 1, 3))(tf.keras.layers.Reshape((-1, self.num_heads, self.dim * 3 // self.num_heads))(qkv))
+        q, k, v = tf.split(qkv, [self.dim // self.num_heads] * 3, axis=-1)
+
+        attn = tf.matmul(q, k, transpose_b=True) * self.scale
+
+        if mask is not None:
+            mask = mask[:, None, None, :]
+
+        attn = tf.keras.layers.Softmax(axis=-1)(attn, mask=mask)
+        attn = self.drop1(attn)
+
+        x = attn @ v
+        x = tf.keras.layers.Reshape((-1, self.dim))(tf.keras.layers.Permute((2, 1, 3))(x))
+        x = self.proj(x)
+        return x
+
+    def get_config(self):
+        base_config = super().get_config()
+        config = {
+            "dim" : self.dim,
+            "num_heads" : self.num_heads,
+            "dropout" : self.dropout,
+        }
+        return {**base_config, **config}
+
+    @classmethod
+    def from_config(cls,config):
+        dim_config = config.pop("dim")
+        dim = keras.saving.deserialize_keras_object(dim_config)
+        num_heads_config = config.pop("num_heads")
+        num_heads = keras.saving.deserialize_keras_object(num_heads_config)
+        dropout_config = config.pop("dropout")
+        dropout = keras.saving.deserialize_keras_object(dropout_config)
+        return cls(dim,num_heads,dropout)
+
+def TransformerBlock(dim=256, num_heads=4, expand=4, attn_dropout=0.2, drop_rate=0.2, activation='swish'):
+    def apply(inputs):
+        x = inputs
+        x = tf.keras.layers.BatchNormalization(momentum=0.95)(x)
+        x = MultiHeadSelfAttention(dim=dim,num_heads=num_heads,dropout=attn_dropout)(x)
+        x = tf.keras.layers.Dropout(drop_rate, noise_shape=(None,1,1))(x)
+        x = tf.keras.layers.Add()([inputs, x])
+        attn_out = x
+
+        x = tf.keras.layers.BatchNormalization(momentum=0.95)(x)
+        x = tf.keras.layers.Dense(dim*expand, use_bias=False, activation=activation)(x)
+        x = tf.keras.layers.Dense(dim, use_bias=False)(x)
+        x = tf.keras.layers.Dropout(drop_rate, noise_shape=(None,1,1))(x)
+        x = tf.keras.layers.Add()([attn_out, x])
+        return x
+    return apply
+
+MAX_LEN = 30 # number of frame
+CHANNELS = 258 # number of keypoint value
+NUM_CLASSES = 20
+PAD = -100
+
+# ----------------------------------------- DEFINE MODEL -----------------------------
+def get_model(max_len=MAX_LEN, dropout_step=0, dim=256):
+    inp = tf.keras.Input((max_len,CHANNELS))
+    # x = tf.keras.layers.Masking(mask_value=PAD,input_shape=(max_len,CHANNELS))(inp) #we don't need masking layer with inference
+    x = inp
+    ksize = 3
+    x = tf.keras.layers.Permute((2,1))(x)
+    x = tf.keras.layers.Dense(dim, use_bias=False,name='stem_conv')(x)
+    x = tf.keras.layers.BatchNormalization(momentum=0.95,name='stem_bn')(x)
+
+    x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+    x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+    x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+    x = TransformerBlock(dim,expand=2)(x)
+
+    x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+    x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+    x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+    x = TransformerBlock(dim,expand=2)(x)
+
+    if dim == 384: #for the 4x sized model
+        x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+        x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+        x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+        x = TransformerBlock(dim,expand=2)(x)
+
+        x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+        x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+        x = Conv1DBlock(dim,ksize,drop_rate=0.2)(x)
+        x = TransformerBlock(dim,expand=2)(x)
+
+    x = tf.keras.layers.Dense(dim*2,activation=None,name='top_conv')(x)
+    x = tf.keras.layers.GlobalAveragePooling1D()(x)
+    # x = LateDropout(0.5, start_step=dropout_step)(x)
+    x = tf.keras.layers.Dense(NUM_CLASSES,name='classifier',activation="softmax")(x)
+    return tf.keras.Model(inp, x)
+
+def load_model(path='1DCNN_Transformer_L-dim256_train8_1405_checkpoint.weights.h5'):
+    model = get_model()
+    module_dir = os.path.dirname(os.path.abspath(__file__))
+    model_path = os.path.join(module_dir,path)
+    model.load_weights(model_path)
+    return model

model_predict.py

@@ -0,0 +1,14 @@
+from model import load_model
+import json
+import numpy as np
+
+classifier = load_model()
+with open("label_list.json","r") as infile:
+    actions = list(json.load(infile).values())
+
+def model_predict(input):
+    """Perform prediction on input (numpy array), return a label (str)"""
+    
+    res = classifier.predict(np.expand_dims(input,axis=0))[0]
+    label = actions[np.argmax(res)]
+    return label

requirement.txt

@@ -0,0 +1,8 @@
+keras>=3.3.3
+tensorflow>=2.16.1
+opencv-python>=4.9.0.80
+request>=2.31.0
+mediapipe>=0.10.14
+uvicorn>=0.29.0
+numpy>=1.26.4
+pillow>=10.3.0

utils.py

@@ -0,0 +1,14 @@
+import base64
+from PIL import Image
+import io
+
+def decode_image(encoded_image):
+    """Decode the base64 string to bytes"""
+    image_data = base64.b64decode(encoded_image)
+    # Open the image using PIL
+    image = Image.open(io.BytesIO(image_data))
+    return image
+
+def encode_image(buffer):
+    encoded_image = base64.b64encode(buffer).decode('utf-8')
+    return encoded_image