Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 48 additions & 18 deletions job_unit_scheduler.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,8 @@
import time



class Unit:


def __init__(self, unit_id: int, capabilities: List[str]):
self.id = unit_id

Expand All @@ -18,40 +16,31 @@ def __str__(self):
return f"[Unit {self.id}] Caps: {', '.join(self.capabilities)}"




class Job:



def __init__(self, job_id, name, description, deadline, priority: int = 5):
def __init__(self, job_id, name, description, deadline, priority=5):
# Add new job object
self.id = job_id
self.name = name
self.description = description
self.deadline = deadline
self.units = []
self.complete = False


self.priority = priority

def __str__(self):

status = "Completed" if self.complete else "Pending"

return f"[{self.id}] {self.name} (P{self.priority}) | Status: {status}"



class JobUnitScheduler:


def __init__(self):

self.jobs: List[Job] = []
self.next_id = 1


self.units: List[Unit] = []

# US6: Dictionary to store priority labels (default values)
Expand All @@ -62,12 +51,15 @@ def __init__(self):
4: "Low",
5: "Background"
}

self.Des_length = 100
self.system_capabilities: set = set()

# US1: Add Job
def add_job(self, name, description, deadline=None, priority=5):


# US Description Validation (if characters exceed >= 100)
if len(description) > self.Des_length:
return "Description too long! Try to add fewer than 100 characters"
# US9 Deadline Handling (if called without a deadline)
if deadline is None:

Expand All @@ -77,7 +69,6 @@ def add_job(self, name, description, deadline=None, priority=5):

deadline_dt = deadline


job = Job(self.next_id, name, description, deadline_dt, priority)
self.jobs.append(job)
self.next_id += 1
Expand Down Expand Up @@ -119,8 +110,40 @@ def delete_job(self, job_id):
return True
return False

# US7: Add a unit inside a job
def add_unit(self, job_id, name):
for job in self.jobs:
if job.id == job_id:
job.units.append(name)
return True
return False

# US8: View units by Job ID
def view_units(self, job_id):
for job in self.jobs:
if job.id == job_id:
return job.units # return the list of units
return None # job not found

# US9: Complete a job
def complete_job(self, job_id):
for job in self.jobs:
if job.id == job_id:
job.complete = True
return True # job marked completed
return False # job not found

# US18: Clear completed jobs
def remove_completed_jobs(self):
prev = len(self.jobs)
current_job = []
for job in self.jobs:
if not job.complete:
current_job.append(job)
self.jobs = current_job
erased = prev - len(self.jobs)

return f"{erased} completed job(s) removed."

def add_unit(self, unit_id, capabilities: List[str]):

Expand All @@ -137,7 +160,14 @@ def us6_set_priority_label(self, priority_level: int, label: str) -> bool:
return True
return False


def us6_get_priority_legend(self) -> Dict[int, str]:

return self.priority_labels
return self.priority_labels

# US55: View Unit History
def us4_view_unit_history(self, unit_id: int) -> List[float]:
"""Returns the list of historical load values for a specific unit."""
for unit in self.units:
if unit.id == unit_id:
return unit.historical_loads
return []
109 changes: 97 additions & 12 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,16 @@
s = JobUnitScheduler()

s.add_unit(1, ['GPU', 'High_Mem', 'NVMe'])
s.units[0].load = 80.0
s.units[0].historical_loads = [60.0, 70.0, 80.0]
s.add_unit(2, ['CPU', 'Storage', 'NVMe'])
s.units[1].load = 15.0
s.units[1].historical_loads = [10.0, 15.0]

print("Welcome to Job Scheduler")



