ادغامِ تنبلِ جریان‌های مرتبِ حسگر در پایتون؛ تساوی و شمارِ برداشت‌ها

فرض کنیم چند گزارشِ حسگر جداگانه داریم که هر کدام از پیش برحسب زمان مرتب شده‌اند. می‌خواهیم آن‌ها را به یک جریان تبدیل کنیم، بدون آنکه برای نخستین نتیجه همهٔ گزارش‌ها را فهرست کنیم. این درس، ادغام چندراهه را با یک تودهٔ کمینه و سیاستِ روشنِ تساوی می‌سازد. داده‌ها محدود و ساختگی‌اند؛ هیچ فایل، حسگر یا ارتباط زنده‌ای خوانده نمی‌شود.

۱. چند جریان مرتب با یک واحد زمان

هر رکورد، زوجِ زمان صحیحِ میلی‌ثانیه و مقدار صحیحِ میلی‌ولت است. زمان‌ها نسبت به یک مبدأ مشترکِ تألیفی‌اند؛ ساعت دستگاه‌ها را همگام نکرده‌ایم. زمانِ هر منبع باید نامنفی و غیرکاهشی باشد، پس زمان برابر مجاز است. مقدارِ میلی‌ولت نیز در بازهٔ بستهٔ منفی پنج‌هزار تا مثبت پنج‌هزار قرار می‌گیرد؛ این حد، قرارداد آموزشی ماست.

منبع صفر سه رکورد در زمان‌های صفر، چهار و چهار دارد؛ منبع یک در زمان‌های یک و چهار و منبع دو در زمان‌های چهار و هفت نمونه دارند. مقدارها برای حفظِ هویتِ رکورد انتخاب شده‌اند. ادغام آن‌ها، میانگین‌گیری، هم‌ترازی کانال یا بازنمونه‌برداری نیست؛ هر هفت مشاهده باید بدون حذف و با منشأ خودش در خروجی باقی بماند.

مرجع heapq.merge پایتون نیز ادغام چند ورودیِ ازپیش‌مرتب را با خروجیِ تکرارگر توضیح می‌دهد. مرتب بودنِ تک‌تکِ منبع‌ها پیش‌شرط مهم است. کد حاضر، الگوریتم و اعتبارسنجیِ خودش را با heappush و heappop می‌نویسد؛ heapq.merge را فراخوانی نمی‌کند و همهٔ رفتارهای آن API را وعده نمی‌دهد.

فهرستِ بیرونیِ منبع‌ها باید list یا tuple واقعی و متناهی باشد. خودِ منبع‌ها تکرارپذیر یا تکرارگرند و در قراردادِ این آزمایش محدود و مستقل فرض می‌شوند. از یک شیء iterator مشترک برای دو منبع استفاده نمی‌کنیم؛ کد این حالت مستقیم را رد می‌کند. پوشش‌های متفاوت بر یک منبعِ پنهانِ مشترک را به‌طور عمومی کشف نمی‌کند.

۲. کوچک‌ترین سرِ جریان و سیاستِ تساوی

از هر منبعِ ناتهی فقط سرِ فعلی را در توده می‌گذاریم. چون ادامهٔ همان منبع از سرش کوچک‌تر نیست، کوچک‌ترین سر میانِ منبع‌ها، کوچک‌ترین رکوردِ باقی‌مانده است. بعد از تحویل آن، فقط جانشینِ همان منبع لازم می‌شود. این استدلال به پیش‌شرطِ مرتب بودن وابسته است؛ توده نمی‌تواند رکوردِ کوچک‌ترِ پنهان در ادامهٔ یک منبعِ نامرتب را حدس بزند.

کلیدِ هر سر، زمان، اندیسِ منبع در فهرستِ ورودی و جایگاهِ رکورد در همان منبع است. بنابراین در زمان چهار، هر دو رکورد منبع صفر، سپس رکورد منبع یک و بعد رکورد منبع دو می‌آیند. مقدارِ ولتاژ معیارِ تساوی نیست. عوض کردنِ ترتیبِ منبع‌ها می‌تواند ترتیبِ رکوردهای هم‌زمان را عوض کند؛ این انتخاب، سیاستِ صریحِ مثال است.

دو رکوردِ کاملاً هم‌مقدار هم حذف نمی‌شوند. جایگاهِ درونِ منبع صفرمبناست و برای توضیح منشأ در خروجی نگه داشته می‌شود؛ شمارهٔ توالیِ حسگر واقعی یا شناسهٔ جهانی نیست. همچنین اندیسِ منبع، نامِ فیزیکیِ دستگاه محسوب نمی‌شود. اگر مصرف‌کننده شناسهٔ پایدار می‌خواهد، باید آن را در قراردادِ رکورد تعریف کند.

