beam: add sliding_window_max (monotonic deque) + tests
This commit is contained in:
1 parent
88419d3c85
commit
d9e4a66561
2 files changed
+111
No files matched your search
@@ -0,0 +1,65 @@
|
||||
"""Sliding window maximum (LeetCode 239).
|
||||
|
||||
Returns the maximum of each contiguous window of size `k` over `nums`
|
||||
in O(n) time using a monotonic deque of indices.
|
||||
|
||||
Subtle points handled:
|
||||
- empty input / k <= 0 -> []
|
||||
- k == 1 -> nums copy
|
||||
- k >= len(nums) -> [max(nums)] (or [] if empty)
|
||||
- duplicate values (deque keeps indices, not values, to handle expiry)
|
||||
- negative numbers
|
||||
"""
|
||||
from collections import deque
|
||||
from typing import List
|
||||
|
||||
|
||||
def sliding_window_max(nums: List[int], k: int) -> List[int]:
|
||||
if not nums or k <= 0:
|
||||
return []
|
||||
if k == 1:
|
||||
return list(nums)
|
||||
n = len(nums)
|
||||
if k >= n:
|
||||
return [max(nums)]
|
||||
|
||||
dq = deque() # indices, nums[dq[0]] is current max; values strictly decreasing
|
||||
out: List[int] = []
|
||||
for i, x in enumerate(nums):
|
||||
# evict indices outside window
|
||||
while dq and dq[0] <= i - k:
|
||||
dq.popleft()
|
||||
# maintain decreasing values
|
||||
while dq and nums[dq[-1]] <= x:
|
||||
dq.pop()
|
||||
dq.append(i)
|
||||
if i >= k - 1:
|
||||
out.append(nums[dq[0]])
|
||||
return out
|
||||
|
||||
|
||||
def _brute(nums, k):
|
||||
if not nums or k <= 0:
|
||||
return []
|
||||
return [max(nums[i:i + k]) for i in range(len(nums) - k + 1)]
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import random
|
||||
rng = random.Random(123)
|
||||
for _ in range(5000):
|
||||
n = rng.randint(0, 40)
|
||||
nums = [rng.randint(-20, 20) for _ in range(n)]
|
||||
k = rng.randint(0, n + 3) if n else rng.randint(0, 5)
|
||||
got = sliding_window_max(nums, k)
|
||||
# compute expected via brute respecting edge rules
|
||||
if not nums or k <= 0:
|
||||
exp = []
|
||||
elif k == 1:
|
||||
exp = list(nums)
|
||||
elif k >= len(nums):
|
||||
exp = [max(nums)]
|
||||
else:
|
||||
exp = [max(nums[i:i + k]) for i in range(len(nums) - k + 1)]
|
||||
assert got == exp, (nums, k, got, exp)
|
||||
print("self-check: 5000 random cases passed")
|
||||
@@ -0,0 +1,46 @@
|
||||
import sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||||
from sliding_window_max import sliding_window_max as f
|
||||
|
||||
def test_basic():
|
||||
assert f([1,3,-1,-3,5,3,6,7], 3) == [3,3,5,5,6,7]
|
||||
|
||||
def test_k1():
|
||||
assert f([5,4,3], 1) == [5,4,3]
|
||||
|
||||
def test_k_large():
|
||||
assert f([1,2,3], 5) == [3]
|
||||
|
||||
def test_empty():
|
||||
assert f([], 3) == []
|
||||
assert f([1,2,3], 0) == []
|
||||
|
||||
def test_all_equal():
|
||||
assert f([7,7,7,7], 2) == [7,7,7]
|
||||
|
||||
def test_negatives():
|
||||
assert f([-1,-2,-3,-4], 2) == [-1,-2,-3]
|
||||
|
||||
def test_single():
|
||||
assert f([42], 1) == [42]
|
||||
assert f([42], 5) == [42]
|
||||
|
||||
# brute fuzz
|
||||
import random
|
||||
rng = random.Random(99)
|
||||
for _ in range(3000):
|
||||
n = rng.randint(0, 25)
|
||||
nums = [rng.randint(-10, 10) for _ in range(n)]
|
||||
k = rng.randint(0, n + 2) if n else rng.randint(0, 4)
|
||||
got = f(nums, k)
|
||||
if not nums or k <= 0:
|
||||
exp = []
|
||||
elif k == 1:
|
||||
exp = list(nums)
|
||||
elif k >= len(nums):
|
||||
exp = [max(nums)]
|
||||
else:
|
||||
exp = [max(nums[i:i+k]) for i in range(len(nums)-k+1)]
|
||||
assert got == exp, (nums, k, got, exp)
|
||||
print("ALL TESTS PASSED")
|
||||
Reference in new issue
Block a user