Add real-time analysis functionality and update simulator structure

Exercises/VerySimpleSimulator/src/real_time_analysis.py

@@ -0,0 +1,88 @@
+import pandas as pd
+import math
+
+
+def read_csv(file):
+    """
+    Read tasks from a CSV file and convert to a list of task dictionaries.
+    """
+    df = pd.read_csv(file)
+    req_columns = ['Task', 'BCET', 'WCET', 'Period', 'Deadline', 'Priority']
+
+    if not all(col in df.columns for col in req_columns):
+        raise ValueError(f"Missing required column(s): {req_columns}")
+
+    tasks = []
+    for _, row in df.iterrows():
+        task = {
+            "name": row['Task'],
+            "bcet": row['BCET'],
+            "wcet": row['WCET'],
+            "period": row['Period'],
+            "deadline": row['Deadline'],
+            "priority": row['Priority']
+        }
+        tasks.append(task)
+
+    return tasks
+
+
+def response_time_analysis(tasks):
+    """
+    Perform Response-Time Analysis.
+    """
+    # Sort tasks by priority (lower number means higher priority)
+    sorted_tasks = sorted(tasks, key=lambda x: x['priority'])
+
+    results = []
+
+    for i, task in enumerate(sorted_tasks):
+        R = task['wcet']  # Initial response time
+        last_R = -1
+        schedulable = True
+
+        while R != last_R:
+            last_R = R
+            interference = 0
+
+            # Calculate interference from higher-priority tasks
+            for j in range(i):
+                higher_task = sorted_tasks[j]
+                interference += math.ceil(R / higher_task['period']) * higher_task['wcet']
+
+            R = task['wcet'] + interference
+
+            # Check if the task is schedulable
+            if R > task['deadline']:
+                schedulable = False
+                break
+
+        # Store results
+        result = {
+            'Task': task['name'],
+            'WCRT': R,
+            'Deadline': task['deadline'],
+            'Status': 'Schedulable' if schedulable else 'Not Schedulable'
+        }
+        results.append(result)
+
+    return results
+
+
+def run_rta(path_file):
+    """
+    Main function to read tasks and perform response time analysis.
+    """
+    # Read tasks from CSV
+    tasks = read_csv(path_file)
+    print(tasks)
+    # Perform response time analysis
+    analysis_results = response_time_analysis(tasks)
+
+    # Print results
+    for result in analysis_results:
+        if result['Status'] == 'Schedulable':
+            print(f"Task {result['Task']} is schedulable with WCRT = {result['WCRT']:.1f}")
+        else:
+            print(f"Task {result['Task']} is not schedulable with WCRT.")
+    return analysis_results

Exercises/VerySimpleSimulator/src/very_simple_simulator.py

@@ -1,17 +1,21 @@
 import pandas as pd
 import numpy as np
-from Exercises.VerySimpleSimulator.src.jobs import (
+from jobs import (
     Jobs,
     check_remaining_jobs,
     get_ready_list,
 )
-from Exercises.VerySimpleSimulator.src.task_handler import output_result_csv
-from Exercises.ResponseTimeAnalysis.src.real_time_analysis import (
-    response_time_analysis,
+from task_handler import output_result_csv
+import typer
+from real_time_analysis import (
     run_rta
 )
+from typing import Annotated
 
-def run_cycle(input_path: str) -> None:
+app = typer.Typer()
+
+@app.command(name="run_cycle")
+def run_cycle(input_path: Annotated[str, typer.Option("--path", "-p")]) -> None:
     # Print the contents of the CSV file
     # Task  BCET(best case execution time)  WCET(worst case execution time)  Period  Deadline  Priority
     rta = run_rta(input_path)
@@ -65,4 +69,7 @@ def run_cycle(input_path: str) -> None:
 
     # Convert wcrt_dict values to float
     wcrt_dict = {k: float(v) for k, v in wcrt_dict.items()}
-    output_result_csv(input_path, wcrt_dict, tasks_dict, rta)
+    output_result_csv(input_path, wcrt_dict, tasks_dict, rta)
+
+if __name__ == "__main__":
+    app()

pyproject.toml

@@ -0,0 +1,42 @@
+[build-system]
+requires = ["setuptools"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "Simulator"
+version = "0.0.1"
+description = "A simulator for distributed real time systems project"
+authors = [
+    { name = "group74", email = "your@email.com" },
+]
+
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "Programming Language :: Python :: 3",
+    "Topic :: Scientific/Engineering :: Artificial Intelligence",
+]
+readme = "README.md"
+requires-python = ">=3.11"
+dynamic = ["dependencies", "optional-dependencies"]
+
+[project.scripts]
+simulator = "VerySimpleSimulator.src.very_simple_simulator:app"
+run_cycle = "VerySimpleSimulator.src.very_simple_simulator:run_cycle"
+
+[tool.setuptools.packages.find]
+where = ["Exercises"]
+include = ["Exercises.*"]
+
+
+[tool.setuptools.dynamic]
+dependencies = {file = ["requirements.txt"]}
+
+[tool.setuptools.dynamic.optional-dependencies]
+dev = {file = ["requirements_dev.txt"]}
+
+[tool.ruff]
+line-length = 120
+lint.select = ["I"]
+
+[tool.coverage.run]
+omit = ["tests/*"]

readme.md

@@ -0,0 +1,9 @@
+## Running the Simulator
+
+To run the simulator, navigate to the `Exercises/VerySimpleSimulator/src` folder and execute the following command:
+
+```bash
+python very_simple_simulator.py -p <path_to_task>
+```
+
+Replace `<path_to_task>` with the path to your task file.

requirements.txt

@@ -0,0 +1,3 @@
+numpy==2.2.4
+pandas==2.2.3
+typer==0.15.2