توده در مسیرِ ادغام حداکثر یک سر از هر منبع نگه می‌دارد؛ کلِ ادامهٔ گزارش‌ها را در توده نمی‌ریزیم. با این حال، حافظهٔ بالادست، فهرستِ خروجیِ ساخته‌شده در آزمون و خودِ منبع‌های ورودی جدا هستند. از این ساختار، حافظهٔ ثابتِ کل برنامه یا صرفه‌جوییِ اندازه‌گیری‌شده نتیجه نمی‌گیریم. مقایسهٔ زمانی با مرتب‌سازیِ کامل هم اجرا نشده است.

۳. کد کاملِ ادغامِ تنبل

فایل را با نام lazy_kway_sensor_merge.py ذخیره و با python lazy_kway_sensor_merge.py اجرا کنید. اجرای ثبت‌شده Python 3.14.5 است و کتابخانهٔ بیرونی لازم ندارد. هر رکورد باید tuple دقیقِ دو عضوی باشد؛ list رکورد، رشتهٔ عددی، float، bool و مقدار نامعتبر به‌صورت ضمنی تبدیل نمی‌شوند. ساختار خروجی چهار مقدارِ زمان، منبع، جایگاه و میلی‌ولت دارد.

"""Own finite sorted-stream merge with explicit tie and lazy validation policy."""
from heapq import heappop, heappush
from itertools import combinations_with_replacement, product
import json
import platform


def merge_sorted(streams):
    # Validation is lazy too: this generator starts on its first next().
    if type(streams) not in (tuple, list):
        raise TypeError('finite list or tuple of independent streams required')
    iterators = [iter(stream) for stream in streams]
    if len({id(iterator) for iterator in iterators}) != len(iterators):
        raise ValueError('the same iterator cannot serve as two streams')
    previous = [None] * len(iterators)
    sequence = [0] * len(iterators)

    def pull(source):
        try:
            record = next(iterators[source])
        except StopIteration:
            return None
        if type(record) is not tuple or len(record) != 2:
            raise TypeError('record must be an exact (time_ms, value_mv) tuple')
        tick, value = record
        if type(tick) is not int or tick < 0:
            raise ValueError('time_ms must be an exact nonnegative integer')
        if type(value) is not int or not -5000 <= value <= 5000:
            raise ValueError('value_mv must be an exact integer in [-5000,5000]')
        if previous[source] is not None and tick < previous[source]:
            raise ValueError('source time moved backwards')
        previous[source] = tick
        position = sequence[source]; sequence[source] += 1
        return tick, source, position, value

    heap = []
    for source in range(len(iterators)):
        head = pull(source)
        if head is not None:
            heappush(heap, head)
    while heap:
        head = heappop(heap)
        yield head
        # Do not request the successor until the consumer resumes the merge.
        successor = pull(head[1])
        if successor is not None:
            heappush(heap, successor)


class Tracked:
    """Observation-only iterator; counts next attempts and successful records."""
    def __init__(self, rows):
        self.iterator = iter(rows)
        self.attempts = 0
        self.successes = 0

    def __iter__(self):
        return self

    def __next__(self):
        self.attempts += 1
        row = next(self.iterator)
        self.successes += 1
        return row


def materialized_oracle(streams):
    # Independent entire-input sort; no heap or merge_sorted call.
    rows = [(tick, source, position, value)
            for source, stream in enumerate(streams)
            for position, (tick, value) in enumerate(stream)]
    return tuple(sorted(rows, key=lambda row: (row[0], row[1], row[2])))


def rejects(call):
    try:
        call()
    except (TypeError, ValueError):
        return True
    return False


