Systèmes Temps Réel

# Illustration des concepts de temps réel
# Simulation de contraintes temporelles

import time
import random

class TacheTempsReel:
    """Représente une tâche avec contraintes temps réel"""

    def __init__(self, nom, periode_ms, wcet_ms, deadline_ms, priorite):
        """
        nom: Nom de la tâche
        periode_ms: Période d'activation (ms)
        wcet_ms: Worst Case Execution Time (pire cas)
        deadline_ms: Échéance relative
        priorite: Niveau de priorité (1 = plus haute)
        """
        self.nom = nom
        self.periode = periode_ms
        self.wcet = wcet_ms
        self.deadline = deadline_ms
        self.priorite = priorite
        self.executions = 0
        self.deadlines_ratees = 0

    def executer(self):
        """Simule l'exécution de la tâche"""
        # Temps d'exécution réel (entre 50% et 100% du WCET)
        temps_exec = random.uniform(self.wcet * 0.5, self.wcet)
        self.executions += 1
        return temps_exec

    def afficher_stats(self):
        """Affiche les statistiques de la tâche"""
        taux_succes = ((self.executions - self.deadlines_ratees) /
                       self.executions * 100) if self.executions > 0 else 100
        print(f"  {self.nom}:")
        print(f"    Période: {self.periode}ms | WCET: {self.wcet}ms | Deadline: {self.deadline}ms")
        print(f"    Exécutions: {self.executions} | Deadlines ratées: {self.deadlines_ratees}")
        print(f"    Taux de succès: {taux_succes:.1f}%")

class SystemeTempsReel:
    """Simulateur de système temps réel"""

    TYPES_TR = {
        'hard': {'tolerance': 0, 'description': 'Aucun dépassement toléré'},
        'firm': {'tolerance': 5, 'description': 'Quelques dépassements tolérés'},
        'soft': {'tolerance': 20, 'description': 'Dépassements dégradent la qualité'}
    }

    def __init__(self, type_tr='hard'):
        self.type = type_tr
        self.taches = []
        self.temps_systeme = 0
        self.violations = []

    def ajouter_tache(self, tache):
        """Ajoute une tâche au système"""
        self.taches.append(tache)

    def verifier_ordonnancabilite(self):
        """Vérifie si le système est ordonnancable (test d'utilisation)"""
        # Test d'utilisation CPU : U = Σ(Ci/Ti) ≤ n(2^(1/n) - 1)
        n = len(self.taches)
        if n == 0:
            return True

        utilisation = sum(t.wcet / t.periode for t in self.taches)
        limite_rms = n * (2 ** (1/n) - 1)  # Limite pour RMS

        print(f"\n📊 ANALYSE D'ORDONNANÇABILITÉ")
        print(f"   Nombre de tâches : {n}")
        print(f"   Utilisation CPU  : {utilisation*100:.1f}%")
        print(f"   Limite RMS       : {limite_rms*100:.1f}%")

        if utilisation <= limite_rms:
            print(f"   ✅ Système ORDONNANÇABLE (U ≤ limite)")
            return True
        elif utilisation <= 1.0:
            print(f"   ⚠️ Ordonnançabilité NON GARANTIE (limite < U ≤ 100%)")
            return None
        else:
            print(f"   ❌ Système NON ORDONNANÇABLE (U > 100%)")
            return False

    def simuler_execution(self, duree_ms, afficher_trace=True):
        """Simule l'exécution du système pendant une durée donnée"""
        print(f"\n🔄 SIMULATION ({duree_ms}ms) - Type: {self.type.upper()}")
        print("-" * 50)

        # Trier par priorité
        taches_triees = sorted(self.taches, key=lambda t: t.priorite)

        # Prochaine activation de chaque tâche
        prochaine_activation = {t.nom: 0 for t in self.taches}
        deadline_courante = {t.nom: t.deadline for t in self.taches}

        t = 0
        while t < duree_ms:
            # Trouver les tâches prêtes à s'exécuter
            taches_pretes = [
                tache for tache in taches_triees
                if prochaine_activation[tache.nom] <= t
            ]

            if taches_pretes:
                # Exécuter la tâche de plus haute priorité
                tache = taches_pretes[0]
                temps_exec = tache.executer()

                # Vérifier la deadline
                temps_fin = t + temps_exec
                deadline = prochaine_activation[tache.nom] + tache.deadline

                if temps_fin > deadline:
                    tache.deadlines_ratees += 1
                    self.violations.append({
                        'tache': tache.nom,
                        'temps': t,
                        'depassement': temps_fin - deadline
                    })
                    symbole = "❌"
                else:
                    symbole = "✅"

                if afficher_trace and t < 200:  # Limiter l'affichage
                    print(f"  t={t:4d}ms: {tache.nom:15s} exec={temps_exec:.1f}ms {symbole}")

                # Planifier la prochaine activation
                prochaine_activation[tache.nom] += tache.periode

                t = int(temps_fin)
            else:
                t += 1  # Idle

        return self.violations

    def rapport(self):
        """Génère un rapport du système"""
        print("\n" + "="*60)
        print("   RAPPORT SYSTÈME TEMPS RÉEL")
        print("="*60)

        print(f"\nType de système : {self.type.upper()}")
        print(f"Tolérance : {self.TYPES_TR[self.type]['description']}")

        print(f"\n📋 TÂCHES ({len(self.taches)}) :")
        for tache in sorted(self.taches, key=lambda t: t.priorite):
            tache.afficher_stats()

        print(f"\n🚨 VIOLATIONS : {len(self.violations)}")
        if self.violations:
            for v in self.violations[:5]:  # Afficher les 5 premières
                print(f"  - {v['tache']} à t={v['temps']}ms (dépassement: {v['depassement']:.1f}ms)")

        # Verdict
        tolerance = self.TYPES_TR[self.type]['tolerance']
        total_exec = sum(t.executions for t in self.taches)
        taux_violation = len(self.violations) / total_exec * 100 if total_exec > 0 else 0

        print(f"\n📊 VERDICT :")
        if taux_violation <= tolerance:
            print(f"  ✅ Système CONFORME (violations: {taux_violation:.1f}% ≤ {tolerance}%)")
        else:
            print(f"  ❌ Système NON CONFORME (violations: {taux_violation:.1f}% > {tolerance}%)")

# Démonstration
print("="*60)
print("   SIMULATION SYSTÈME TEMPS RÉEL")
print("="*60)

# Créer un système temps réel dur
systeme = SystemeTempsReel(type_tr='hard')

# Ajouter des tâches (inspirées d'un système automobile)
print("\n📝 CONFIGURATION DES TÂCHES")
print("-"*50)

taches = [
    TacheTempsReel("Airbag_Check", periode_ms=10, wcet_ms=2, deadline_ms=10, priorite=1),
    TacheTempsReel("ABS_Control", periode_ms=20, wcet_ms=5, deadline_ms=15, priorite=2),
    TacheTempsReel("Engine_Mgmt", periode_ms=50, wcet_ms=10, deadline_ms=40, priorite=3),
    TacheTempsReel("Dashboard", periode_ms=100, wcet_ms=15, deadline_ms=100, priorite=4),
]

for t in taches:
    systeme.ajouter_tache(t)
    print(f"  + {t.nom} (P={t.priorite}, T={t.periode}ms, C={t.wcet}ms, D={t.deadline}ms)")

# Vérifier l'ordonnançabilité
systeme.verifier_ordonnancabilite()

# Simuler
systeme.simuler_execution(duree_ms=500, afficher_trace=True)

# Rapport final
systeme.rapport()