def show_menu():
print("===Job Scheduler Menu ===)") # manage your jobs
print("1. Add Job (US1)")
Expand All @@ -17,17 +22,22 @@ def show_menu():
print("4. Edit Job Description (US4)")
print("5. Rename Job (Us5)")
print("6. Delete Job (Us6)")
print("7. Set Job Priority Label (US58)")
print("7. Add Unit (US7)")
print("8. View Units (US8)")
print("9. Complete Job (US9)")
print("10. Clear Completed Jobs (US18)")
print("11. Job/Unit Configuration Menu (US58/US55)")
print("0. Exit")





def show_config_menu():
print("\n--- Configuration & Creation Menu (Option 11) ---")
print("1. Set Job Priority Label (US58)")
print("2. View Unit Load History(US55)")
print("0. Back to Main Menu")
while True:
show_menu()
print("\n")
choice = input("Enter your choice ( 1 to 7):- ")
choice=input("Enter your choice ( 1 to 11):- ")

# US1: Add job
if choice == "1":
Expand All @@ -37,7 +47,10 @@ def show_menu():
deadline = input("Enter job deadline (YYYY-MM-DD): ")

job = s.add_job(name, description, deadline)

# US16: Shows Description too long without crash
if isinstance(job, str):
print("\n" + job)
continue
print(f"\nJob added successfully! Job ID: {job.id}")

# US2: List all Jobs
Expand Down Expand Up @@ -101,9 +114,57 @@ def show_menu():
else:
print("\nJob not found.")

#US7 Add Unit
elif choice=="7":
print("\n=> Add Job Unit")
job_id = int(input("Enter job ID to add unit into: "))
name = input("Enter unit name: ")

Done = s.add_unit(job_id, name)

# 1. US58: Set Job Priority Label
elif choice == "7":
if Done:
print("\nUnit added successfully!")
else:
print("\nJob not found. Please enter a valid Job ID.")

# US8: View Units
elif choice == "8":
print("\n=> View Units")
job_id = int(input("Enter job ID to view units: "))
units = s.view_units(job_id)
if units is None:
print("\nJob not found.")
else:
if len(units) == 0:
print("\nNo units added yet.")
else:
print("\nUnits for this job are: ")
for unit in units:
print(f"\t-,{unit}")

#US9: Complete Job
elif choice=="9":
print("\n=> Complete Job")
job_id = int(input("Enter job ID to complete: "))
unit = s.complete_job(job_id)
if unit:
print("\nJob marked as completed!")
else:
print("\nJob not found.")

# US18: Clear completed Job
elif choice == "10":
print("\n=== Clear Completed Jobs ===")
info = s.remove_completed_jobs()
print(info)

# US58/US55 Configuration Menu
elif choice == "11":
while True:
show_config_menu()
config_choice = input("\nEnter your configuration choice (1-2):- ")
# US58: Set Job Priority Label
if config_choice == "1":
print("\n--- Set Priority Label (US58) ---")
try:
level = int(input("Enter Priority Level to change (1 to 5): "))
Expand All @@ -120,12 +181,36 @@ def show_menu():
except ValueError:
print("Invalid priority level entered. Must be an integer between 1 and 5.")

# US55: View Unit History
elif config_choice == "2":
print("\n=> View Unit History")
try:
unit_id = int(input("Enter Unit ID (e.g., 1 or 2) to view history: "))

# Call the US4 method added to your scheduler
history = s.us4_view_unit_history(unit_id)


if history:
print(f"\n--- History for Unit {unit_id} ---")
print(f"Total entries: {len(history)}")
print(f"Load History: {history}")
else:
print(f"Unit {unit_id} not found or history is empty.")
except ValueError:
print("Invalid input. Please enter a number.")
except AttributeError:
# Catch if the setup or method definition was skipped
print("Error: The us4_view_unit_history method is not fully implemented in the scheduler.")

# 0. Back to Main Menu
elif config_choice == "0":
break
else:
print("Invalid choice. Please try again.")
# Exit from menu
elif choice == "0":
print("Exiting...")
break
else:
print("Invalid choice. Please try again.")

else:
print("Invalid choice. Please try again.")