def run():
    streams = ([(0,1000),(4,1010),(4,1020)], [(1,-100),(4,-110)], [(4,3300),(7,3310)])
    result = tuple(merge_sorted(streams))
    expected = ((0,0,0,1000),(1,1,0,-100),(4,0,1,1010),(4,0,2,1020),
                (4,1,1,-110),(4,2,0,3300),(7,2,1,3310))
    checks = {}
    checks['hand_merged_trace_and_origin'] = result == expected
    checks['whole_sort_independent_oracle'] = result == materialized_oracle(streams)
    checks['duplicate_times_and_values_not_deduplicated'] = tuple(merge_sorted(([(2,7),(2,7)],[(2,7)]))) == ((2,0,0,7),(2,0,1,7),(2,1,0,7))
    tracked = [Tracked(rows) for rows in streams]
    merged = merge_sorted(tracked)
    before = [item.successes for item in tracked]
    first = next(merged); after_one = [item.successes for item in tracked]
    second = next(merged); after_two = [item.successes for item in tracked]
    third = next(merged); after_three = [item.successes for item in tracked]
    merged.close(); after_close = [item.successes for item in tracked]
    checks['consumer_demand_pull_counts'] = (before == [0,0,0] and after_one == [1,1,1]
        and after_two == [2,1,1] and after_three == [2,2,1]
        and (first,second,third) == expected[:3])
    checks['close_does_not_pull_unrequested_successor'] = after_close == after_three and list(merged) == []
    full_tracked = [Tracked(rows) for rows in streams]
    tuple(merge_sorted(full_tracked))
    checks['successful_records_and_stop_attempts_distinct'] = ([item.successes for item in full_tracked] == [3,2,2]
        and [item.attempts for item in full_tracked] == [4,3,3])
    checks['empty_outer_and_empty_inner_streams'] = (tuple(merge_sorted([])) == ()
        and tuple(merge_sorted(([],[(0,0)],[]))) == ((0,1,0,0),))
    checks['one_stream_and_zero_time'] = tuple(merge_sorted(([(0,-5000),(0,5000)],))) == ((0,0,0,-5000),(0,0,1,5000))
    late = merge_sorted(([(0,100),(5,101),(3,102)],[(1,200),(10,201)]))
    delivered = [next(late), next(late), next(late)]
    checks['backwards_time_detected_only_when_consumed'] = (delivered == [(0,0,0,100),(1,1,0,200),(5,0,1,101)]
        and rejects(lambda: next(late)) and list(late) == [])
    hidden = merge_sorted(([(0,100),('bad',101)],))
    prefix = next(hidden); hidden.close()
    checks['unconsumed_invalid_suffix_not_claimed_validated'] = prefix == (0,0,0,100)
    bad_records = ([0,1], (0,), (True,1), (-1,1), (0,False), (0,5001), (0,float('nan')), ('0',1))
    checks['exact_record_types_and_units_contract'] = all(rejects(lambda bad=bad: list(merge_sorted(([bad],)))) for bad in bad_records)
    reused = iter([(0,1),(1,2)])
    checks['outer_shape_and_direct_iterator_alias_rejected'] = (rejects(lambda: list(merge_sorted(iter(streams))))
        and rejects(lambda: list(merge_sorted((reused,reused)))) and rejects(lambda: list(merge_sorted((123,)))))
    patterns = [tuple(x) for n in range(4) for x in combinations_with_replacement(range(3),n)]
    oracle_cases = 0; oracle_ok = True
    for left, right in product(patterns,repeat=2):
        data = ([(t,100+i) for i,t in enumerate(left)], [(t,-100-i) for i,t in enumerate(right)])
        oracle_ok &= tuple(merge_sorted(data)) == materialized_oracle(data)
        oracle_cases += 1
    checks['400_finite_sorted_pairs_oracle_ONE_group'] = bool(oracle_ok) and oracle_cases == 400
    renamed = tuple(merge_sorted((streams[2],streams[0],streams[1])))
    checks['stream_order_is_explicit_tie_policy'] = renamed == materialized_oracle((streams[2],streams[0],streams[1])) and renamed != result
    closed = []
    def owned_source():
        try:
            yield (0,100); yield (1,101)
        finally:
            closed.append('source-cleanup')
    source = owned_source(); wrapper = merge_sorted((source,)); next(wrapper); wrapper.close()
    not_closed_by_wrapper = closed == []
    source.close()
    checks['source_resource_lifetime_owned_by_caller'] = not_closed_by_wrapper and closed == ['source-cleanup']
    assert all(type(value) is bool and value for value in checks.values()), checks
    return {'python':platform.python_version(),'dependencies':'standard library only',
            'scope':'finite independent presorted sources, no IO/timers/threads or benchmark',
            'units':{'time':'integer ms from common authored origin','value':'integer mV'},
            'tie_policy':'time_ms, source-list index, within-source position',
            'merged_trace':result,'source_successful_pulls':{'before':before,'after_one_output':after_one,
                'after_two_outputs':after_two,'after_three_outputs':after_three,'after_wrapper_close':after_close},
            'full_successful_records':[item.successes for item in full_tracked],
            'full_next_attempts_including_exhaustion':[item.attempts for item in full_tracked],
            'prefix_delivered_before_backwards_time_error':delivered,
            'materialized_oracle_cases_inside_one_group':oracle_cases,'checks':checks}


if __name__ == '__main__':
    print(json.dumps(run(),ensure_ascii=False,indent=2))

