Answered step by step
Verified Expert Solution
Link Copied!

Question

00
1 Approved Answer

#@title ` AND _ OR _ Scheduler ` implementation class AND _ OR _ Scheduler ( object ) : def _ _ init _ _

#@title `AND_OR_Scheduler` implementation
class AND_OR_Scheduler(object):
def __init__(self):
# It is up to you to implement the initialization.
### YOUR SOLUTION HERE
self.tasks ={} # Dictionary to store tasks and their dependencies
self.completed = set() # Set to store completed tasks
self.available = set() # Set to store tasks available for execution
def add_and_task(self, t, dependencies):
"""Adds an AND task t with given dependencies."""
### YOUR SOLUTION HERE
self.tasks[t]=[set(dependencies), 'and'] # Store task with its dependencies and type
self.available.update(dependencies) # Update available tasks with dependencies
def add_or_task(self, t, dependencies):
"""Adds an OR task t with given dependencies."""
### YOUR SOLUTION HERE
self.tasks[t]=[set(dependencies),'or'] # Store task with its dependencies and type
self.available.update(dependencies) # Update available tasks with dependencies
@property
def done(self):
### YOUR SOLUTION HERE
return len(self.tasks)== len(self.completed)
@property
def available_tasks(self):
"""Returns the set of tasks that can be done in parallel.
A task can be done if:
- It is an AND task, and all its predecessors have been completed, or
- It is an OR task, and at least one of its predecessors has been completed.
And of course, we don't return any task that has already been
completed."""
### YOUR SOLUTION HERE
return self.available - self.completed
def mark_completed(self, t):
"""Marks the task t as completed, and returns the additional
set of tasks that can be done (and that could not be
previously done) once t is completed."""
### YOUR SOLUTION HERE
self.completed.add(t) # Mark task as completed
added_tasks = set()
for task, details in self.tasks.items(): # Iterate through tasks
deps, task_type = details
if ((task_type == 'and' and all(d in self.completed for d in deps)) or
(task_type =='or' and any(d in self.completed for d in deps))):
self.available.add(task) # Add task to available if all dependencies are completed
added_tasks.add(task)
return added_tasks
def show(self):
"""You can use the nx graph to display the graph. You may want to ensure
that you display AND and OR nodes differently."""
### YOUR SOLUTION HERE
and_tasks =[k for k, v in self.tasks.items() if v[1]== 'and'] # Extract AND tasks
or_tasks =[k for k, v in self.tasks.items() if v[1]=='or'] # Extract OR tasks
print("AND Tasks:", and_tasks)
print("OR Tasks:", or_tasks)
s = AND_OR_Scheduler()
s.add_and_task('a',['b','c'])
assert s.available_tasks =={'b','c'}
r = s.mark_completed('b')
assert r == set()
assert s.available_tasks =={'c'}
r = s.mark_completed('c')
assert r =={'a'}
assert s.available_tasks =={'a'}
r = s.mark_completed('a')
assert r == set()
assert s.available_tasks == set()

Step by Step Solution

There are 3 Steps involved in it

Step: 1

blur-text-image

Get Instant Access with AI-Powered Solutions

See step-by-step solutions with expert insights and AI powered tools for academic success

Step: 2

blur-text-image

Step: 3

blur-text-image

Ace Your Homework with AI

Get the answers you need in no time with our AI-driven, step-by-step assistance

Get Started

Students also viewed these Databases questions