Skip to content

Style: Auto-format Python code (main) #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
oleh-devlab merged 1 commit into from
Jul 17, 2026
Merged

Style: Auto-format Python code (main) #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
8 changes: 7 additions & 1 deletion src/data_read.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,13 @@ def load_data(filepath):
for t in data.get("user_tasks", [])
]
time_blocks = [
TimeBlock(start=_parse_datetime(b["start"]), end=_parse_datetime(b["end"]), daily=b.get("daily", True), name=b.get("name", ""), id=b.get("id"))
TimeBlock(
start=_parse_datetime(b["start"]),
end=_parse_datetime(b["end"]),
daily=b.get("daily", True),
name=b.get("name", ""),
id=b.get("id"),
)
for b in data.get("time_blocks", [])
]

Expand Down
18 changes: 11 additions & 7 deletions src/restrictions.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,15 +16,14 @@ def calculate_horizon(user_tasks, time_blocks, min_horizon_days=14, step_minutes
raise ValueError(f"min_horizon_days must be greater than 0, got {min_horizon_days}")

steps_per_day = 1440 // step_minutes

# Calculate a pessimistic safe maximum bound to generate time windows
base_horizon = sum(task.duration_steps for task in user_tasks)
max_deadline = max(
(t.deadline_steps for t in user_tasks if getattr(t, "deadline_steps", None) is not None), default=0
)
pessimistic_max = max(base_horizon * 3 + steps_per_day, min_horizon_days * steps_per_day, max_deadline)


blocked_intervals = generate_blocked_intervals(time_blocks, pessimistic_max, step_minutes)
free_windows = []
curr = 0
Expand All @@ -35,7 +34,6 @@ def calculate_horizon(user_tasks, time_blocks, min_horizon_days=14, step_minutes
if curr < pessimistic_max:
free_windows.append((curr, pessimistic_max))


task_by_id = {t.id: t for t in user_tasks if getattr(t, "id", None) is not None}
in_degree = {t.id: 0 for t in user_tasks if getattr(t, "id", None) is not None}
adj = {t.id: [] for t in user_tasks if getattr(t, "id", None) is not None}
Expand Down Expand Up @@ -72,13 +70,15 @@ def calculate_horizon(user_tasks, time_blocks, min_horizon_days=14, step_minutes
if getattr(t, "id", None) is None or t.id not in added:
sorted_tasks.append(t)


release_times = {t.id: getattr(t, "start_steps", 0) for t in user_tasks if getattr(t, "id", None) is not None}
simulated_horizon = 0

for task in sorted_tasks:
chunks = []
if getattr(task, "min_chunk_duration_steps", None) is not None and task.duration_steps > task.min_chunk_duration_steps:
if (
getattr(task, "min_chunk_duration_steps", None) is not None
and task.duration_steps > task.min_chunk_duration_steps
):
max_chunks = math.ceil(task.duration_steps / task.min_chunk_duration_steps)
remainder = task.duration_steps - (max_chunks - 1) * task.min_chunk_duration_steps
for _ in range(max_chunks - 1):
Expand All @@ -87,7 +87,11 @@ def calculate_horizon(user_tasks, time_blocks, min_horizon_days=14, step_minutes
else:
chunks.append(task.duration_steps + task.break_duration_steps)

t_curr = release_times.get(task.id, getattr(task, "start_steps", 0)) if getattr(task, "id", None) is not None else getattr(task, "start_steps", 0)
t_curr = (
release_times.get(task.id, getattr(task, "start_steps", 0))
if getattr(task, "id", None) is not None
else getattr(task, "start_steps", 0)
)
deadline = task.deadline_steps if getattr(task, "deadline_steps", None) is not None else math.inf

for chunk_size in chunks:
Expand All @@ -103,7 +107,7 @@ def calculate_horizon(user_tasks, time_blocks, min_horizon_days=14, step_minutes
new_windows.append((w_start, start_time))
if w_end > start_time + chunk_size:
new_windows.append((start_time + chunk_size, w_end))
new_windows.extend(free_windows[i+1:])
new_windows.extend(free_windows[i + 1 :])
free_windows = new_windows
placed = True
break
Expand Down
6 changes: 4 additions & 2 deletions src/routine_expansion.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ def expand_routines(routines, now, horizon_minutes, step_minutes=1):
continue

# Check if routine should be skipped
if getattr(routine, 'resume_after', None) and current_date <= routine.resume_after:
if getattr(routine, "resume_after", None) and current_date <= routine.resume_after:
continue

# It's a valid day for this routine
Expand All @@ -51,7 +51,9 @@ def expand_routines(routines, now, horizon_minutes, step_minutes=1):
end_steps = start_steps + routine.duration_steps

if end_steps > 0 and start_steps <= horizon_minutes:
extra_blocks.append(TimeBlock(start_steps, end_steps, daily=False, name=routine.name, id=routine.id))
extra_blocks.append(
TimeBlock(start_steps, end_steps, daily=False, name=routine.name, id=routine.id)
)
routine_info.append(
{
"name": routine.name,
Expand Down
56 changes: 33 additions & 23 deletions src/scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,12 +108,14 @@ def _expand_timeblocks_for_export(
if curr_end > 0: # at least partially in the future
start_dt = now + timedelta(minutes=curr_start * step_minutes)
end_dt = now + timedelta(minutes=curr_end * step_minutes)
result.append(ScheduledTimeBlock(
name=tb.name,
start_time=start_dt,
end_time=end_dt,
id=tb.id,
))
result.append(
ScheduledTimeBlock(
name=tb.name,
start_time=start_dt,
end_time=end_dt,
id=tb.id,
)
)
curr_start += steps_per_day
curr_end += steps_per_day
else:
Expand All @@ -122,12 +124,14 @@ def _expand_timeblocks_for_export(
if tb.end > 0: # at least partially in the future
start_dt = now + timedelta(minutes=tb.start * step_minutes)
end_dt = now + timedelta(minutes=tb.end * step_minutes)
result.append(ScheduledTimeBlock(
name=tb.name,
start_time=start_dt,
end_time=end_dt,
id=tb.id,
))
result.append(
ScheduledTimeBlock(
name=tb.name,
start_time=start_dt,
end_time=end_dt,
id=tb.id,
)
)
return result


Expand Down Expand Up @@ -215,17 +219,23 @@ def solve(
steps_per_day = 1440 // self.step_minutes
base_horizon = sum(task.duration_steps for task in active_tasks)
max_deadline = max(
(getattr(t, "deadline_steps", 0) for t in self.tasks if getattr(t, "deadline_steps", None) is not None), default=0
(getattr(t, "deadline_steps", 0) for t in self.tasks if getattr(t, "deadline_steps", None) is not None),
default=0,
)
pessimistic_max = max(base_horizon * 3 + steps_per_day, actual_horizon_days * steps_per_day, max_deadline)

sim_extra_tasks, sim_extra_blocks, _ = expand_routines(self.routines, now, pessimistic_max, self.step_minutes)
combined_sim_tasks = active_tasks + sim_extra_tasks
combined_sim_blocks = processed_blocks + sim_extra_blocks

# 2. Calculate mathematical minimum horizon using simulation (tracks only non-routine tasks)
simulated_horizon = calculate_horizon(combined_sim_tasks, combined_sim_blocks, min_horizon_days=actual_horizon_days, step_minutes=self.step_minutes)

simulated_horizon = calculate_horizon(
combined_sim_tasks,
combined_sim_blocks,
min_horizon_days=actual_horizon_days,
step_minutes=self.step_minutes,
)

# 3. Add 1 day of slack and snap to end of day to give the solver breathing room
horizon = simulated_horizon + steps_per_day
horizon = math.ceil(horizon / steps_per_day) * steps_per_day
Expand All @@ -234,8 +244,10 @@ def solve(
extra_tasks, extra_blocks, routine_info = expand_routines(self.routines, now, horizon, self.step_minutes)

# Pre-filter expired routine tasks
expired_routine_tasks = [t for t in extra_tasks if getattr(t, 'deadline_steps', None) is not None and t.deadline_steps <= 0]
extra_tasks = [t for t in extra_tasks if getattr(t, 'deadline_steps', None) is None or t.deadline_steps > 0]
expired_routine_tasks = [
t for t in extra_tasks if getattr(t, "deadline_steps", None) is not None and t.deadline_steps <= 0
]
extra_tasks = [t for t in extra_tasks if getattr(t, "deadline_steps", None) is None or t.deadline_steps > 0]

# Combine base and extra data
combined_tasks = active_tasks + extra_tasks
Expand All @@ -260,7 +272,7 @@ def solve(
packer_status = solver.solve(model)
if packer_status == cp_model.MODEL_INVALID:
print("MODEL_INVALID Details:", model.Validate())

gravity_status = None

if packer_status in (cp_model.OPTIMAL, cp_model.FEASIBLE):
Expand Down Expand Up @@ -288,9 +300,7 @@ def solve(

result = ScheduleResult(packer_status_name=packer_str, gravity_status_name=gravity_str)
result.horizon = horizon * self.step_minutes
result.scheduled_timeblocks = _expand_timeblocks_for_export(
combined_blocks, horizon, now, self.step_minutes
)
result.scheduled_timeblocks = _expand_timeblocks_for_export(combined_blocks, horizon, now, self.step_minutes)

# Add pre-filtered expired items to skipped lists
for task in expired_tasks:
Expand Down
6 changes: 5 additions & 1 deletion src/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,11 @@ def process_time_blocks(time_blocks, now, step_minutes=1):

if end_min > 0:
new_block = TimeBlock(
start=math.floor(start_min / step_minutes), end=math.ceil(end_min / step_minutes), daily=False, name=b.name, id=b.id
start=math.floor(start_min / step_minutes),
end=math.ceil(end_min / step_minutes),
daily=False,
name=b.name,
id=b.id,
)
processed_blocks.append(new_block)

Expand Down