ساختن مولد، حتی اعتبارسنجیِ فهرستِ بیرونی را اجرا نمی‌کند؛ بدنه در نخستین next آغاز می‌شود. در همین آغاز، از هر منبع یک تلاشِ خواندن داریم تا سرها معلوم شوند. پس تنبل بودن به معنی «برای نخستین خروجی فقط یک منبع خوانده می‌شود» نیست. اگر سرِ منبعی نامعتبر باشد، ممکن است سرِ منبع‌های قبلی قبلاً مصرف شده باشد، ولی هنوز خروجی نداشته باشیم.

قرارداد رسمیِ تکرارگرهای پایتون، next و پایان با StopIteration را توضیح می‌دهد. در این برنامه، پس از تشخیص پایانِ یک منبع، دیگر آن را در توده نمی‌گذاریم. توقف منبع، مقدارِ حسگر نیست و با رکوردِ دارای زمان یا ولتاژ صفر یکی گرفته نمی‌شود. منبعِ محدودِ تمام‌شده با همان شیء خودکار از ابتدا آغاز نمی‌شود.

yield پیش از درخواستِ جانشین قرار دارد؛ بنابراین تا مصرف‌کننده دوباره ادامه ندهد، سرِ بعدیِ منبعِ برنده خوانده نمی‌شود. این جایگاه در کد بخشی از قرارداد است. صرفِ جابه‌جا کردنِ pull به پیش از yield، شمارِ برداشت‌ها در توقفِ زودهنگام را تغییر می‌دهد، حتی اگر خروجیِ کاملِ هر دو برنامه برای ورودیِ سالم یکسان باشد.

۴. شمارِ برداشت‌ها و خطای دیرهنگام

در منبع‌های مشاهده‌پذیرِ Tracked، پیش از نخستین درخواست شمارِ رکوردهای برداشته‌شده صفر/صفر/صفر است. پس از یک خروجی به یک/یک/یک، پس از دو خروجی به دو/یک/یک و پس از سه خروجی به دو/دو/یک می‌رسد. بستنِ همین مولد در آن نقطه، شمار را بیشتر نمی‌کند. این‌ها برداشتِ موفق از iterator ساختگی‌اند؛ خواندنِ دیسک یا دریافت از شبکه نیستند.

{
  "python": "3.14.5",
  "dependencies": "standard library only",
  "scope": "finite independent presorted sources, no IO/timers/threads or benchmark",
  "units": {
    "time": "integer ms from common authored origin",
    "value": "integer mV"
  },
  "tie_policy": "time_ms, source-list index, within-source position",
  "merged_trace": [
    [
      0,
      0,
      0,
      1000
    ],
    [
      1,
      1,
      0,
      -100
    ],
    [
      4,
      0,
      1,
      1010
    ],
    [
      4,
      0,
      2,
      1020
    ],
    [
      4,
      1,
      1,
      -110
    ],
    [
      4,
      2,
      0,
      3300
    ],
    [
      7,
      2,
      1,
      3310
    ]
  ],
  "source_successful_pulls": {
    "before": [
      0,
      0,
      0
    ],
    "after_one_output": [
      1,
      1,
      1
    ],
    "after_two_outputs": [
      2,
      1,
      1
    ],
    "after_three_outputs": [
      2,
      2,
      1
    ],
    "after_wrapper_close": [
      2,
      2,
      1
    ]
  },
  "full_successful_records": [
    3,
    2,
    2
  ],
  "full_next_attempts_including_exhaustion": [
    4,
    3,
    3
  ],
  "prefix_delivered_before_backwards_time_error": [
    [
      0,
      0,
      0,
      100
    ],
    [
      1,
      1,
      0,
      200
    ],
    [
      5,
      0,
      1,
      101
    ]
  ],
  "materialized_oracle_cases_inside_one_group": 400,
  "checks": {
    "hand_merged_trace_and_origin": true,
    "whole_sort_independent_oracle": true,
    "duplicate_times_and_values_not_deduplicated": true,
    "consumer_demand_pull_counts": true,
    "close_does_not_pull_unrequested_successor": true,
    "successful_records_and_stop_attempts_distinct": true,
    "empty_outer_and_empty_inner_streams": true,
    "one_stream_and_zero_time": true,
    "backwards_time_detected_only_when_consumed": true,
    "unconsumed_invalid_suffix_not_claimed_validated": true,
    "exact_record_types_and_units_contract": true,
    "outer_shape_and_direct_iterator_alias_rejected": true,
    "400_finite_sorted_pairs_oracle_ONE_group": true,
    "stream_order_is_explicit_tie_policy": true,
    "source_resource_lifetime_owned_by_caller": true
  }
}

