Commit 35855c5b authored by Marco Maida's avatar Marco Maida 🌱
Browse files

Add task set parsing, input handling

parent 53e0d050
Pipeline #57971 failed with stages
in 41 seconds
# Python - Byte-compiled / optimized / DLL files
\ No newline at end of file
task set:
- id: 1
period: 100
worst-case execution time: 50
priority: 1
\ No newline at end of file
task set:
- id: 1
period: 100
worst-case execution time: 50
priority: 1
- id: 2
period: 200
jitter: 0
burst: 1
worst-case execution time: 20
priority: 2
\ No newline at end of file
import os, argparse, sys
import parser
def run():
### Parsing and checking input
opts = parse_args()
input_path = opts.input
output_path = opts.output
check_condition(os.path.exists(input_path), "Input file not found")
and os.path.isfile(input_path),
f"File not found: {input_path}")
check_condition(output_path is not None, "Please specify an output file.")
### Generating trace
problem_instance = parser.parse_file(input_path)
assert problem_instance is not None
def parse_args():
parser = argparse.ArgumentParser(description="The Trace Generation Tool")
parser.add_argument('input', help='Input file.')
parser.add_argument('-o', '--output', default=None, action='store',
help='Path of the file in which the trace will be generated.')
return parser.parse_args()
def check_condition(condition, error_message):
if not condition:
def save_file(path, text):
with open(path, "w") as f:
except Exception as e:
print("Error while saving certificate to '{path}'")
\ No newline at end of file
This module handles the parsing of the specification file.
Basic specification errors (type checking, absence of information)
are handled here, but specific validity checks (e.g. RBF monotone)
are left to other modules.
import yaml
import traceback
from task import Task
from problem_instance import ProblemInstance
DEBUG_PARSER = False # If true, prints tracebacks
# ********** Standard tag names **********
TASK_SET_TAG = "task set"
TASK_ID_TAG = "id"
TASK_WCET_TAG = "worst-case execution time"
TASK_PRIORITY_TAG = "priority"
TASK_PERIOD_TAG = "period"
TASK_JITTER_TAG = "jitter"
TASK_BURST_TAG = "burst"
allowed_tags_at_root_level = [TASK_SET_TAG]
# ****************************************
parser_status = "" # Used to print meaningful error messages
def parse_file (file):
# Parses the given file according to the PG specifications.
global parser_status
parser_status = "opening the file"
with open(file, 'r') as stream:
root = yaml.safe_load(stream)
parser_status = "parsing the root file"
check_for_allowed_tags(root, allowed_tags_at_root_level)
task_set_object = parse_required(root, TASK_SET_TAG)
parser_status = "parsing the task set information"
parse_fun = lambda t: parse_task (t)
task_set = list(map(parse_fun, task_set_object))
parser_status = "finishing the task set creation"
return ProblemInstance (task_set)
except Exception as e:
print(f"\nError while {parser_status}:")
if DEBUG_PARSER: print(traceback.format_exc())
def check_for_allowed_tags (father, allowed_tags):
# Checks that the tag of every object contained in `father` is
# contained in the allowed-objects list.
for key, _ in father.items():
assert key in allowed_tags, f"'{key}' is an unrecognized tag. Allowed tags are: {allowed_tags}."
def parse_required (father, tag):
# Given a father dictionary, fetches the given tag object.
# Raises errors if the object is not present, or if it's empty.
assert tag in father, f"The information is required: {tag}."
value = father[tag]
assert value is not None, f"The information cannot be empty: {tag}."
return father[tag]
def parse_natural_number (father, tag, default = None):
# Parses the value of a given tag in a given iterable father dictionary.
# Checks and raises exceptions if the tag is missing, or if its value is
# not a natural number.
if default is not None and tag not in father:
value = default
value = parse_required (father, tag)
assert value >= 0
try: return int(value)
except: raise Exception(f"'{tag}' has value '{value}', which is not a positive number.")
def parse_task (task):
# Given a dictionary, it parses a task.
global parser_status
parser_status = "parsing a task"
assert task is not None, "the task cannot be empty."
id = parse_natural_number(task, TASK_ID_TAG)
wcet = parse_natural_number(task, TASK_WCET_TAG)
priority = parse_natural_number(task, TASK_PRIORITY_TAG)
period = parse_natural_number(task, TASK_PERIOD_TAG)
jitter = parse_natural_number(task, TASK_JITTER_TAG, 0)
burst = parse_natural_number(task, TASK_BURST_TAG, 1)
check_for_allowed_tags(task, allowed_tags_at_task_level)
t = Task(id = id, wcet = wcet, priority=priority, period=period, jitter=jitter, burst=burst)
return t
\ No newline at end of file
class ProblemInstance:
def __init__(self, task_set):
# Checking task set
assert task_set is not None
ids = list(map(lambda t:, task_set))
assert len(set(ids)) == len(ids), "The task ids must be unique!"
self.task_set = task_set
def __str__ (self):
res = "\n------- Problem instance -------\n"
res += f"Tasks:\n"
for t in self.task_set:
res += f"{str(t)}\n"
res += "\n"
return res
__repr__ = __str__
def total_utilization(self):
return sum([task.utilization() for task in self.task_set])
\ No newline at end of file
This module contains the specification of task and a task set.
class Task:
def __init__(self, id, wcet, priority, period, jitter, burst): = id
self.wcet = wcet
self.priority = priority
self.period = period
self.jitter = jitter
self.burst = burst
def __str__(self):
task = f"(T{}: wcet={self.wcet}, priority={self.priority},"
task += f"period={self.period}, jitter={self.jitter}, burst={self.burst})"
return task
__repr__ = __str__
def utilization(self):
return self.wcet*self.burst/self.period
\ No newline at end of file
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment