RTOS/tools/configuration/task_priority_calculator.py

#!/usr/bin/env python3
"""
@file task_priority_calculator.py
@brief Task priority calculator for Rate Monotonic Scheduling
"""

import sys
import argparse
from collections import defaultdict
from math import ceil

class TaskPriorityCalculator:
    """Calculate task priorities using RMS/DMS algorithms"""

    def __init__(self):
        self.tasks = []

    def add_task(self, name, period_ms, execution_time_ms, deadline_ms=None):
        """Add task for analysis"""
        if deadline_ms is None:
            deadline_ms = period_ms

        self.tasks.append({
            'name': name,
            'period': period_ms,
            'execution_time': execution_time_ms,
            'deadline': deadline_ms
        })

    def calculate_utilization(self):
        """Calculate CPU utilization"""
        total_utilization = 0

        for task in self.tasks:
            utilization = task['execution_time'] / task['period']
            task['utilization'] = utilization
            total_utilization += utilization

        return total_utilization

    def rate_monotonic_priority(self):
        """Assign priorities using Rate Monotonic Scheduling"""
        # Sort by period (shortest period = highest priority)
        sorted_tasks = sorted(self.tasks, key=lambda x: x['period'])

        for i, task in enumerate(sorted_tasks):
            task['rms_priority'] = i

        return sorted_tasks

    def deadline_monotonic_priority(self):
        """Assign priorities using Deadline Monotonic Scheduling"""
        # Sort by deadline (shortest deadline = highest priority)
        sorted_tasks = sorted(self.tasks, key=lambda x: x['deadline'])

        for i, task in enumerate(sorted_tasks):
            task['dms_priority'] = i

        return sorted_tasks

    def check_schedulability(self):
        """Check if task set is schedulable"""
        n = len(self.tasks)
        total_utilization = self.calculate_utilization()

        # Liu & Layland bound for RMS
        rms_bound = n * (2 ** (1/n) - 1)

        # Exact schedulability test
        schedulable = True

        for i, task in enumerate(sorted(self.tasks, key=lambda x: x['priority'])):
            # Calculate response time
            response_time = task['execution_time']

            while True:
                interference = 0
                for j in range(i):
                    higher_task = self.tasks[j]
                    interference += ceil(response_time / higher_task['period']) * \
                                   higher_task['execution_time']

                new_response = task['execution_time'] + interference

                if new_response == response_time:
                    break

                response_time = new_response

                if response_time > task['deadline']:
                    schedulable = False
                    break

            task['response_time'] = response_time

            if not schedulable:
                break

        return schedulable, total_utilization, rms_bound

    def generate_report(self):
        """Generate priority assignment report"""
        print("=" * 80)
        print("Task Priority Calculation Report")
        print("=" * 80)
        print()

        # Calculate utilization
        total_utilization = self.calculate_utilization()

        print("Task Set:")
        print("-" * 80)
        print(f"{'Task':20s} {'Period':>8s} {'Exec Time':>10s} {'Deadline':>8s} {'Util':>8s}")
        print("-" * 80)

        for task in self.tasks:
            print(f"{task['name']:20s} {task['period']:8d} "
                  f"{task['execution_time']:10d} {task['deadline']:8d} "
                  f"{task['utilization']*100:7.2f}%")

        print("-" * 80)
        print(f"{'Total':20s} {'':>8s} {'':>10s} {'':>8s} {total_utilization*100:7.2f}%")
        print()

        # Rate Monotonic
        print("Rate Monotonic Priority Assignment:")
        print("-" * 40)
        rms_tasks = self.rate_monotonic_priority()
        for task in rms_tasks:
            print(f"  Priority {task['rms_priority']}: {task['name']}")
        print()

        # Deadline Monotonic
        print("Deadline Monotonic Priority Assignment:")
        print("-" * 40)
        dms_tasks = self.deadline_monotonic_priority()
        for task in dms_tasks:
            print(f"  Priority {task['dms_priority']}: {task['name']}")
        print()

        # Schedulability
        schedulable, utilization, bound = self.check_schedulability()

        print("Schedulability Analysis:")
        print("-" * 40)
        print(f"CPU Utilization: {utilization*100:.2f}%")
        print(f"Theoretical Bound: {bound*100:.2f}%")

        if schedulable:
            print("✓ Task set is schedulable")
        else:
            print("✗ Task set is NOT schedulable")

        print()
        print("Response Times:")
        print("-" * 40)
        for task in self.tasks:
            if 'response_time' in task:
                print(f"  {task['name']}: {task['response_time']}ms "
                      f"(deadline: {task['deadline']}ms)")

def main():
    parser = argparse.ArgumentParser(description='Task priority calculator')
    parser.add_argument('--tasks', nargs='+', help='Task definitions: name:period:exec[:deadline]')
    parser.add_argument('--file', help='Task definition file')

    args = parser.parse_args()

    calculator = TaskPriorityCalculator()

    # Load tasks from file or command line
    if args.file:
        with open(args.file, 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    parts = line.split(':')
                    if len(parts) >= 3:
                        name = parts[0]
                        period = int(parts[1])
                        exec_time = int(parts[2])
                        deadline = int(parts[3]) if len(parts) > 3 else period
                        calculator.add_task(name, period, exec_time, deadline)
    elif args.tasks:
        for task_def in args.tasks:
            parts = task_def.split(':')
            if len(parts) >= 3:
                name = parts[0]
                period = int(parts[1])
                exec_time = int(parts[2])
                deadline = int(parts[3]) if len(parts) > 3 else period
                calculator.add_task(name, period, exec_time, deadline)
    else:
        # Example tasks
        print("Using example task set:")
        calculator.add_task("Engine Control", 1, 0.5)
        calculator.add_task("Brake Control", 5, 1)
        calculator.add_task("CAN Communication", 10, 2)
        calculator.add_task("Sensor Reading", 2, 0.3)
        calculator.add_task("Display Update", 50, 10)

    calculator.generate_report()

if __name__ == '__main__':
    main()