مصرف کامل هفت رکورد، شمار موفقِ سه/دو/دو دارد، اما تلاش‌های next چهار/سه/سه‌اند. تفاوت، یک تلاش برای تشخیصِ پایان هر منبع است. در آزمونِ دیگر، منبعی زمان‌های صفر، پنج و سپس سه دارد. پیش از کشفِ عقب‌رفتنِ زمان، خروجی‌های زمان صفر، یک و پنج تحویل شده‌اند. رکوردِ زمان سه خطا می‌دهد؛ خروجی‌های قبلی پس گرفته نمی‌شوند.

ادامهٔ نامعتبرِ خوانده‌نشده نیز معتبر اعلام نمی‌شود. اگر پس از نخستین رکورد متوقف شویم، کد آن ادامه را بررسی نکرده است. حتی بعضی سرهای منبع‌های دیگر ممکن است پیش‌خوانی شده باشند و هنوز تحویل نشده باشند. این ادغام، تراکنش، ذخیرهٔ پایدار یا بازگردانیِ حالتِ منبع‌ها نیست؛ تضمینِ خروجیِ کاملاً مرتب به قانونی بودنِ همهٔ ورودی‌ها مشروط است.

مرجع مستقل همهٔ رکوردهای محدود را با منشأ و جایگاه فهرست و یک‌باره مرتب می‌کند؛ از توده یا merge_sorted استفاده نمی‌کند. چهارصد زوجِ جریانِ مرتب با طول‌های صفر تا سه و زمان‌های صفر تا دو با این مرجع مقایسه می‌شوند؛ همهٔ آن‌ها یک گروه آزمون‌اند. پاسخ دستی، تساوی، پایان، نوع‌ها، مالکیت منبع و شمارِ برداشت، گروه‌های دیگر را تشکیل می‌دهند؛ تعداد واقعی پانزده است.

دسته‌بندی دادهٔ حسگر با مولد، پوششِ آخرین دسته و گذرِ دوباره را بررسی می‌کند؛ اینجا چند منبعِ مرتب را ادغام می‌کنیم. در مقابل، زمان‌بند رویداد حسگر افزودن، لغو و افقِ زمان مجازی دارد. اشتراکِ ابزار heapq باعث نمی‌شود این دو قرارداد یکی شوند؛ در ادغام حاضر رویداد تازه زمان‌بندی نمی‌کنیم.

۵. چهار پرسش پاسخ‌دار و مرزِ ادغام

پرسش اول: نخستین خروجیِ نمونه، چند رکورد را از منبع‌ها برداشته است؟ پاسخ: سه رکورد؛ یکی از هر منبعِ ناتهی. فقط یکی تحویل شده و دو سرِ دیگر برای انتخاب‌های بعدی نگه داشته شده‌اند. ساختنِ مولد پیش از آن هیچ رکوردی برنمی‌دارد.

پرسش دوم: در زمان چهار، چرا مقدارِ کوچک‌تر زودتر نمی‌آید؟ پاسخ: کلیدِ تساوی، اندیسِ منبع و جایگاه است؛ مقدارِ میلی‌ولت نقشی ندارد. دو نمونهٔ منبع صفر پیش از نمونهٔ منبع یک قرار می‌گیرند و هیچ مشاهده‌ای حذف نمی‌شود.

پرسش سوم: گرفتنِ یک پیشوندِ سالم یعنی ادامه نیز بررسی شده است؟ پاسخ: خیر؛ زمانِ عقب‌رونده فقط هنگام مصرف کشف می‌شود و خطا نتایجِ تحویل‌شده را بازنمی‌گرداند. برای انتشارِ اتمیِ کل گزارش، باید قرارداد و مرحلهٔ دیگری طراحی شود.

پرسش چهارم: بستنِ ادغام، منبع‌ها را هم می‌بندد؟ پاسخ: این پوشش مالک منبع‌ها نیست و آن‌ها را صریح نمی‌بندد. آزمون نشان می‌دهد finally منبعِ نگه‌داشته‌شده فقط با close صریحِ مالک اجرا می‌شود. منابع در ۲۶ مهر ۱۴۰۵ با مستندات 3.14.8 خوانده شدند؛ محدود بودنِ ورودی فرض است و اتصال به جریان بی‌پایان، تضمینِ پایان ایجاد نمی‌کند.

منابع و مطالعهٔ بیشتر

• Python 3.14؛ heapq، تودهٔ کمینه و ورودی‌های ازپیش‌مرتب merge
• Python 3.14؛ قرارداد کامل Iterator Types و پایان تکرارگر

بررسی منابع: ۲۶ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.