File size: 4,971 Bytes
59d73a5 |
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 |
import gradio as gr
from collections import defaultdict
import cv2
import numpy as np
import tempfile
from ultralytics import YOLO
import os
import telegram # Import de la bibliothèque
import asyncio # Nécessaire pour les opérations asynchrones
# --- CONFIGURATION TELEGRAM ---
# Récupère les identifiants depuis les variables d'environnement
TELEGRAM_BOT_TOKEN = "8489885905:AAFoNtXgszbL28yX8ogvQdL6lBHXa9lwHHI"
TELEGRAM_CHAT_ID = "-1002689269933"
# 1. Charger le modèle une seule fois au démarrage de l'application
print("Chargement du modèle YOLOv8...")
model = YOLO("yolov11n.pt")
print("Modèle chargé.")
# 2. Fonction de traitement vidéo (inchangée)
def track_objects(video_in_path):
cap = cv2.VideoCapture(video_in_path)
if not cap.isOpened():
raise gr.Error("Erreur : Impossible d'ouvrir la vidéo d'entrée.")
width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
fps = cap.get(cv2.CAP_PROP_FPS)
video_out_path = tempfile.NamedTemporaryFile(suffix=".mp4", delete=False).name
fourcc = cv2.VideoWriter_fourcc(*'mp4v')
out = cv2.VideoWriter(video_out_path, fourcc, fps, (width, height))
track_history = defaultdict(lambda: [])
total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
print(f"Début du traitement de {total_frames} images.")
while cap.isOpened():
success, frame = cap.read()
if not success:
break
results = model.track(frame, persist=True)
annotated_frame = results[0].plot() # .plot() gère déjà le cas où il n'y a pas de détection
if results[0].boxes is not None and results[0].boxes.id is not None:
boxes = results[0].boxes.xywh.cpu()
track_ids = results[0].boxes.id.int().cpu().tolist()
for box, track_id in zip(boxes, track_ids):
x, y, w, h = box
track = track_history[track_id]
track.append((float(x), float(y)))
if len(track) > 30:
track.pop(0)
points = np.hstack(track).astype(np.int32).reshape((-1, 1, 2))
cv2.polylines(annotated_frame, [points], isClosed=False, color=(230, 230, 230), thickness=3)
out.write(annotated_frame)
cap.release()
out.release()
print("Traitement vidéo terminé.")
return video_out_path
# 3. Nouvelle fonction pour envoyer la vidéo sur Telegram
async def send_to_telegram(video_path):
if not TELEGRAM_BOT_TOKEN or not TELEGRAM_CHAT_ID:
error_msg = "Erreur : Les variables d'environnement TELEGRAM_BOT_TOKEN et TELEGRAM_CHAT_ID ne sont pas configurées."
print(error_msg)
return error_msg
try:
print("Initialisation du bot Telegram...")
bot = telegram.Bot(token=TELEGRAM_BOT_TOKEN)
print(f"Envoi de la vidéo {video_path} au chat ID {TELEGRAM_CHAT_ID}...")
# Envoi de la vidéo
async with bot:
with open(video_path, 'rb') as video_file:
await bot.send_video(
chat_id=TELEGRAM_CHAT_ID,
video=video_file,
caption="Voici la vidéo traitée avec le suivi d'objets YOLOv8."
)
success_msg = "Vidéo envoyée avec succès sur Telegram !"
print(success_msg)
return success_msg
except Exception as e:
error_msg = f"Une erreur est survenue lors de l'envoi sur Telegram : {e}"
print(error_msg)
return error_msg
# 4. Fonction principale qui orchestre le tout pour Gradio
async def process_and_send(video_in_path):
if video_in_path is None:
raise gr.Error("Veuillez d'abord uploader une vidéo.")
# Étape 1: Traiter la vidéo
gr.Info("Le traitement de la vidéo a commencé...")
processed_video_path = track_objects(video_in_path)
# Étape 2: Envoyer sur Telegram
gr.Info("Traitement terminé. Envoi sur Telegram...")
telegram_status = await send_to_telegram(processed_video_path)
# Étape 3: Renvoyer les résultats à l'interface Gradio
return processed_video_path, telegram_status
# 5. Créer l'interface Gradio
with gr.Blocks() as demo:
gr.Markdown("# Démonstration de Suivi d'Objets avec YOLOv8 et envoi sur Telegram")
gr.Markdown("Uploadez une vidéo. Une fois le traitement terminé, le résultat s'affichera ici et sera envoyé sur votre chat Telegram.")
with gr.Row():
video_input = gr.Video(label="Vidéo d'entrée")
video_output = gr.Video(label="Vidéo avec suivi")
status_textbox = gr.Textbox(label="Statut de l'envoi Telegram", interactive=False)
btn = gr.Button("Lancer le suivi et Envoyer sur Telegram")
btn.click(
fn=process_and_send,
inputs=video_input,
outputs=[video_output, status_textbox]
)
# Lancer l'application
if __name__ == "__main__":
demo.launch() |