رویداد DTMF در RTP با پایتون؛ چهار بایت payload و تکرار گزارش پایان

وقتی گزارش پایانِ کلید تلفن چند بار برسد، نباید صرفاً به تعداد بسته‌ها، چند بار فشردن کلید نتیجه بگیریم. یک رویداد نام‌دار می‌تواند در چند گزارش با مدتِ رو به رشد و پایانِ تکراری ظاهر شود. این درس چهار بایت payload ساختگیِ telephone-event را می‌خواند و یک دفتر محدودِ گزارش‌ها می‌سازد؛ نه صوت داریم، نه دریافت شبکه و نه اجرای فرمان تلفنی.

۱. چهار بایت رویداد را با صوت و سرآیند یکی نگیریم

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

RFC 4733 قالب نام‌دار را تعریف می‌کند. در این قالب، شمارهٔ payload type به‌صورت پویا تعیین می‌شود؛ از روی یک شمارهٔ دلخواه نباید آن را حدس زد. کد حاضر اصلاً payload type یا توافق SDP را نمی‌خواند؛ فرضِ جداشدن و شناسایی نوع ورودی، مسئولیت مرحلهٔ قبلیِ آزمایش است.

درس خواندن سرآیند RTP مرز payload را پیدا می‌کرد و محتوای آن را تفسیر نمی‌کرد. از سوی دیگر، آزمایش Goertzel برای DTMF از نمونه‌های صوتی آغاز می‌شود. اینجا فقط کدِ رویدادِ از پیش گزارش‌شده را می‌خوانیم؛ صحت تشخیص صوتِ فرستنده را ثابت نمی‌کنیم.

۲. بیت پایان و مدت را با واحد ساعت رسانه بخوانیم

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

شش بیت پایین، کد حجم را می‌سازند. برای این رویدادهای تُنی، عدد ده نمایندهٔ منفی ده dBm0 است؛ هرچه کد بزرگ‌تر، سطح گزارش‌شده پایین‌تر است. این فیلد دامنهٔ PCM یا توان اندازه‌گیری‌شده توسط برنامه نیست. ما اصلاً نمونهٔ صوت نداریم و از این عدد برای تولید موج یا تأیید کیفیت استفاده نمی‌کنیم.

مدت، تعداد تیکِ ساعت رسانه است. ۲۴۰ تیک در ساعت ۸۰۰۰ هرتز، سی میلی‌ثانیه و در ۱۶۰۰۰ هرتز، پانزده میلی‌ثانیه است. ساعت را صریح می‌دهیم؛ مقدار پیش‌فرضِ نمونه ۸۰۰۰ است. صفر را برای پروفایل DTMF این درس رد می‌کنیم؛ رویدادهای state و تفسیر مدت صفر در قالب گسترده‌تر، خارج از دامنه‌اند.

۳. هویت رویداد و تکرار پایان را صریح تعریف کنیم

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

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

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

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

۴. کد و بایت‌های دستی را مستقل مقایسه کنیم

فایل rtp_dtmf_event.py با Python 3.14.5 و کتابخانهٔ استاندارد اجرا شده است. مستندات struct ترتیب ! را شبکه‌ای و big-endian معرفی می‌کند. قالب !BBH چهار بایتِ بدون هم‌ترازیِ بومی می‌خواند؛ دو بایت مدت نباید مطابق معماری دستگاه، وارونه تفسیر شوند.

"""Bounded single-stream/single-epoch telephone-event fixture, not a full receiver."""
from fractions import Fraction
import json
import platform
import struct

DIGITS = '0123456789*#ABCD'


def uint(value, upper):
    if type(value) is not int or not 0 <= value <= upper:
        raise ValueError('expected bounded actual integer')


def parse_event(payload):
    if type(payload) is not bytes or len(payload) != 4:
        raise ValueError('single four-byte event payload required')
    event, flags, duration = struct.unpack('!BBH', payload)
    if event > 15 or duration == 0:
        raise ValueError('outside positive-duration DTMF profile')
    return {'event': event, 'digit': DIGITS[event], 'ended': bool(flags & 0x80),
            'reserved_seen': bool(flags & 0x40),  # Observed only: receiver ignores R.
            'volume_code': flags & 0x3F, 'duration_ticks': duration}


