نگاشت ساعت RTP با گزارش RTCP؛ دو مرجع زمانی و کسر دقیق در پایتون

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

دو ساعتِ گزارش فرستنده را از زمان ورود جدا کنیم

در تعریف گزارش فرستنده در RFC 3550، زمان NTP و timestamp رسانه به یک لحظهٔ متناظر اشاره دارند. مقدار رسانه همان واحد و مبدأ تصادفی جریان را دارد؛ لزوماً برابر timestamp بستهٔ دادهٔ مجاور نیست. این جفت، دو مشاهدهٔ زمان ورود در گیرنده محسوب نمی‌شود. کد حاضر فقط همین فیلدهای ازپیش‌استخراج‌شده را می‌گیرد.

دو رکورد با SSRC ساختگی ۱۷ داریم. مرجع اول در ثانیهٔ ۱۰۰۰ و مرجع دوم در ثانیهٔ ۱۰۰۱ قرار گرفته است. timestamp اول برابر ۴۲۹۴۹۶۳۲۹۶، یعنی چهار هزار واحد پیش از پایان چرخهٔ ۳۲بیتی است؛ timestamp دوم ۴۰۰۸ است. عدد کوچک‌ترِ دوم، در این سناریو نشانهٔ عقب رفتن ساعت نیست: شمارنده یک بار از مرز عبور کرده است.

استخراج بایت‌ها مسئولیت دیگری دارد؛ خواندن سرآیند RTP و مرز افزونه و padding آن قرارداد ساختاری را توضیح می‌دهد. اینجا نه RTP از RTCP تشخیص داده می‌شود و نه ساختار یک بستهٔ SR اعتبارسنجی می‌شود. ورودی، چهار عددِ نام‌دار برای هر رکورد است؛ داشتن آن‌ها اصالت منبع یا ادامه داشتن همان ساعت را ثابت نمی‌کند.

ثانیه و کسرِ NTP را دقیق بخوانیم

طبق بخش انواع داده در RFC 5905، قالب timestamp دارای بخش ثانیهٔ ۳۲بیتی و بخش کسری ۳۲بیتی است؛ اطلاعات era برای تفسیر کامل تاریخ از بیرون لازم می‌شود. در برنامه، عدد ثانیه را با کسرِ «کلمهٔ کسری تقسیم بر دو به توان ۳۲» جمع می‌کنیم. کلمهٔ کسریِ نصف چرخه، دقیقاً نیم ثانیه است، نه نیم میلی‌ثانیه.

از Fraction استفاده می‌کنیم تا تبدیلِ این عددهای صحیح به ثانیه، کسری باقی بماند. کوچک‌ترین گامِ نمایش، یک تقسیم بر دو به توان ۳۲ ثانیه است؛ این گام، گواه دقت فیزیکی ساعت نیست. برای نمونه، افزودن کلمهٔ کسریِ یک به ثانیهٔ ۱۰۰۰ در آزمون بررسی می‌شود، اما هیچ عدم‌قطعیت اندازه‌گیری از آن استخراج نمی‌کنیم.

ثانیهٔ ۱۰۰۰ در این fixture فقط روی یک مرجع انتخاب‌شده معنا دارد؛ آن را تاریخ امروز یا ساعت تهران نمی‌نامیم. در قرارداد درس، هر دو رکورد به یک era و یک مرجع تعلق دارند و عبور era باز نمی‌شود. زمان تمام‌صفر نیز ناموجود تلقی و رد می‌شود. متن RFC 3550 امکان مرجع نسبی یا NTP صفر را مطرح می‌کند؛ انتخاب ما رد کردن حالتِ بدون مرجع قابل استفاده است.

بین دو مرجع زمانی یک نگاشت محدود بسازیم

