volt: Q111 -- weighted_interval_scheduling handles zero-length intervals, rejects reversed input; new test_*.py cell collected by beam CHECK (rc 0, 415 passed / 2007 subtests)
This commit is contained in:
1 parent
df7a972b25
commit
a6491d724a
2 files changed
+155
-7
No files matched your search
@@ -0,0 +1,114 @@
|
|||||||
|
"""Q111: cells for weighted_interval_scheduling's zero-length and reversed input.
|
||||||
|
|
||||||
|
The module's own docstring defines overlap as intersection with POSITIVE
|
||||||
|
length, so a zero-length interval (start == end) is compatible with every
|
||||||
|
interval and with every other zero-length interval. Before this file the
|
||||||
|
module's only check was a `__main__` self-test whose generator emitted
|
||||||
|
`length = randint(1, 4)`, so no cell could construct the failing case and
|
||||||
|
`beam/CHECK` (which runs pytest over `test_*.py`) never imported the
|
||||||
|
module at all.
|
||||||
|
|
||||||
|
The enumerator below is written here rather than imported from the module's
|
||||||
|
`brute_force`, so the differential compares the DP against independent code.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import random
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from weighted_interval_scheduling import weighted_interval_scheduling as wis
|
||||||
|
|
||||||
|
|
||||||
|
def _overlap(a, b):
|
||||||
|
"""Positive-length intersection, exactly the rule in the module docstring."""
|
||||||
|
return min(a[1], b[1]) - max(a[0], b[0]) > 0
|
||||||
|
|
||||||
|
|
||||||
|
def _enumerate(intervals):
|
||||||
|
"""Exhaustive optimum over all subsets; independent of the module's DP."""
|
||||||
|
n = len(intervals)
|
||||||
|
best = 0
|
||||||
|
for mask in range(1 << n):
|
||||||
|
chosen = [intervals[k] for k in range(n) if mask >> k & 1]
|
||||||
|
blocked = False
|
||||||
|
for i in range(len(chosen)):
|
||||||
|
for j in range(i + 1, len(chosen)):
|
||||||
|
if _overlap(chosen[i], chosen[j]):
|
||||||
|
blocked = True
|
||||||
|
break
|
||||||
|
if blocked:
|
||||||
|
break
|
||||||
|
if not blocked:
|
||||||
|
best = max(best, sum(w for _, _, w in chosen))
|
||||||
|
return best
|
||||||
|
|
||||||
|
|
||||||
|
# --- the filed repro and its immediate neighbours -------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_zero_length_inside_positive_interval_is_compatible():
|
||||||
|
# Q111's minimal repro: the point interval at t=1 does not block [0,3].
|
||||||
|
assert wis([(1, 1, 5), (0, 3, 1)]) == 6
|
||||||
|
|
||||||
|
|
||||||
|
def test_zero_length_only():
|
||||||
|
assert wis([(2, 2, 7)]) == 7
|
||||||
|
assert wis([(2, 2, 7), (5, 5, 3)]) == 10
|
||||||
|
assert wis([(2, 2, 7), (2, 2, 3)]) == 10 # same point: still no overlap
|
||||||
|
|
||||||
|
|
||||||
|
def test_negative_weight_zero_length_is_dropped():
|
||||||
|
assert wis([(2, 2, -7), (0, 9, 4)]) == 4
|
||||||
|
assert wis([(2, 2, -7)]) == 0
|
||||||
|
assert wis([(2, 2, -7), (4, 4, -3), (0, 0, 1)]) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_zero_length_points_add_to_a_packed_schedule():
|
||||||
|
# two touching positive intervals [0,2] and [2,4] can both be taken;
|
||||||
|
# points sitting inside them are free on top of that.
|
||||||
|
assert wis([(0, 2, 3), (2, 4, 5), (1, 1, 2), (2, 2, 4), (3, 3, 1)]) == 15
|
||||||
|
|
||||||
|
|
||||||
|
def test_reversed_interval_is_rejected():
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
wis([(3, 1, 5)])
|
||||||
|
with pytest.raises(ValueError):
|
||||||
|
wis([(0, 2, 1), (5, 4, 9)])
|
||||||
|
|
||||||
|
|
||||||
|
# --- the positive-length behaviour this file must not disturb -------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_documented_example_and_boundaries():
|
||||||
|
assert wis([]) == 0
|
||||||
|
assert wis([(1, 3, 5), (2, 5, 6), (4, 6, 5), (6, 7, 4)]) == 14
|
||||||
|
assert wis([(0, 1, 5), (1, 2, 3)]) == 8 # touching endpoints coexist
|
||||||
|
assert wis([(0, 10, 100), (2, 8, 90)]) == 100 # nested: pick one
|
||||||
|
assert wis([(0, 2, -5), (3, 4, -1)]) == 0 # all negative: take nothing
|
||||||
|
|
||||||
|
|
||||||
|
# --- differential against the independent enumerator ----------------------
|
||||||
|
|
||||||
|
|
||||||
|
def _random_intervals(rng, allow_zero_length):
|
||||||
|
n = rng.randint(0, 7)
|
||||||
|
out = []
|
||||||
|
for _ in range(n):
|
||||||
|
s = rng.randint(0, 6)
|
||||||
|
length = rng.randint(0 if allow_zero_length else 1, 4)
|
||||||
|
out.append((s, s + length, rng.randint(-5, 9)))
|
||||||
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
def test_differential_positive_length_only():
|
||||||
|
rng = random.Random(7)
|
||||||
|
for _ in range(400):
|
||||||
|
ivs = _random_intervals(rng, allow_zero_length=False)
|
||||||
|
assert wis(ivs) == _enumerate(ivs), ivs
|
||||||
|
|
||||||
|
|
||||||
|
def test_differential_with_zero_length_intervals():
|
||||||
|
rng = random.Random(11)
|
||||||
|
for _ in range(400):
|
||||||
|
ivs = _random_intervals(rng, allow_zero_length=True)
|
||||||
|
assert wis(ivs) == _enumerate(ivs), ivs
|
||||||
@@ -23,10 +23,38 @@ Interval = Tuple[int, int, int] # (start, end, weight)
|
|||||||
|
|
||||||
|
|
||||||
def weighted_interval_scheduling(intervals: List[Interval]) -> int:
|
def weighted_interval_scheduling(intervals: List[Interval]) -> int:
|
||||||
|
"""Maximum total weight of a mutually compatible subset.
|
||||||
|
|
||||||
|
Compatibility is the docstring's rule -- an intersection of positive
|
||||||
|
length is the only thing that blocks a pair -- so a zero-length
|
||||||
|
interval (start == end) overlaps nothing, not even another zero-length
|
||||||
|
interval at the same point. Every zero-length interval with positive
|
||||||
|
weight can therefore be taken unconditionally and is added to the
|
||||||
|
optimum instead of entering the recurrence, which can only express
|
||||||
|
predecessor relations between positive-length ranges.
|
||||||
|
|
||||||
|
A reversed interval (end < start) is not an interval under this rule
|
||||||
|
(its range is empty in the other direction) and is rejected rather
|
||||||
|
than silently scheduled.
|
||||||
|
"""
|
||||||
if not intervals:
|
if not intervals:
|
||||||
return 0
|
return 0
|
||||||
|
free = 0
|
||||||
|
admissible: List[Interval] = []
|
||||||
|
for s, e, w in intervals:
|
||||||
|
if e < s:
|
||||||
|
raise ValueError(
|
||||||
|
"invalid interval (start %r > end %r): %r" % (s, e, (s, e, w))
|
||||||
|
)
|
||||||
|
if e == s:
|
||||||
|
if w > 0:
|
||||||
|
free += w
|
||||||
|
else:
|
||||||
|
admissible.append((s, e, w))
|
||||||
|
if not admissible:
|
||||||
|
return free
|
||||||
# sort by end time; keep original weights
|
# sort by end time; keep original weights
|
||||||
iv = sorted(intervals, key=lambda x: x[1])
|
iv = sorted(admissible, key=lambda x: x[1])
|
||||||
n = len(iv)
|
n = len(iv)
|
||||||
ends = [iv[i][1] for i in range(n)]
|
ends = [iv[i][1] for i in range(n)]
|
||||||
# dp[i] = best using first i intervals (1-indexed); dp[0]=0
|
# dp[i] = best using first i intervals (1-indexed); dp[0]=0
|
||||||
@@ -40,11 +68,17 @@ def weighted_interval_scheduling(intervals: List[Interval]) -> int:
|
|||||||
take = w + dp[j]
|
take = w + dp[j]
|
||||||
skip = dp[i - 1]
|
skip = dp[i - 1]
|
||||||
dp[i] = take if take > skip else skip
|
dp[i] = take if take > skip else skip
|
||||||
return dp[n]
|
return free + dp[n]
|
||||||
|
|
||||||
|
|
||||||
def brute_force(intervals: List[Interval]) -> int:
|
def brute_force(intervals: List[Interval]) -> int:
|
||||||
"""Exhaustive enumeration for verification on small inputs."""
|
"""Exhaustive enumeration for verification on small inputs.
|
||||||
|
|
||||||
|
Applies the positive-length overlap rule directly and does no input
|
||||||
|
validation, so it is only a reference for inputs the scheduler accepts:
|
||||||
|
reversed intervals (end < start) are rejected there and are outside
|
||||||
|
this function's contract.
|
||||||
|
"""
|
||||||
n = len(intervals)
|
n = len(intervals)
|
||||||
best = 0
|
best = 0
|
||||||
for mask in range(1 << n):
|
for mask in range(1 << n):
|
||||||
@@ -71,18 +105,18 @@ if __name__ == "__main__":
|
|||||||
# self-test with randomized comparison
|
# self-test with randomized comparison
|
||||||
import random
|
import random
|
||||||
random.seed(42)
|
random.seed(42)
|
||||||
for trial in range(2000):
|
for trial in range(4000):
|
||||||
n = random.randint(0, 8)
|
n = random.randint(0, 8)
|
||||||
ivs = []
|
ivs = []
|
||||||
for _ in range(n):
|
for _ in range(n):
|
||||||
s = random.randint(0, 6)
|
s = random.randint(0, 6)
|
||||||
length = random.randint(1, 4)
|
length = random.randint(0, 4) # 0 -> zero-length, admissible
|
||||||
e = s + length
|
e = s + length
|
||||||
w = random.randint(1, 9)
|
w = random.randint(-3, 9) # signed weights: negatives must be dropped
|
||||||
ivs.append((s, e, w))
|
ivs.append((s, e, w))
|
||||||
a = weighted_interval_scheduling(ivs)
|
a = weighted_interval_scheduling(ivs)
|
||||||
b = brute_force(ivs)
|
b = brute_force(ivs)
|
||||||
if a != b:
|
if a != b:
|
||||||
raise SystemExit(f"MISMATCH trial={trial} got={a} exp={b} ivs={ivs}")
|
raise SystemExit(f"MISMATCH trial={trial} got={a} exp={b} ivs={ivs}")
|
||||||
print("OK: 2000 randomized trials passed")
|
print("OK: 4000 randomized trials passed (zero-length and signed weights included)")
|
||||||
print("example:", weighted_interval_scheduling([(1, 3, 5), (2, 5, 6), (4, 6, 5), (6, 7, 4)]))
|
print("example:", weighted_interval_scheduling([(1, 3, 5), (2, 5, 6), (4, 6, 5), (6, 7, 4)]))
|
||||||
Reference in new issue
Block a user