Create app.py

app.py

@@ -0,0 +1,281 @@
+#curl -H "Accept: application/json" https://Og2-wstest.hf.space/
+#curl -H "Accept: application/json" https://Og2-wstest.hf.space/hello/
+#curl -X POST https://Og2-wstest.hf.space/muws2/ -H "Content-Type: application/json" -d "{\"text\": \"Ceci est un texte exemple.\"}"
+#curl -X POST "https://Og2-wstest.hf.space/upload-video/" -H "Content-Type: multipart/form-data" -F "file=@E:\nosave\MyDocs\AndroidStudio\PythonProjects\SportHobbyStats\Foosball\VideoAnnotation\MoviNet2\train\Block_1-2.1/CNFT2Toulouse_Block_1729.avi"
+
+
+from fastapi import FastAPI, File, Form, UploadFile, HTTPException
+from pathlib import Path
+import os
+from pydantic import BaseModel
+import json
+
+app = FastAPI()
+
+# Définir un modèle pour recevoir un texte
+class TextInput(BaseModel):
+    text: str
+
+@app.post("/muws2/")
+def process_text(input_data: TextInput):
+    # Récupérer le texte depuis l'entrée
+    input_text = input_data.text
+    
+    # Créer un dictionnaire JSON avec le texte en sortie
+    output = {
+        "text": input_text
+    }
+    
+    return output
+
+@app.get("/hello/")
+def greet_hello():
+    return {"msg": "Hello World!"}
+
+@app.get("/")
+def greet_hello():
+    return {"msg": "Ok!"}
+
+
+@app.post("/upload-video/")
+async def upload_video(file: UploadFile = File(...)):
+    # Lire le contenu du fichier
+    content = await file.read()
+    # Vous pouvez sauvegarder la vidéo ou la traiter ici
+    with open(f"uploaded_{file.filename}", "wb") as f:
+        f.write(content)
+
+    return {"filename": file.filename, "content_type": file.content_type}
+
+
+
+UPLOAD_DIR = "uploads"
+os.makedirs(UPLOAD_DIR, exist_ok=True)
+
+@app.post("/upload-dropzone/")
+async def upload_file(
+    file: UploadFile = File(...),
+    chunkIndex: int = Form(...),
+    totalChunks: int = Form(...),
+    fileName: str = Form(...),
+    directory: str = Form(...),
+):
+    try:
+        print(f"Received: chunkIndex={chunkIndex}, totalChunks={totalChunks}, fileName={fileName}, directory={directory}")
+        # Create the directory if it doesn't exist
+        target_dir = Path(UPLOAD_DIR) / directory
+        target_dir.mkdir(parents=True, exist_ok=True)
+
+        # Save the chunk
+        chunk_path = target_dir / f"{fileName}.part{chunkIndex}"
+        with open(chunk_path, "wb") as f:
+            f.write(await file.read())
+
+        # If it's the last chunk, reconstruct the file
+        if chunkIndex + 1 == totalChunks:
+            final_file_path = target_dir / fileName
+            with open(final_file_path, "wb") as final_file:
+                for i in range(totalChunks):
+                    part_path = target_dir / f"{fileName}.part{i}"
+                    with open(part_path, "rb") as part_file:
+                        final_file.write(part_file.read())
+                    os.remove(part_path)  # Remove the chunk after merging
+            
+            print(f"Final file path: {final_file_path}")
+
+            return {
+                "status": "success",
+                "message": "Chunk uploaded successfully.",
+                "file_path": str(final_file_path)
+            }
+
+        return {"status": "success", "message": "Chunk uploaded successfully."}
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")
+
+
+
+import tensorflow as tf
+import numpy as np
+import cv2
+import keras
+from keras.saving import register_keras_serializable
+from keras import layers
+from huggingface_hub import hf_hub_download
+from keras.applications.densenet import DenseNet121
+#from tensorflow_docs.vis import embed
+
+# Available backend options are: "jax", "torch", "tensorflow".
+os.environ["KERAS_BACKEND"] = "tensorflow"
+
+
+# Charger le modèle Keras
+MAX_SEQ_LENGTH = 8
+NUM_FEATURES = 1024
+IMG_SIZE = 128
+
+#center_crop_layer = layers.CenterCrop(IMG_SIZE, IMG_SIZE)
+# Au lieu de CenterCrop
+center_crop_layer = layers.Resizing(IMG_SIZE, IMG_SIZE)
+
+
+
+def crop_center(frame):
+    cropped = center_crop_layer(frame[None, ...])
+    cropped = keras.ops.convert_to_numpy(cropped)
+    cropped = keras.ops.squeeze(cropped)
+    return cropped
+
+
+# Following method is modified from this tutorial:
+# https://www.tensorflow.org/hub/tutorials/action_recognition_with_tf_hub
+def load_video(path, max_frames=0, offload_to_cpu=False):
+    cap = cv2.VideoCapture(path)
+    frames = []
+    try:
+        while True:
+            ret, frame = cap.read()
+            if not ret:
+                break
+            frame = frame[:, :, [2, 1, 0]]
+            frame = crop_center(frame)
+            if offload_to_cpu and keras.backend.backend() == "torch":
+                frame = frame.to("cpu")
+            frames.append(frame)
+
+            if len(frames) == max_frames:
+                break
+    finally:
+        cap.release()
+    if offload_to_cpu and keras.backend.backend() == "torch":
+        return np.array([frame.to("cpu").numpy() for frame in frames])
+    return np.array(frames)
+
+def build_feature_extractor():
+    feature_extractor = DenseNet121(
+        weights="imagenet",
+        include_top=False,
+        pooling="avg",
+        input_shape=(IMG_SIZE, IMG_SIZE, 3),
+    )
+    preprocess_input = keras.applications.densenet.preprocess_input
+
+    inputs = keras.Input((IMG_SIZE, IMG_SIZE, 3))
+    preprocessed = preprocess_input(inputs)
+
+    outputs = feature_extractor(preprocessed)
+    return keras.Model(inputs, outputs, name="feature_extractor")
+
+
+feature_extractor = build_feature_extractor()
+
+
+@keras.saving.register_keras_serializable()
+class PositionalEmbedding(layers.Layer):
+    def __init__(self, sequence_length, output_dim, **kwargs):
+        super().__init__(**kwargs)
+        self.position_embeddings = layers.Embedding(
+            input_dim=sequence_length, output_dim=output_dim
+        )
+        self.sequence_length = sequence_length
+        self.output_dim = output_dim
+
+    def build(self, input_shape):
+        self.position_embeddings.build(input_shape)
+
+    def call(self, inputs):
+        # The inputs are of shape: `(batch_size, frames, num_features)`
+        inputs = keras.ops.cast(inputs, self.compute_dtype)
+        length = keras.ops.shape(inputs)[1]
+        positions = keras.ops.arange(start=0, stop=length, step=1)
+        embedded_positions = self.position_embeddings(positions)
+        return inputs + embedded_positions
+
+@keras.saving.register_keras_serializable()
+class TransformerEncoder(layers.Layer):
+    def __init__(self, embed_dim, dense_dim, num_heads, **kwargs):
+        super().__init__(**kwargs)
+        self.embed_dim = embed_dim
+        self.dense_dim = dense_dim
+        self.num_heads = num_heads
+        self.attention = layers.MultiHeadAttention(
+            num_heads=num_heads, key_dim=embed_dim, dropout=0.3
+        )
+        self.dense_proj = keras.Sequential(
+            [
+                layers.Dense(dense_dim, activation=keras.activations.gelu),
+                layers.Dense(embed_dim),
+            ]
+        )
+        self.layernorm_1 = layers.LayerNormalization()
+        self.layernorm_2 = layers.LayerNormalization()
+
+    def call(self, inputs, mask=None):
+        attention_output = self.attention(inputs, inputs, attention_mask=mask)
+        proj_input = self.layernorm_1(inputs + attention_output)
+        proj_output = self.dense_proj(proj_input)
+        return self.layernorm_2(proj_input + proj_output)
+
+
+#model = keras.saving.load_model("hf://Og2/videoclassif")
+model = keras.saving.load_model("hf://Og2/videoclassif", custom_objects={'PositionalEmbedding': PositionalEmbedding, 'TransformerEncoder': TransformerEncoder})
+
+# Identifier le modèle Hugging Face et le fichier que vous voulez lire
+model_repo = "Og2/videoclassif"  # Remplacez par votre modèle spécifique
+file_name = "labels.txt"  # Le fichier que vous voulez télécharger
+
+# Télécharger le fichier depuis Hugging Face Hub
+labels_file_path = hf_hub_download(repo_id=model_repo, filename=file_name)
+
+# Liste des classes du modèle
+# Nom du fichier de classe
+#input_file = "hf://Og2/videoclassif/labels.txt"
+# Lecture du fichier et création de la liste
+with open(labels_file_path, "r") as file:
+    class_labels = [line.strip() for line in file]
+#print("Tableau recréé à partir du fichier :")
+#print(class_labels)
+
+
+# test on video from val dataset
+def prepare_single_video(frames):
+    frame_features = np.zeros(shape=(1, MAX_SEQ_LENGTH, NUM_FEATURES), dtype="float32")
+
+    # Pad shorter videos.
+    if len(frames) < MAX_SEQ_LENGTH:
+        diff = MAX_SEQ_LENGTH - len(frames)
+        padding = np.zeros((diff, IMG_SIZE, IMG_SIZE, 3))
+        frames = np.concatenate(frames, padding)
+
+    frames = frames[None, ...]
+
+    # Extract features from the frames of the current video.
+    for i, batch in enumerate(frames):
+        video_length = batch.shape[0]
+        length = min(MAX_SEQ_LENGTH, video_length)
+        for j in range(length):
+            if np.mean(batch[j, :]) > 0.0:
+                frame_features[i, j, :] = feature_extractor.predict(batch[None, j, :])
+            else:
+                frame_features[i, j, :] = 0.0
+
+    return frame_features
+
+
+async def predict_action(video):
+    print("##### predict_action started #####")
+    frames = load_video(video, offload_to_cpu=True)
+    frame_features = prepare_single_video(frames)
+    probabilities = model.predict(frame_features)[0]
+    # Obtenir le top 5
+    top_5_indices = np.argsort(probabilities)[::-1][:5]
+    results = {class_labels[i]: float(probabilities[i]) for i in top_5_indices}
+    #return results
+
+    # Sauvegarder le JSON dans un fichier temporaire
+    output_file = "result.json"
+    with open(output_file, "w") as f:
+        json.dump(results, f)
+    
+    return results