فاصلهٔ رو‌به‌جلوی رسانه از تفاضل timestampها به پیمانهٔ دو به توان ۳۲ به دست می‌آید: چهار هزار واحد تا مرز و ۴۰۰۸ واحد پس از آن، جمعاً ۸۰۰۸ tick. فاصلهٔ مرجع یک ثانیه است؛ پس شیبِ این دو نقطه ۸۰۰۸ tick بر ثانیه می‌شود. نرخ اسمیِ ورودی را ۸۰۰۰ انتخاب کرده‌ایم؛ تفاوت نسبی این دو نرخ، هزار قسمت در میلیون است.

برای پرس‌وجوی timestamp برابر چهار، فاصله از مرجع اول ۴۰۰۴ tick است. این فاصله دقیقاً نصف ۸۰۰۸ است؛ بنابراین جواب، نصف فاصلهٔ یک‌ثانیه‌ای پس از ۱۰۰۰، یعنی ۱۰۰۰٫۵ ثانیه خواهد بود. فرمول نگاشت، زمان اول به اضافهٔ حاصل‌ضربِ سهمِ فاصلهٔ رسانه در فاصلهٔ دو مرجع است. مقدار tick با ثانیه جمع نمی‌شود؛ ابتدا نسبتِ بدون واحد ساخته می‌شود.

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

کران شصت ثانیه برای فاصلهٔ مثبتِ NTP و بازهٔ یک تا ۱۹۲۰۰۰ برای نرخ اسمی، سیاست محدود همین تمرین‌اند؛ الزام عمومی RTCP نیستند. نرخ اسمی را فراخواننده از قرارداد جریان می‌دهد و از تعداد بسته‌ها حدس نمی‌زنیم. نرخ ساعت رسانه نیز بدون بررسی قالب، برابر نرخ نمونه‌برداری فایل صوت فرض نمی‌شود. هیچ نرخ اسمیِ جهانی برای همهٔ payloadها پیشنهاد نشده است.

کد را با خط زمانیِ مستقل بسنجیم

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

"""Two synthetic RTCP SR clock anchors; no network or calendar conversion."""
from dataclasses import dataclass, replace
from fractions import Fraction
import json
import platform

MOD = 1 << 32
MAX_INTERVAL_SECONDS = 60  # Own teaching contract, not an RTCP limit.


@dataclass(frozen=True)
class SenderReport:
    ssrc: int
    ntp_seconds: int
    ntp_fraction: int
    rtp_timestamp: int


def uint32(value, name):
    if type(value) is not int or not 0 <= value < MOD:
        raise ValueError(f"{name} must be an exact uint32 integer")
    return value


def report_time(report):
    if type(report) is not SenderReport:
        raise ValueError("report must be exactly SenderReport")
    for name in ("ssrc", "ntp_seconds", "ntp_fraction", "rtp_timestamp"):
        uint32(getattr(report, name), name)
    if report.ntp_seconds == report.ntp_fraction == 0:
        raise ValueError("unavailable all-zero NTP timestamp")
    return Fraction(report.ntp_seconds * MOD + report.ntp_fraction, MOD)


def map_sender_clock(first, second, query_rtp, nominal_rate):
    """Caller guarantees one continuing source, era, and true RTP step < 2**31.

    Exact arithmetic does not validate clock synchronization or that guarantee.
    Returned time is in seconds on the supplied sender reference, not receiver time.
    """
    t0, t1 = report_time(first), report_time(second)
    uint32(query_rtp, "query_rtp")
    if type(nominal_rate) is not int or not 1 <= nominal_rate <= 192000:
        raise ValueError("nominal_rate must be an integer in 1..192000 tick/s")
    if first.ssrc != second.ssrc:
        raise ValueError("different sources")
    duration = t1 - t0
    if not 0 < duration <= MAX_INTERVAL_SECONDS:
        raise ValueError("need increasing NTP within one era and at most 60 s")
    step = (second.rtp_timestamp - first.rtp_timestamp) % MOD
    if not 0 < step < MOD // 2:
        raise ValueError("ambiguous, absent or backward forward-RTP interval")
    offset = (query_rtp - first.rtp_timestamp) % MOD
    if offset > step:
        raise ValueError("query outside anchor interval; no extrapolation")
    rate = Fraction(step, 1) / duration
    mapped_time = t0 + Fraction(offset, step) * duration
    rate_difference_ppm = (rate / nominal_rate - 1) * 1_000_000
    return {
        "mapped_seconds": mapped_time,
        "observed_ticks_per_second": rate,
        "relative_rate_difference_ppm": rate_difference_ppm,
        "anchor_ticks": step,
        "query_offset_ticks": offset,
    }


