شبیه‌سازی بافر پخش در پایتون؛ مهلت ثابت، بستهٔ دیررس و ترتیب دریافت

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

۱. مهلت پخش را از زمان انتظار جدا کنیم

شش بستهٔ مورد انتظار با شماره‌های صفر تا پنج داریم. زمان قراردادی تولید هرکدام، شماره ضرب در ۲۰ میلی‌ثانیه است. مهلت پخش را با افزودن D به همین زمان می‌سازیم؛ D یک بار ۲۰ و بار دیگر ۴۰ میلی‌ثانیه انتخاب می‌شود. همهٔ عددها روی یک محور زمان ساختگی قرار دارند. برنامه هیچ ساعت سیستم، sleep یا اندازه‌گیری تأخیر واقعی را مصرف نمی‌کند.

پارامتر D از مبدأ ساختگی سنجیده می‌شود؛ برابر با «زمان دریافت اولین بسته به اضافهٔ D» نیست. همچنین همهٔ بسته‌های پذیرفته‌شده، دقیقاً D میلی‌ثانیه در بافر نمی‌مانند. زمان انتظار هر بسته، مهلت خودش منهای زمان ورود خودش است. این تفاوت برای تفسیر خروجی ضروری است: تغییر برنامهٔ پخش، با اندازه‌گیری میانگین تأخیر یک‌طرفهٔ شبکه یا گزارش سنجهٔ اسمی یک گیرندهٔ واقعی یکسان نیست.

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

۲. دادهٔ دریافت، جابه‌جایی ترتیب و نسخهٔ تکراری

بستهٔ صفر در زمان ۱۰، بستهٔ دو در ۴۵ و نسخهٔ تکراری آن در ۵۵ میلی‌ثانیه می‌رسد. بستهٔ یک در ۵۷ وارد می‌شود؛ پس ترتیب دریافت با ترتیب شماره‌ها فرق دارد. برای بستهٔ سه هیچ رخداد دریافتی در داده نداریم. بسته‌های چهار و پنج نیز به ترتیب در ۱۲۰ و ۱۳۰ میلی‌ثانیه ظاهر می‌شوند. این زمان‌ها برای آشکار کردن مرزها انتخاب شده‌اند و توزیع آماری یک شبکه را نمایش نمی‌دهند.

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

برچسب غایب فقط یعنی در تمام دادهٔ آزمایش، ورودی برای آن شماره ثبت نشده است. این مشاهده، اثبات گم شدن بسته در شبکه نیست. در لحظهٔ پخش، برنامه فقط می‌داند داده حاضر نیست. پس از پایان پردازش همهٔ رخدادها، گزارش تشخیصی مشخص می‌کند که بسته بعداً رسیده و دیررس بوده یا در داده اصلاً دیده نشده است. حلقهٔ تصمیم پخش برای پر کردن شکاف، آینده را نگاه نمی‌کند.

۳. کد کامل شبیه‌سازی و هجده بررسی

کد زیر با پایتون ۳٫۱۴٫۵ اجرا شده و فقط کتابخانهٔ استاندارد لازم دارد. آن را با نام playout_buffer.py ذخیره و با دستور python playout_buffer.py اجرا کنید. دو جدول پاسخ مورد انتظار به صورت مستقل نوشته شده‌اند. اجرا، هم شمارش‌ها و هم رفتار ترتیب، تکرار، مرز مهلت و زمان انتظار را می‌سنجد؛ این مجموعه اثبات همهٔ حالت‌های ممکن نیست.

"""Deterministic, offline deadline-buffer model; no RTP, audio or wall-clock API."""
import heapq
import json
import platform

PACKET_COUNT = 6
PERIOD_MS = 20
ARRIVALS = ((0, 10), (2, 45), (2, 55), (1, 57), (4, 120), (5, 130))