class EventLedger:
    def __init__(self, clock_hz=8000, max_events=8):
        uint(clock_hz, 192000)
        uint(max_events, 8)
        if clock_hz == 0 or max_events == 0:
            raise ValueError('positive clock and capacity required')
        self.clock_hz = clock_hz
        self.max_events = max_events
        self._records = {}

    def observe(self, timestamp, payload):
        uint(timestamp, 0xFFFFFFFF)
        report = parse_event(payload)
        key = (timestamp, report['event'])
        old = self._records.get(key)
        if old is None and len(self._records) >= self.max_events:
            raise ValueError('capacity reached; no silent eviction')
        duration = report['duration_ticks']
        completion = None
        if old is not None and old['ended']:
            if duration > old['duration_ticks'] or (report['ended'] and duration != old['duration_ticks']):
                raise ValueError('duration conflicts with retained final report')
            status = 'replayed_end' if report['ended'] else 'late_after_end'
        elif old is not None and duration < old['duration_ticks']:
            status = 'stale'
        else:
            status = 'duplicate_progress' if old is not None and duration == old['duration_ticks'] and not report['ended'] else 'updated'
            replacement = {'duration_ticks': duration, 'ended': report['ended']}
            if report['ended']:
                status = 'ended'
                completion = {'timestamp': timestamp, 'event': report['event'],
                              'digit': report['digit'], 'duration_ticks': duration}
            self._records[key] = replacement
        return {'status': status, 'new_completion': completion}

    def snapshot(self):
        return [{'timestamp': timestamp, 'event': event, **record}
                for (timestamp, event), record in sorted(self._records.items())]

    def reset(self):
        self._records.clear()

    def seconds(self, duration_ticks):
        uint(duration_ticks, 65535)
        return Fraction(duration_ticks, self.clock_hz)


