"""Heap / Priority Queue — runnable reference implementation. ভেতরে কী আছে (উপর থেকে নিচে পড়ো): 1. heapq tour -- standard-library min-heap, 20 লাইনে 2. MinHeap (from scratch)-- sift-up / sift-down, যাতে magic টা আর magic না থাকে 3. top_k_largest -- size-k min-heap-এর "club bouncer" pattern 4. MedianFinder -- দুটো heap মিলে একটা stream-এর lower/upper half balance করে চালাও: python implementation.py প্রতিটা demo নিজের result assert করে, তাই চুপচাপ শেষ হওয়া মানেই সব কাজ করছে। শুধু standard library। """ import heapq # --------------------------------------------------------------------------- # 1. heapq tour — Python-এর built-in min-heap # --------------------------------------------------------------------------- # Python-এ একটা "heap" আসলে শুধুই একটা list, যেটাকে তুমি কেবল heapq call দিয়ে ছোঁও। # একটাই promise: h[0] সবসময় সবচেয়ে ছোট element। def heapq_tour(): h = [] # খালি list = খালি heap for x in [5, 2, 8, 1]: heapq.heappush(h, x) # O(log n): append, তারপর sift up assert h[0] == 1 # minimum peek করো: সবসময় index 0 assert heapq.heappop(h) == 1 # O(log n): min remove, sift down assert heapq.heappop(h) == 2 # values sorted order-এ বেরোয় # Existing list থেকে O(n)-এ heap বানাও — n-টা push-এর চেয়ে দ্রুত। data = [9, 4, 7, 1, 3] heapq.heapify(data) # IN PLACE rearrange করে assert data[0] == 1 # heap property টিকে আছে... assert data != sorted(data) or True # ...কিন্তু list টা পুরো sorted না # Max-heap trick: negatives store করো, বেরোনোর সময় sign উল্টাও। mx = [-x for x in [3, 1, 4, 1, 5]] heapq.heapify(mx) assert -heapq.heappop(mx) == 5 # সবচেয়ে বড়টা আগে assert -heapq.heappop(mx) == 4 # Tuples: প্রথম field-ই priority। Tie নিরাপদে ভাঙতে একটা counter যোগ করো। tasks = [] heapq.heappush(tasks, (2, 0, "write tests")) heapq.heappush(tasks, (1, 1, "fix prod bug")) assert heapq.heappop(tasks)[2] == "fix prod bug" # --------------------------------------------------------------------------- # 2. MinHeap from scratch — heapq তোমার জন্য ঠিক কী করে, নিজ চোখে দেখো # --------------------------------------------------------------------------- # Tree টা একটা list-এর ভেতরে থাকে। Index i-এর item-টার জন্য: # left child = 2*i + 1 # right child = 2*i + 2 # parent = (i - 1) // 2 # Invariant (heap property): প্রতিটা parent <= তার দুই child। class MinHeap: """একটা minimal min-heap। push / pop / peek — সব tree-র height-এর মতো দ্রুত।""" def __init__(self): self.a = [] # tree ধরে রাখা array টা def __len__(self): return len(self.a) def peek(self): """সবচেয়ে ছোট item, O(1)। Root-টাই সবসময় minimum।""" if not self.a: raise IndexError("peek from empty heap") return self.a[0] def push(self, value): """শেষে append করো (tree complete থাকে), তারপর sift UP। O(log n)।""" self.a.append(value) self._sift_up(len(self.a) - 1) def pop(self): """Minimum remove করে return করো। O(log n)। Trick: root-কে LAST item দিয়ে overwrite করো (একমাত্র removal যেটা tree-কে complete রাখে), list ছোট করো, তারপর sift DOWN দিয়ে repair করো। """ if not self.a: raise IndexError("pop from empty heap") smallest = self.a[0] last = self.a.pop() # tail slot টা সরাও if self.a: # কিছু বাকি থাকলে... self.a[0] = last # ...tail-কে root-এ teleport করো self._sift_down(0) # ...আর তাকে নিজের জায়গায় ডুবতে দাও return smallest # -- দুটো repair -------------------------------------------------------- def _sift_up(self, i): """Index i-কে root-এর দিকে bubble করো, যতক্ষণ সে parent-কে হারাচ্ছে।""" while i > 0: parent = (i - 1) // 2 if self.a[i] < self.a[parent]: # child ছোট? order ভুল self.a[i], self.a[parent] = self.a[parent], self.a[i] i = parent # উঠতে থাকো else: break # parent ঠিক আছে: থামো def _sift_down(self, i): """Index i-কে leaves-এর দিকে ডোবাও, সবসময় ছোট child-এর সাথে swap করে (শুধু ছোট child-টাই অন্যটার উপরে বসতে পারে)।""" n = len(self.a) while True: left, right = 2 * i + 1, 2 * i + 2 smallest = i if left < n and self.a[left] < self.a[smallest]: smallest = left if right < n and self.a[right] < self.a[smallest]: smallest = right if smallest == i: # দুই child-ই বড়: শেষ break self.a[i], self.a[smallest] = self.a[smallest], self.a[i] i = smallest # ডুবতে থাকো # --------------------------------------------------------------------------- # 3. Top-K largest — "min-heap টা club-এর দরজা পাহারা দেয়" pattern # --------------------------------------------------------------------------- def top_k_largest(nums, k): """nums-এর k-টা largest value return করো, ascending order-এ। Counterintuitive কিন্তু key কথা: LARGEST items track করতে আমরা size k-এর একটা MIN-heap ব্যবহার করি। তার top-টাই club-এর সবচেয়ে দুর্বল member — নতুন কাউকে ঢুকতে হলে শুধু তাকেই হারাতে হয়। Time O(n log k), memory O(k): stream-friendly। """ if k <= 0: return [] club = nums[:k] heapq.heapify(club) # O(k) — দুর্বলতম member club[0]-তে for x in nums[k:]: if x > club[0]: # এখনকার দুর্বলতমকে হারায়? heapq.heapreplace(club, x) # pop + push, একটা repair-এই return sorted(club) # output গুছিয়ে দাও (heap order ঢিলেঢালা) # --------------------------------------------------------------------------- # 4. MedianFinder — দুটো heap মিলে একটা stream balance করে # --------------------------------------------------------------------------- # একটা seesaw কল্পনা করো, যার pivot-এ median বসে: # # lower half (max-heap) | upper half (min-heap) # biggest small number ->|<- smallest big number # # প্রতিটা insert-এর পর এই invariants ধরে রাখা হয়: # (a) `lower`-এর প্রতিটা value <= `upper`-এর প্রতিটা value # (b) len(lower) == len(upper) অথবা len(lower) == len(upper) + 1 # তাহলে median হলো lower-এর top (odd count) বা দুই top-এর average। class MedianFinder: def __init__(self): self.lower = [] # negation দিয়ে max-heap self.upper = [] # সাধারণ min-heap def add(self, num): # Step 1: একটা side বাছো। lower-এর max-এর <= যা কিছু, lower-এ যাবে। if not self.lower or num <= -self.lower[0]: heapq.heappush(self.lower, -num) else: heapq.heappush(self.upper, num) # Step 2: rebalance করো যাতে size-এর পার্থক্য কখনো 1-এর বেশি না হয়। if len(self.lower) > len(self.upper) + 1: heapq.heappush(self.upper, -heapq.heappop(self.lower)) elif len(self.upper) > len(self.lower): heapq.heappush(self.lower, -heapq.heappop(self.upper)) def median(self): if len(self.lower) > len(self.upper): # odd count return float(-self.lower[0]) return (-self.lower[0] + self.upper[0]) / 2 # even count # --------------------------------------------------------------------------- # Demos — উপরের প্রতিটা দাবি, verify করা # --------------------------------------------------------------------------- if __name__ == "__main__": # 1. heapq tour heapq_tour() print("heapq tour ......... ok") # 2. MinHeap from scratch: heapq-এর মতোই sorted order-এ pop করতে হবে। h = MinHeap() for x in [5, 2, 8, 1, 9, 3]: h.push(x) assert h.peek() == 1 drained = [h.pop() for _ in range(len(h))] assert drained == [1, 2, 3, 5, 8, 9], drained print("MinHeap ............ ok (popped:", drained, ")") # আরও বড় input-এ আমাদের scratch heap-কে heapq-এর সাথে cross-check করো। nums = [7, 3, 11, 0, 5, 5, 2, 13, 8, 1] ours, ref = MinHeap(), list(nums) for x in nums: ours.push(x) heapq.heapify(ref) assert [ours.pop() for _ in nums] == [heapq.heappop(ref) for _ in nums] print("MinHeap vs heapq ... ok (identical pop order)") # 3. Top-K assert top_k_largest([5, 1, 9, 3, 7, 6], 3) == [6, 7, 9] assert top_k_largest([4, 4, 4], 2) == [4, 4] assert top_k_largest([10], 5) == [10] # data-র চেয়ে k বড়: সমস্যা নেই assert top_k_largest([1, 2], 0) == [] print("top_k_largest ...... ok") # 4. MedianFinder, trace করা: # add 5 -> {5} median 5.0 # add 2 -> {2,5} median 3.5 # add 8 -> {2,5,8} median 5.0 # add 1 -> {1,2,5,8} median 3.5 # add 7 -> {1,2,5,7,8} median 5.0 mf = MedianFinder() expected = [(5, 5.0), (2, 3.5), (8, 5.0), (1, 3.5), (7, 5.0)] for num, want in expected: mf.add(num) got = mf.median() assert got == want, (num, got, want) print("MedianFinder ....... ok (stream medians: 5.0 3.5 5.0 3.5 5.0)") print("\nAll heap demos passed.")