def simulate(arrivals, delay_ms, count=PACKET_COUNT, period_ms=PERIOD_MS):
    if (type(delay_ms) is not int or delay_ms < 0 or type(count) is not int
            or count <= 0 or type(period_ms) is not int or period_ms <= 0):
        raise ValueError('Integer delay>=0, count>0 and period>0 required')
    arrivals = tuple(arrivals)
    for sequence, arrival_ms in arrivals:
        if (type(sequence) is not int or not 0 <= sequence < count
                or type(arrival_ms) is not int or arrival_ms < 0):
            raise ValueError('Arrival must contain an expected sequence and nonnegative integer ms')
    deadlines = [sequence * period_ms + delay_ms for sequence in range(count)]
    # At equal time, arrival(priority0) precedes playout(priority1): inclusive deadline.
    events = [(arrival_ms, 0, index, sequence)
              for index, (sequence, arrival_ms) in enumerate(arrivals)]
    events += [(deadline, 1, sequence, sequence)
               for sequence, deadline in enumerate(deadlines)]
    heapq.heapify(events)
    first_arrival = {}
    buffer = {}
    playout = []
    arrival_order = []
    duplicates = 0
    peak_buffer_packets = 0
    while events:
        time_ms, kind, _, sequence = heapq.heappop(events)
        if kind == 0:
            if sequence in first_arrival:
                duplicates += 1
                continue
            first_arrival[sequence] = time_ms
            arrival_order.append(sequence)
            if time_ms <= deadlines[sequence]:
                buffer[sequence] = time_ms
                peak_buffer_packets = max(peak_buffer_packets, len(buffer))
        else:
            received_ms = buffer.pop(sequence, None)
            playout.append({'sequence': sequence, 'deadline_ms': time_ms,
                            'played': received_ms is not None,
                            'waiting_ms': None if received_ms is None else time_ms - received_ms})
    # Explain gaps only after the complete fixture trace; playout did not look ahead.
    for row in playout:
        sequence = row['sequence']
        row['first_arrival_ms'] = first_arrival.get(sequence)
        row['status'] = ('played' if row['played'] else
                         'late' if sequence in first_arrival else 'missing_in_fixture')
    return {'scheduled_delay_ms': delay_ms, 'period_ms': period_ms,
            'rows': playout, 'first_arrival_sequence_order': arrival_order,
            'counts': {status: sum(row['status'] == status for row in playout)
                       for status in ('played', 'late', 'missing_in_fixture')},
            'duplicates_ignored': duplicates, 'peak_buffer_packets': peak_buffer_packets,
            'remaining_buffer_packets': len(buffer)}


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


def run_example():
    short = simulate(ARRIVALS, 20)
    long = simulate(ARRIVALS, 40)
    statuses = lambda result: [row['status'] for row in result['rows']]
    expected_short = ['played', 'late', 'played', 'missing_in_fixture', 'late', 'late']
    expected_long = ['played', 'played', 'played', 'missing_in_fixture', 'played', 'played']
    extra_duplicate = simulate(ARRIVALS + ((0, 200),), 40)
    checks = {
        'short_six_expected_statuses': statuses(short) == expected_short,
        'long_six_expected_statuses': statuses(long) == expected_long,
        'short_counts_match_reference': short['counts'] == {'played': 2, 'late': 3, 'missing_in_fixture': 1},
        'long_counts_match_reference': long['counts'] == {'played': 5, 'late': 0, 'missing_in_fixture': 1},
        'all_scheduled_deadlines_shift_by_twenty_ms': all(b['deadline_ms'] - a['deadline_ms'] == 20
                                                       for a, b in zip(short['rows'], long['rows'])),
        'arrival_exactly_on_deadline_is_played': long['rows'][4]['played'] and long['rows'][4]['waiting_ms'] == 0,
        'arrival_one_ms_after_deadline_is_late': simulate(((0, 41),), 40, count=1)['rows'][0]['status'] == 'late',
        'out_of_order_arrival_observed': long['first_arrival_sequence_order'] == [0, 2, 1, 4, 5],
        'playout_order_follows_expected_sequence': [r['sequence'] for r in long['rows'] if r['played']] == [0, 1, 2, 4, 5],
        'first_copy_wins_and_one_duplicate_ignored': long['duplicates_ignored'] == 1 and long['rows'][2]['first_arrival_ms'] == 45,
        'post_playout_duplicate_cannot_replay': extra_duplicate['rows'] == long['rows'] and extra_duplicate['duplicates_ignored'] == 2,
        'input_listing_order_does_not_change_result': simulate(tuple(reversed(ARRIVALS)), 40) == long,
        'played_waits_match_independent_reference': [r['waiting_ms'] for r in long['rows']] == [30, 3, 35, None, 0, 10],
        'fixture_arrivals_not_before_generation_and_waits_nonnegative': all(t >= sequence * PERIOD_MS for sequence, t in ARRIVALS)
                                                                     and all(r['waiting_ms'] >= 0 for result in (short, long) for r in result['rows'] if r['played']),
        'each_expected_sequence_has_one_playout_decision': all([r['sequence'] for r in result['rows']] == list(range(6)) for result in (short, long)),
        'buffers_empty_after_trace_with_known_peaks': short['remaining_buffer_packets'] == long['remaining_buffer_packets'] == 0
                                                   and (short['peak_buffer_packets'], long['peak_buffer_packets']) == (1, 2),
        'invalid_timing_and_sequence_rejected': all(rejects(call) for call in (
            lambda: simulate(ARRIVALS, -1), lambda: simulate(ARRIVALS, True),
            lambda: simulate(((6, 10),), 40), lambda: simulate(((0, -1),), 40))),
        'empty_arrival_trace_marks_all_expected_packets_missing': simulate((), 40)['counts'] == {'played': 0, 'late': 0, 'missing_in_fixture': 6},
    }
    assert all(checks.values()), checks
    return {'python': platform.python_version(),
            'units': 'integer milliseconds on one synthetic time axis',
            'arrivals_sequence_and_ms': ARRIVALS, 'short_delay': short, 'long_delay': long,
            'real_network_or_audio_test': False, 'checks': checks}