def packet(event=5, duration=80, *, ended=False, reserved=False, volume=10):
    # Synthetic fixture helper, not a negotiated RTP sender.
    return bytes([event, (0x80 if ended else 0) | (0x40 if reserved else 0) | volume,
                  duration // 256, duration % 256])


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


def run_example():
    raw = bytes.fromhex('058a00f0')
    hand = {'event': 5, 'digit': '5', 'ended': True, 'reserved_seen': False,
            'volume_code': 10, 'duration_ticks': 240}
    ledger = EventLedger()
    stream = [packet(duration=80), packet(duration=160), raw, raw, raw]
    result = [ledger.observe(1000, payload) for payload in stream]
    expected_completion = {'timestamp': 1000, 'event': 5, 'digit': '5', 'duration_ticks': 240}
    stale = EventLedger();stale.observe(1000, packet(duration=160))
    stale_result = stale.observe(1000, packet(duration=80, ended=True))
    final = EventLedger();final.observe(1000, raw);final_before = final.snapshot()
    conflicts = all(rejects(lambda p=p: final.observe(1000, p)) for p in [packet(duration=241), packet(duration=200, ended=True)])
    capacity = EventLedger(max_events=1);capacity.observe(1, raw);capacity_before = capacity.snapshot()
    cap_reject = rejects(lambda: capacity.observe(2, raw))
    oracle_cases = []
    for event in range(16):
        for duration in [1, 255, 256, 65535]:
            for flags in [0, 0x80, 0x40, 0xC0]:
                p = bytes([event, flags | 17, duration // 256, duration % 256])
                decoded = parse_event(p)
                # Independent arithmetic extraction; no struct or parse helper.
                oracle = {'event': p[0], 'digit': DIGITS[p[0]], 'ended': p[1] // 128 == 1,
                          'reserved_seen': (p[1] // 64) % 2 == 1,
                          'volume_code': p[1] % 64, 'duration_ticks': p[2] * 256 + p[3]}
                oracle_cases.append(decoded == oracle)
    same_code = EventLedger();same_code.observe(1, raw);same_code.observe(2, raw)
    same_stamp = EventLedger();same_stamp.observe(1, raw);same_stamp.observe(1, packet(event=6, duration=240, ended=True))
    reset = EventLedger();reset.observe(1, raw);reset.reset()
    copy_test = EventLedger();copy_test.observe(1, raw);snapshot = copy_test.snapshot();snapshot[0]['duration_ticks'] = 9
    checks = {
        'manual_four_byte_fixture': parse_event(raw) == hand,
        'network_order_duration': parse_event(bytes.fromhex('010a0102'))['duration_ticks'] == 258,
        'end_flag_does_not_change_volume': parse_event(packet(ended=True, volume=63))['volume_code'] == 63 and parse_event(packet(volume=0))['volume_code'] == 0,
        'reserved_receiver_bit_is_ignored_semantically': parse_event(packet(reserved=True))['reserved_seen'] is True and EventLedger().observe(1, packet(reserved=True, ended=True))['status'] == 'ended',
        'dtmf_symbol_mapping': [parse_event(packet(event=i))['digit'] for i in range(16)] == list('0123456789*#ABCD'),
        'payload_type_and_single_record_length': all(rejects(lambda p=p: parse_event(p)) for p in [b'', b'123', b'12345', raw * 2, bytearray(raw), '058a00f0']),
        'positive_duration_dtmf_profile': rejects(lambda: parse_event(packet(duration=0))) and rejects(lambda: parse_event(packet(event=16))),
        'cumulative_duration_and_one_completion': [x['status'] for x in result] == ['updated', 'updated', 'ended', 'replayed_end', 'replayed_end'] and [x['new_completion'] for x in result if x['new_completion'] is not None] == [expected_completion],
        'stale_shorter_report_does_not_close_event': stale_result == {'status': 'stale', 'new_completion': None} and stale.snapshot()[0] == {'timestamp': 1000, 'event': 5, 'duration_ticks': 160, 'ended': False},
        'completed_duration_conflicts_leave_state_unchanged': conflicts and final.snapshot() == final_before,
        'late_nonend_after_finish_is_ignored': final.observe(1000, packet(duration=80)) == {'status': 'late_after_end', 'new_completion': None} and final.snapshot() == final_before,
        'same_digit_new_timestamp_is_new_identity': len(same_code.snapshot()) == 2,
        'same_timestamp_different_event_is_new_identity': len(same_stamp.snapshot()) == 2,
        'capacity_rejects_without_eviction': cap_reject and capacity.snapshot() == capacity_before,
        'clock_units_are_explicit': ledger.seconds(240) == Fraction(3, 100) and EventLedger(16000).seconds(240) == Fraction(3, 200),
        'timestamp_clock_capacity_types_bounds': all(rejects(fn) for fn in [lambda: EventLedger(True), lambda: EventLedger(0), lambda: EventLedger(max_events=0), lambda: EventLedger(max_events=9), lambda: ledger.observe(True, raw), lambda: ledger.observe(0x100000000, raw)]),
        'independent_arithmetic_byte_oracle': all(oracle_cases),
        'snapshot_copy_and_explicit_reset_scope': copy_test.snapshot()[0]['duration_ticks'] == 240 and reset.snapshot() == [] and reset.observe(1, raw)['new_completion'] is not None,
    }
    assert len(checks) == 18 and all(type(x) is bool and x for x in checks.values()), checks
    return {'python': platform.python_version(), 'dependencies': 'Python standard library only',
            'example_kind': 'Synthetic single-stream single-epoch single-event DTMF profile; no RTP/audio/network receiver',
            'manual_payload_hex': raw.hex(), 'decoded_manual': parse_event(raw),
            'stream_results': result, 'final_records': ledger.snapshot(),
            'manual_duration_ms_at_8000_hz': 30,
            'independent_byte_cases_inside_one_group': len(oracle_cases), 'checks': checks}


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

جواب دستیِ بایت‌های 05 8A 00 F0 این است: رویداد پنج، E روشن، R خاموش، کد حجم ده و مدت ۲۴۰. جریان نمونه ابتدا مدت‌های هشتاد و صدوشصت، سپس سه گزارشِ پایان با مدت ۲۴۰ دارد. خروجی وضعیت‌ها updated، updated، ended و دو replayed_end است؛ فقط یک new_completion غیرتهی و یک هویت نهایی باقی می‌ماند.

مرجع مستقل با ضربِ بایت بالای مدت در ۲۵۶ و جمع بایت پایین، و تقسیم و باقیمانده برای بیت‌ها ساخته شده است؛ struct یا تابع اصلی را فراخوانی نمی‌کند. ۲۵۶ ترکیبِ کد، مدت و پرچم در یک گروه بررسی می‌شوند. هجده گروه نام‌دار داریم؛ جواب دستی و مسیر پایانِ تکراری جدا از این جدول‌اند.

۵. مرز پروفایل آموزشی و چهار پاسخ

خروجی snapshot نسخه‌ای تازه از رکوردهای اسکالر می‌دهد، ولی کلاس، مرز احراز هویت یا ساختار هم‌زمانِ thread-safe نیست. reset شاهدهای تکرار را پاک می‌کند و دریافت همان پایان پس از آن دوباره رکورد می‌سازد. پس خاصیت محلیِ «یک پایان تا وقتی کلید نگه داشته شده» تضمین اجرای یک‌بارهٔ اقدام، نگهداری پایدار یا تحویل شبکه نیست.

{
  "python": "3.14.5",
  "dependencies": "Python standard library only",
  "example_kind": "Synthetic single-stream single-epoch single-event DTMF profile; no RTP/audio/network receiver",
  "manual_payload_hex": "058a00f0",
  "decoded_manual": {
    "event": 5,
    "digit": "5",
    "ended": true,
    "reserved_seen": false,
    "volume_code": 10,
    "duration_ticks": 240
  },
  "stream_results": [
    {
      "status": "updated",
      "new_completion": null
    },
    {
      "status": "updated",
      "new_completion": null
    },
    {
      "status": "ended",
      "new_completion": {
        "timestamp": 1000,
        "event": 5,
        "digit": "5",
        "duration_ticks": 240
      }
    },
    {
      "status": "replayed_end",
      "new_completion": null
    },
    {
      "status": "replayed_end",
      "new_completion": null
    }
  ],
  "final_records": [
    {
      "timestamp": 1000,
      "event": 5,
      "duration_ticks": 240,
      "ended": true
    }
  ],
  "manual_duration_ms_at_8000_hz": 30,
  "independent_byte_cases_inside_one_group": 256,
  "checks": {
    "manual_four_byte_fixture": true,
    "network_order_duration": true,
    "end_flag_does_not_change_volume": true,
    "reserved_receiver_bit_is_ignored_semantically": true,
    "dtmf_symbol_mapping": true,
    "payload_type_and_single_record_length": true,
    "positive_duration_dtmf_profile": true,
    "cumulative_duration_and_one_completion": true,
    "stale_shorter_report_does_not_close_event": true,
    "completed_duration_conflicts_leave_state_unchanged": true,
    "late_nonend_after_finish_is_ignored": true,
    "same_digit_new_timestamp_is_new_identity": true,
    "same_timestamp_different_event_is_new_identity": true,
    "capacity_rejects_without_eviction": true,
    "clock_units_are_explicit": true,
    "timestamp_clock_capacity_types_bounds": true,
    "independent_arithmetic_byte_oracle": true,
    "snapshot_copy_and_explicit_reset_scope": true
  }
}

منابع در ۲۵ مهر ۱۴۰۵ خوانده شدند؛ RFC مربوط به دسامبر ۲۰۰۶ است. نسخهٔ مستندات پایتون 3.14.8 از اجرای 3.14.5 جدا ثبت شده است. داده‌ها تألیفی‌اند؛ هیچ نرخ موفقیت، جیتر، تأخیر یا کیفیت گیرندهٔ واقعی از این تمرین به دست نیامده و این مقاله خبر تازه نیست.

پرسش اول: سه بستهٔ E با یک timestamp یعنی سه کلید؟ پاسخ: خیر؛ در نمونه همان هویت و مدت‌اند و فقط یک پایان تازه ثبت می‌شود.

پرسش دوم: ۲۴۰ همیشه سی میلی‌ثانیه است؟ پاسخ: خیر؛ فقط در ساعت ۸۰۰۰ هرتز. واحد اصلی تیک است و ساعت باید معلوم باشد.

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

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

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

• RFC4733؛ ساختار telephone-event و پایان تکراری
• Python؛ struct، طول دقیق و ترتیب بایت شبکه

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