def encoded_report(seconds, unwrapped_rtp, ssrc=17):
    """Test fixture encoder, independent of the mapper/decoder."""
    whole = seconds.numerator // seconds.denominator
    fraction_word = (seconds - whole) * MOD
    if fraction_word.denominator != 1:
        raise ValueError("fixture time is not exactly representable")
    return SenderReport(ssrc, whole, fraction_word.numerator, unwrapped_rtp % MOD)


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


def run():
    first = SenderReport(17, 1000, 0, MOD - 4000)
    second = SenderReport(17, 1001, 0, 4008)
    main = map_sender_clock(first, second, 4, 8000)
    checks = {}
    checks["hand_wrapped_midpoint_rate_and_ppm"] = (
        main["mapped_seconds"] == Fraction(2001, 2)
        and main["anchor_ticks"] == 8008
        and main["query_offset_ticks"] == 4004
        and main["observed_ticks_per_second"] == 8008
        and main["relative_rate_difference_ppm"] == 1000
    )
    checks["both_anchor_endpoints"] = all(
        map_sender_clock(first, second, r.rtp_timestamp, 8000)["mapped_seconds"] == t
        for r, t in ((first, Fraction(1000)), (second, Fraction(1001)))
    )
    checks["fraction_word_resolution_not_clock_accuracy"] = (
        report_time(SenderReport(17, 1000, 1, 0)) - Fraction(1000) == Fraction(1, MOD)
        and report_time(SenderReport(17, 1000, MOD // 2, 0)) == Fraction(2001, 2)
    )
    # Independent expected answers come from known unwrapped clocks BEFORE encoding.
    oracle_cases = []
    origin_time = Fraction(4001, 4)
    for known_rate in (8000, 8008, 16000, 48000):
        for duration in (Fraction(1, 4), Fraction(1, 2), Fraction(1)):
            tick_step = known_rate * duration
            for origin_rtp in (123, MOD - 2000):
                a = encoded_report(origin_time, origin_rtp)
                b = encoded_report(origin_time + duration, origin_rtp + int(tick_step))
                for portion in (Fraction(0), Fraction(1, 2), Fraction(1)):
                    known_time = origin_time + portion * duration
                    known_query = origin_rtp + int(portion * tick_step)
                    got = map_sender_clock(a, b, known_query % MOD, 8000)
                    oracle_cases.append(
                        got["mapped_seconds"] == known_time
                        and got["observed_ticks_per_second"] == known_rate
                        and got["relative_rate_difference_ppm"]
                        == Fraction(known_rate - 8000, 8000) * 1_000_000
                    )
    checks["generated_known_clock_oracle_72_queries_one_group"] = all(oracle_cases)
    nominal_only = Fraction(1000) + Fraction(4004, 8000)
    checks["nominal_rate_mismatch_half_millisecond_witness"] = (
        nominal_only == Fraction(2001001, 2000)
        and main["mapped_seconds"] - nominal_only == Fraction(-1, 2000)
    )
    shifted_a = replace(first, ntp_seconds=1234, rtp_timestamp=first.rtp_timestamp + 100)
    shifted_b = replace(second, ntp_seconds=1235, rtp_timestamp=second.rtp_timestamp + 100)
    shifted = map_sender_clock(shifted_a, shifted_b, 104, 8000)
    checks["common_clock_origin_translation"] = (
        shifted["mapped_seconds"] == main["mapped_seconds"] + 234
        and shifted["observed_ticks_per_second"] == main["observed_ticks_per_second"]
    )
    checks["outside_queries_rejected_both_sides"] = all(
        raises_value(lambda q=q: map_sender_clock(first, second, q, 8000))
        for q in (MOD - 4001, 4009)
    )
    checks["source_mismatch_rejected"] = raises_value(
        lambda: map_sender_clock(first, replace(second, ssrc=18), 4, 8000)
    )
    checks["zero_equal_decreasing_and_era_crossing_ntp_rejected"] = all(
        raises_value(lambda a=a, b=b: map_sender_clock(a, b, b.rtp_timestamp, 8000))
        for a, b in (
            (replace(first, ntp_seconds=0), second),
            (first, replace(second, ntp_seconds=1000)),
            (first, replace(second, ntp_seconds=999)),
            (replace(first, ntp_seconds=MOD - 1), replace(second, ntp_seconds=1)),
        )
    )
    checks["zero_backward_and_half_cycle_rtp_rejected"] = all(
        raises_value(lambda r=r: map_sender_clock(first, replace(second, rtp_timestamp=r), r, 8000))
        for r in (first.rtp_timestamp, (first.rtp_timestamp - 1) % MOD,
                  (first.rtp_timestamp + MOD // 2) % MOD)
    )
    invalid_fields = []
    for name in ("ssrc", "ntp_seconds", "ntp_fraction", "rtp_timestamp"):
        for bad in (True, 1.0, -1, MOD):
            invalid_fields.append(raises_value(
                lambda name=name, bad=bad: map_sender_clock(replace(first, **{name: bad}), second, 4, 8000)
            ))
    checks["field_types_ranges_and_report_type_rejected"] = (
        all(invalid_fields)
        and raises_value(lambda: map_sender_clock((17, 1000, 0, 0), second, 4, 8000))
        and all(raises_value(lambda q=q: map_sender_clock(first, second, q, 8000))
                for q in (True, 4.0, -1, MOD))
    )
    checks["nominal_rate_contract_rejected"] = all(
        raises_value(lambda rate=rate: map_sender_clock(first, second, 4, rate))
        for rate in (True, 8000.0, 0, -1, 192001)
    )
    sixty = SenderReport(17, 1060, 0, (first.rtp_timestamp + 480000) % MOD)
    checks["sixty_second_boundary_and_sixty_one_rejection"] = (
        map_sender_clock(first, sixty, sixty.rtp_timestamp, 8000)["mapped_seconds"] == 1060
        and raises_value(lambda: map_sender_clock(first, replace(sixty, ntp_seconds=1061), sixty.rtp_timestamp, 8000))
    )
    checks["input_records_unchanged"] = (
        first == SenderReport(17, 1000, 0, MOD - 4000)
        and second == SenderReport(17, 1001, 0, 4008)
    )
    if not all(type(v) is bool and v for v in checks.values()):
        raise AssertionError(checks)
    def ratio(value):
        if isinstance(value, Fraction):
            return {"numerator": value.numerator, "denominator": value.denominator}
        return value
    return {
        "python": platform.python_version(),
        "dependencies": "standard library only",
        "case_kind": "synthetic extracted sender fields, same chosen NTP era/reference",
        "first_report": first.__dict__, "second_report": second.__dict__,
        "query_rtp_timestamp": 4, "nominal_ticks_per_second": 8000,
        "result": {k: ratio(v) for k, v in main.items()},
        "nominal_only_mapped_seconds": ratio(nominal_only),
        "affine_minus_nominal_seconds": ratio(main["mapped_seconds"] - nominal_only),
        "oracle_query_count_within_one_group": len(oracle_cases),
        "checks": checks, "check_count": len(checks), "all_passed": True,
        "limitations": ["No actual clock, arrival, latency, jitter, NTP synchronization or packet parser",
                        "True forward RTP step under half a cycle is a caller assumption",
                        "No source reset, extrapolation or NTP era unwrapping"],
    }


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

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

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

{
  "python": "3.14.5",
  "dependencies": "standard library only",
  "case_kind": "synthetic extracted sender fields, same chosen NTP era/reference",
  "first_report": {
    "ssrc": 17,
    "ntp_seconds": 1000,
    "ntp_fraction": 0,
    "rtp_timestamp": 4294963296
  },
  "second_report": {
    "ssrc": 17,
    "ntp_seconds": 1001,
    "ntp_fraction": 0,
    "rtp_timestamp": 4008
  },
  "query_rtp_timestamp": 4,
  "nominal_ticks_per_second": 8000,
  "result": {
    "mapped_seconds": {
      "numerator": 2001,
      "denominator": 2
    },
    "observed_ticks_per_second": {
      "numerator": 8008,
      "denominator": 1
    },
    "relative_rate_difference_ppm": {
      "numerator": 1000,
      "denominator": 1
    },
    "anchor_ticks": 8008,
    "query_offset_ticks": 4004
  },
  "nominal_only_mapped_seconds": {
    "numerator": 2001001,
    "denominator": 2000
  },
  "affine_minus_nominal_seconds": {
    "numerator": -1,
    "denominator": 2000
  },
  "oracle_query_count_within_one_group": 72,
  "checks": {
    "hand_wrapped_midpoint_rate_and_ppm": true,
    "both_anchor_endpoints": true,
    "fraction_word_resolution_not_clock_accuracy": true,
    "generated_known_clock_oracle_72_queries_one_group": true,
    "nominal_rate_mismatch_half_millisecond_witness": true,
    "common_clock_origin_translation": true,
    "outside_queries_rejected_both_sides": true,
    "source_mismatch_rejected": true,
    "zero_equal_decreasing_and_era_crossing_ntp_rejected": true,
    "zero_backward_and_half_cycle_rtp_rejected": true,
    "field_types_ranges_and_report_type_rejected": true,
    "nominal_rate_contract_rejected": true,
    "sixty_second_boundary_and_sixty_one_rejection": true,
    "input_records_unchanged": true
  },
  "check_count": 14,
  "all_passed": true,
  "limitations": [
    "No actual clock, arrival, latency, jitter, NTP synchronization or packet parser",
    "True forward RTP step under half a cycle is a caller assumption",
    "No source reset, extrapolation or NTP era unwrapping"
  ]
}

اگر فقط فاصلهٔ ۴۰۰۴ را بر نرخ اسمیِ ۸۰۰۰ تقسیم کنیم، زمان ۱۰۰۰٫۵۰۰۵ به دست می‌آید؛ نیم میلی‌ثانیه دیرتر از نگاشت دو مرجعِ ساختگی. علامت اختلافِ «نگاشت منهای روش اسمی» منفی است. این شاهد توضیح می‌دهد چرا مبدأ و شیب باید جدا دیده شوند؛ نیم میلی‌ثانیه را خطای اندازه‌گیری یک دستگاه یا بهبود همگام‌سازی واقعی گزارش نمی‌کنیم.

نگاشتِ ساعت را تأخیر یا دقت همگام‌سازی ننامیم

پرسش اول: چرا timestamp چهار نزدیک مرجع اول است؟ پاسخ: پس از ۴۲۹۴۹۶۳۲۹۶، شمارنده تا پایان چرخه چهار هزار واحد و سپس چهار واحد جلو رفته است؛ فاصله ۴۰۰۴ است. بزرگیِ تفاضل معمولیِ بدون پیمانه، فاصلهٔ این سناریوی رو‌به‌جلو را نشان نمی‌دهد.

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

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

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

منابع در ۲۵ مهر ۱۴۰۵ خوانده شدند؛ تاریخ انتشار RFCها به‌ترتیب ژوئیهٔ ۲۰۰۳ و ژوئن ۲۰۱۰ است و روز انتشار در آن‌ها مشخص نشده است. این مطلب آموزش پایهٔ تألیفی است، نه خبر تازه یا آزمون تجهیزات. برای استفادهٔ عملی، مدیریت era، بازنشانی و تغییر منبع، صحت داده و رفتار ساعت باید جدا طراحی و آزمایش شوند.

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

• RFC 3550؛ دو مرجع متناظر در گزارش فرستنده و قرارداد زمان RTP
• RFC 5905؛ ثانیه، بخش کسری و era در قالب timestamp

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