if __name__ == '__main__':
    print(json.dumps(run_example(), indent=2))

صف اولویت heapq کوچک‌ترین رخداد را برمی‌گرداند. کلید ما زمان، نوع رخداد و شاخص جداکننده دارد؛ در زمان مشترک، اولویت صفر برای دریافت و یک برای پخش است. این سیاست متعلق به برنامه است، نه ویژگی خودکار یک بافر صوتی. heap، ترتیب زمانی پردازش را می‌سازد و نگاشت buffer دادهٔ آمادهٔ هر شماره را نگه می‌دارد؛ این دو ساختار مسئولیت متفاوتی دارند.

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

۴. دو انتخاب تأخیر و تفسیر نتیجه

با D برابر ۲۰ میلی‌ثانیه، دو بسته در مهلت پخش می‌شوند، سه بسته دیررس‌اند و یک شماره در داده غایب است. با D برابر ۴۰، پنج بسته پخش می‌شوند، دیررس نداریم و همان یک شماره همچنان غایب است. هر مهلت حالت دوم دقیقاً ۲۰ میلی‌ثانیه جابه‌جا شده است. نتیجه نشان می‌دهد در این جدول خاص، فرصت بیشتر سه شکاف ناشی از دیررسی را برطرف می‌کند؛ قاعدهٔ بهینهٔ عمومی به دست نیامده است.

{
  "python": "3.14.5",
  "units": "integer milliseconds on one synthetic time axis",
  "arrivals_sequence_and_ms": [
    [
      0,
      10
    ],
    [
      2,
      45
    ],
    [
      2,
      55
    ],
    [
      1,
      57
    ],
    [
      4,
      120
    ],
    [
      5,
      130
    ]
  ],
  "short_delay": {
    "scheduled_delay_ms": 20,
    "period_ms": 20,
    "rows": [
      {
        "sequence": 0,
        "deadline_ms": 20,
        "played": true,
        "waiting_ms": 10,
        "first_arrival_ms": 10,
        "status": "played"
      },
      {
        "sequence": 1,
        "deadline_ms": 40,
        "played": false,
        "waiting_ms": null,
        "first_arrival_ms": 57,
        "status": "late"
      },
      {
        "sequence": 2,
        "deadline_ms": 60,
        "played": true,
        "waiting_ms": 15,
        "first_arrival_ms": 45,
        "status": "played"
      },
      {
        "sequence": 3,
        "deadline_ms": 80,
        "played": false,
        "waiting_ms": null,
        "first_arrival_ms": null,
        "status": "missing_in_fixture"
      },
      {
        "sequence": 4,
        "deadline_ms": 100,
        "played": false,
        "waiting_ms": null,
        "first_arrival_ms": 120,
        "status": "late"
      },
      {
        "sequence": 5,
        "deadline_ms": 120,
        "played": false,
        "waiting_ms": null,
        "first_arrival_ms": 130,
        "status": "late"
      }
    ],
    "first_arrival_sequence_order": [
      0,
      2,
      1,
      4,
      5
    ],
    "counts": {
      "played": 2,
      "late": 3,
      "missing_in_fixture": 1
    },
    "duplicates_ignored": 1,
    "peak_buffer_packets": 1,
    "remaining_buffer_packets": 0
  },
  "long_delay": {
    "scheduled_delay_ms": 40,
    "period_ms": 20,
    "rows": [
      {
        "sequence": 0,
        "deadline_ms": 40,
        "played": true,
        "waiting_ms": 30,
        "first_arrival_ms": 10,
        "status": "played"
      },
      {
        "sequence": 1,
        "deadline_ms": 60,
        "played": true,
        "waiting_ms": 3,
        "first_arrival_ms": 57,
        "status": "played"
      },
      {
        "sequence": 2,
        "deadline_ms": 80,
        "played": true,
        "waiting_ms": 35,
        "first_arrival_ms": 45,
        "status": "played"
      },
      {
        "sequence": 3,
        "deadline_ms": 100,
        "played": false,
        "waiting_ms": null,
        "first_arrival_ms": null,
        "status": "missing_in_fixture"
      },
      {
        "sequence": 4,
        "deadline_ms": 120,
        "played": true,
        "waiting_ms": 0,
        "first_arrival_ms": 120,
        "status": "played"
      },
      {
        "sequence": 5,
        "deadline_ms": 140,
        "played": true,
        "waiting_ms": 10,
        "first_arrival_ms": 130,
        "status": "played"
      }
    ],
    "first_arrival_sequence_order": [
      0,
      2,
      1,
      4,
      5
    ],
    "counts": {
      "played": 5,
      "late": 0,
      "missing_in_fixture": 1
    },
    "duplicates_ignored": 1,
    "peak_buffer_packets": 2,
    "remaining_buffer_packets": 0
  },
  "real_network_or_audio_test": false,
  "checks": {
    "short_six_expected_statuses": true,
    "long_six_expected_statuses": true,
    "short_counts_match_reference": true,
    "long_counts_match_reference": true,
    "all_scheduled_deadlines_shift_by_twenty_ms": true,
    "arrival_exactly_on_deadline_is_played": true,
    "arrival_one_ms_after_deadline_is_late": true,
    "out_of_order_arrival_observed": true,
    "playout_order_follows_expected_sequence": true,
    "first_copy_wins_and_one_duplicate_ignored": true,
    "post_playout_duplicate_cannot_replay": true,
    "input_listing_order_does_not_change_result": true,
    "played_waits_match_independent_reference": true,
    "fixture_arrivals_not_before_generation_and_waits_nonnegative": true,
    "each_expected_sequence_has_one_playout_decision": true,
    "buffers_empty_after_trace_with_known_peaks": true,
    "invalid_timing_and_sequence_rejected": true,
    "empty_arrival_trace_marks_all_expected_packets_missing": true
  }
}

بستهٔ چهار در حالت دوم، دقیقاً در مهلت ۱۲۰ میلی‌ثانیه وارد می‌شود و با انتظار صفر پذیرفته می‌شود. در مقابل، بستهٔ دو از ۴۵ تا ۸۰، یعنی ۳۵ میلی‌ثانیه در بافر می‌ماند. بنابراین حتی با D برابر ۴۰، انتظار همهٔ بسته‌ها یکسان نیست. همهٔ ورودهای جدول پس از زمان تولید ساختگی‌اند؛ این شرط نیز بررسی می‌شود. افزایش فرصت، شمارهٔ غایب را نمی‌سازد و هیچ روش بازسازی صوتی در برنامه وجود ندارد.

سنجهٔ interarrival jitter در RFC 3550 موضوع متفاوتی از تصمیم مهلت پخش است؛ مهر زمانی RTP نیز به ساعت نمونه‌برداری مربوط می‌شود و مستقیماً همین میلی‌ثانیه‌های ساختگی نیست. مقالهٔ محاسبهٔ نوسان بین دریافت‌های RTP در پایتون آن سنجه را جدا بررسی می‌کند. کد حاضر نه آن برآوردگر را اجرا می‌کند و نه تطبیق ساعت فرستنده و گیرنده را حل می‌کند.

۵. تمرین‌های پاسخ‌دار و مرز استفاده در تلفن هوشمند

تمرین اول: چرا بستهٔ چهار با انتظار صفر پذیرفته می‌شود؟ پاسخ: ورود و مهلت هر دو ۱۲۰ هستند و قرارداد ما ورود را پیش از پخشِ همان زمان پردازش می‌کند. تمرین دوم: چرا افزایش D، بستهٔ سه را قابل پخش نکرد؟ پاسخ: هیچ رخداد دریافتی برای آن در داده وجود ندارد؛ تغییر مهلت، دادهٔ غایب تولید نمی‌کند.

تمرین سوم: چرا زمان انتظار بستهٔ دو ۳۵ میلی‌ثانیه است؟ پاسخ: مهلت آن در حالت دوم ۸۰ و ورودش ۴۵ است؛ اختلاف برابر ۳۵ می‌شود. D زمان انتظار ثابت هر بسته نیست. تمرین چهارم: آیا پنج پخش موفق یعنی کیفیت تماس بهتر اندازه‌گیری شده است؟ پاسخ: خیر؛ این فقط شمارش تصمیم‌های یک مدل است. هیچ شنونده، صوت، امتیاز ادراکی یا سامانهٔ پاسخ‌گوی هوشمند آزمایش نشده است.

برای مسیر واقعی، قرارداد محموله هم باید مستقل روشن باشد. آموزش ذخیره‌سازی WAV با PCM16 مرز دیگری از دادهٔ صوت را توضیح می‌دهد؛ سالم بودن فایل به معنی آماده بودن انتقال زنده نیست. منابع این درس در ۲۴ مهر ۱۴۰۵ خوانده شده‌اند. متن، آموزش پایهٔ تألیفی است و نتیجهٔ محصول، توصیهٔ تأخیر بهینه یا گزارش جدیدترین پژوهش‌ها محسوب نمی‌شود.

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

• RFC7005؛ مفهوم بافر رفع نوسان تأخیر و مرز زمان پخش
• RFC3550؛ زمان نمونه‌برداری RTP و سنجهٔ interarrival jitter
• Python3.14؛ heapq و صف اولویت رخدادها

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