میانهٔ پنجرهٔ زمانی حسگر در پایتون؛ مرز باز، دادهٔ هم‌زمان و آزمون مستقل

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

۱. پنجرهٔ زمانی را با مرز صریح تعریف کنیم

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

راهنمای rolling در pandas تفاوت پنجرهٔ تعدادمحور و زمانی و نقش مرز closed را توضیح می‌دهد. از این تفکیک برای تعریف تمرین استفاده می‌کنیم؛ pandas را اجرا نمی‌کنیم و API زمان تقویمی آن را بازسازی نمی‌کنیم. زمان ما میلی‌ثانیهٔ صحیح از مبدأ فرضی صفر است؛ تاریخ شمسی، زمان تهران یا UTC نیست.

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

۲. میانه، وزن مشاهده و حداقل تعداد را جدا ببینیم

در تعریف میانهٔ عددی پایتون، برای تعداد فرد عضو میانی و برای تعداد زوج میانگین دو عضو میانی انتخاب می‌شود. بنابراین میانه همیشه یکی از اندازه‌گیری‌های ورودی نیست. برای یک و یک‌ویک‌چهارم ولت، میانگین زوج مرکزی ۱٫۱۲۵ ولت است. بهتر است پاسخ عددی را مستقیماً کنار نمونه‌های معلوم ثبت کنیم تا انتخاب عضو پایین‌تر به‌جای میانگین، اشتباه پنهان نماند.

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

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

۳. کد کامل، صف زمانی و مجموعهٔ مرتب مقدارها

فایل time_window_median.py با Python 3.14.5 اجرا شده است. داده باید list یا tuple از زوج زمان و ولتاژ باشد. زمان صحیح نامنفی و ولتاژ int یا float دقیق، متناهی و در بازهٔ بستهٔ منفی پنج تا مثبت پنج ولت است. bool، NaN، بی‌نهایت، رشته و زمانِ رو به عقب رد می‌شوند. پهنا و حداقل تعداد، صحیح مثبت‌اند و واحد فقط V پذیرفته می‌شود.

"""Finite offline prefix windows (t-W,t], exact integer ms, synthetic [-5,5] V."""
from bisect import bisect_left, insort
from collections import deque
from fractions import Fraction
import json
import math
import platform


def validate(events, width_ms, min_count, unit):
    if type(width_ms) is not int or width_ms <= 0:
        raise ValueError('width_ms_positive_exact_int')
    if type(min_count) is not int or min_count <= 0:
        raise ValueError('min_count_positive_exact_int')
    if type(unit) is not str or unit != 'V':
        raise ValueError('unit_must_be_V')
    if type(events) not in (list, tuple):
        raise ValueError('events_list_or_tuple')
    accepted, previous = [], -1
    for event in events:
        if type(event) not in (list, tuple) or len(event) != 2:
            raise ValueError('event_pair')
        at, voltage = event
        if type(at) is not int or at < 0 or at < previous:
            raise ValueError('nondecreasing_nonnegative_integer_ms')
        if type(voltage) not in (int, float):
            raise ValueError('voltage_type')
        # Compare integer magnitude before float conversion.
        if not -5 <= voltage <= 5 or not math.isfinite(float(voltage)):
            raise ValueError('finite_voltage_within_closed_minus5_plus5_V')
        accepted.append((at, float(voltage)))
        previous = at
    return tuple(accepted)


def rolling_median(events, width_ms, min_count=1, unit='V'):
    accepted = validate(events, width_ms, min_count, unit)
    recent, ordered, output = deque(), [], []
    for position, (at, value) in enumerate(accepted):
        left = at - width_ms
        while recent and recent[0][0] <= left:
            _, old = recent.popleft()
            index = bisect_left(ordered, old)
            if index == len(ordered) or ordered[index] != old:
                raise AssertionError('queue_and_multiset_disagree')
            ordered.pop(index)  # Exactly one occurrence, not all equal values.
        recent.append((at, value))
        insort(ordered, value)
        count, middle = len(ordered), len(ordered) // 2
        if count < min_count:
            median = None
        elif count % 2:
            median = ordered[middle]
        else:
            # The validated +/-5 V range keeps this sum finite; float rounding remains.
            median = (ordered[middle - 1] + ordered[middle]) / 2
        output.append({'position': position, 'time_ms': at, 'count': count, 'median_V': median})
    return output


def independent_prefix_reference(events, width_ms, min_count):
    """Full prefix scan plus sorted exact rational rank; no deque/bisect/main calls."""
    output = []
    for position, (at, _) in enumerate(events):
        selected = sorted(Fraction.from_float(float(value))
                          for earlier_at, value in events[:position + 1]
                          if at - width_ms < earlier_at <= at)
        count, middle = len(selected), len(selected) // 2
        if count < min_count:
            value = None
        elif count % 2:
            value = float(selected[middle])
        else:
            value = float((selected[middle - 1] + selected[middle]) / 2)
        output.append({'position': position, 'time_ms': at, 'count': count, 'median_V': value})
    return output


def rejects(events, width_ms=10, min_count=1, unit='V'):
    try:
        rolling_median(events, width_ms, min_count, unit)
    except ValueError:
        return True
    return False


def run_example():
    events = [(0, 1), (4, 1.25), (9, 5), (10, 1.5), (10, 1.75),
              (14, 1.25), (21, 1), (40, 2)]
    result = rolling_median(events, 10, min_count=2)
    counts = [1, 2, 3, 3, 4, 4, 2, 1]
    medians = [None, 1.125, 1.25, 1.5, 1.625, 1.625, 1.125, None]
    reference_ok = True
    for case in range(48):
        example = [((index * index + case) // 3, ((7 * index + case) % 21 - 10) / 4)
                   for index in range(case % 12)]
        width, minimum = 1 + case % 9, 1 + case % 3
        reference_ok = reference_ok and rolling_median(example, width, minimum) == independent_prefix_reference(example, width, minimum)
    boundary = rolling_median([(0, 1), (10, 3)], 10)
    duplicate_value = rolling_median([(0, 1), (5, 1), (10, 2)], 10)
    base = 2 ** 60
    shifted = rolling_median([(at + base, value) for at, value in events], 10, 2)
    checks = {}
    checks['hand_eight_window_counts_and_medians'] = [r['count'] for r in result] == counts and [r['median_V'] for r in result] == medians
    checks['left_boundary_is_open_current_boundary_closed'] = boundary[-1]['count'] == 1 and boundary[-1]['median_V'] == 3.0
    checks['same_timestamp_is_prefix_order_not_future_group'] = [r['count'] for r in result[3:5]] == [3, 4] and [r['median_V'] for r in result[3:5]] == [1.5, 1.625]
    checks['minimum_observations_returns_none_not_zero'] = result[0]['median_V'] is None and result[-1]['median_V'] is None and result[1]['median_V'] is not None
    checks['minimum_one_allows_first_observation'] = rolling_median(events, 10)[0]['median_V'] == 1.0
    checks['long_gap_expires_all_old_values'] = result[-1]['count'] == 1 and rolling_median(events, 10)[-1]['median_V'] == 2.0
    checks['time_window_is_not_last_three_observations'] = result[6]['median_V'] == 1.125 and sorted(value for _, value in events[4:7])[1] == 1.25
    checks['forty_eight_finite_schedules_match_exact_prefix_reference'] = reference_ok
    checks['negative_odd_and_even_medians'] = [r['median_V'] for r in rolling_median([(0, -3), (1, -1), (2, -2)], 10)] == [-3.0, -2.0, -2.0]
    checks['expiry_removes_one_occurrence_of_duplicate_value'] = duplicate_value[-1]['count'] == 2 and duplicate_value[-1]['median_V'] == 1.5
    checks['adding_future_or_later_equal_time_cannot_change_saved_prefix'] = rolling_median(events[:4], 10, 2) == result[:4]
    checks['large_exact_integer_timestamp_translation_preserves_windows'] = [(r['count'], r['median_V']) for r in shifted] == [(r['count'], r['median_V']) for r in result]
    checks['closed_voltage_endpoints_and_rank_between_extremes'] = rolling_median([(0, -5), (1, 5)], 10)[-1]['median_V'] == 0.0 and all(r['median_V'] is None or -5 <= r['median_V'] <= 5 for r in result)
    checks['rejects_invalid_width_and_minimum_types'] = all(rejects(events, width_ms=v) and rejects(events, min_count=v) for v in (0, -1, True, 1.0, float('nan')))
    checks['rejects_time_type_negative_or_backwards_order'] = all(rejects([(v, 1)]) for v in (-1, True, 0.0, float('nan'), float('inf'), '0')) and rejects([(1, 1), (0, 2)])
    checks['rejects_nonfinite_nonnumeric_or_outside_voltage'] = all(rejects([(0, v)]) for v in (True, '1', None, 1j, float('nan'), float('inf'), -5.001, 5.001, 10 ** 400))
    checks['unit_shape_and_empty_contract'] = rejects(events, unit='mV') and rejects(events, unit=True) and rejects(iter(events)) and rejects([(0, 1, 2)]) and rolling_median([], 10) == []
    saved = list(events)
    checks['repeatable_tuple_and_list_without_mutation'] = rolling_median(tuple(events), 10, 2) == result and events == saved
    if len(checks) != 18 or not all(type(value) is bool and value for value in checks.values()):
        raise AssertionError(checks)
    return {'python_version': platform.python_version(), 'runtime': 'Python standard library only',
            'units': {'time': 'authored exact integer ms', 'voltage': 'V within closed[-5,5]'},
            'window_contract': 'per-arrival prefix with t-W < prior_timestamp <= t; same-time later observations unavailable',
            'width_ms': 10, 'min_count': 2, 'input_events': [list(event) for event in events],
            'hand_expected_counts': counts, 'hand_expected_medians_V': medians, 'result': result,
            'reference_schedules': 48, 'checks': checks, 'unique_checks': 18,
            'limitations': ['Every observation has equal rank weight, not duration weighting; duplicate times are separate observations.',
                            'No padding, future values, NaN omission, resampling, physical sensor, clock or noise-removal guarantee.',
                            'Maintained sorted-list insertion/deletion shifts elements; not O(log k) total updates, hard realtime or fixed-memory.',
                            'A fixed time width does not bound arrival density; the finite input/output lists also consume memory.',
                            'Floating-point midpoint rounding remains; original decimal measurements are not recovered.']}


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

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

برای درج و یافتن محل حذف از bisect استفاده می‌کنیم. جست‌وجوی دودویی، هزینهٔ جابه‌جایی اعضای list را حذف نمی‌کند؛ کل به‌روزرسانی را لگاریتمی معرفی نمی‌کنیم. صف و فهرستِ مقدارها در این برنامه ترتیبی‌اند و قفل یا تضمین هم‌زمانی ندارند. پهنای زمانی ثابت نیز در تراکم ورود دلخواه، سقف تعداد نمونه یا حافظه ایجاد نمی‌کند.

۴. هشت خروجی معلوم و مرجع مستقل پیشوند

زمان‌های دادهٔ اصلی صفر، چهار، نه، ده، ده، چهارده، بیست‌ویک و چهل میلی‌ثانیه‌اند. ولتاژها به ترتیب یک، ۱٫۲۵، پنج، ۱٫۵، ۱٫۷۵، ۱٫۲۵، یک و دو هستند. با پهنای ده و حداقل دو، تعداد اعضا یک، دو، سه، سه، چهار، چهار، دو و یک می‌شود. در مرز ده، حذف زمان صفر و افزودن جداگانهٔ دو رخداد هم‌زمان باید در پاسخ دیده شود.

{
  "python_version": "3.14.5",
  "runtime": "Python standard library only",
  "units": {
    "time": "authored exact integer ms",
    "voltage": "V within closed[-5,5]"
  },
  "window_contract": "per-arrival prefix with t-W < prior_timestamp <= t; same-time later observations unavailable",
  "width_ms": 10,
  "min_count": 2,
  "input_events": [
    [
      0,
      1
    ],
    [
      4,
      1.25
    ],
    [
      9,
      5
    ],
    [
      10,
      1.5
    ],
    [
      10,
      1.75
    ],
    [
      14,
      1.25
    ],
    [
      21,
      1
    ],
    [
      40,
      2
    ]
  ],
  "hand_expected_counts": [
    1,
    2,
    3,
    3,
    4,
    4,
    2,
    1
  ],
  "hand_expected_medians_V": [
    null,
    1.125,
    1.25,
    1.5,
    1.625,
    1.625,
    1.125,
    null
  ],
  "result": [
    {
      "position": 0,
      "time_ms": 0,
      "count": 1,
      "median_V": null
    },
    {
      "position": 1,
      "time_ms": 4,
      "count": 2,
      "median_V": 1.125
    },
    {
      "position": 2,
      "time_ms": 9,
      "count": 3,
      "median_V": 1.25
    },
    {
      "position": 3,
      "time_ms": 10,
      "count": 3,
      "median_V": 1.5
    },
    {
      "position": 4,
      "time_ms": 10,
      "count": 4,
      "median_V": 1.625
    },
    {
      "position": 5,
      "time_ms": 14,
      "count": 4,
      "median_V": 1.625
    },
    {
      "position": 6,
      "time_ms": 21,
      "count": 2,
      "median_V": 1.125
    },
    {
      "position": 7,
      "time_ms": 40,
      "count": 1,
      "median_V": null
    }
  ],
  "reference_schedules": 48,
  "checks": {
    "hand_eight_window_counts_and_medians": true,
    "left_boundary_is_open_current_boundary_closed": true,
    "same_timestamp_is_prefix_order_not_future_group": true,
    "minimum_observations_returns_none_not_zero": true,
    "minimum_one_allows_first_observation": true,
    "long_gap_expires_all_old_values": true,
    "time_window_is_not_last_three_observations": true,
    "forty_eight_finite_schedules_match_exact_prefix_reference": true,
    "negative_odd_and_even_medians": true,
    "expiry_removes_one_occurrence_of_duplicate_value": true,
    "adding_future_or_later_equal_time_cannot_change_saved_prefix": true,
    "large_exact_integer_timestamp_translation_preserves_windows": true,
    "closed_voltage_endpoints_and_rank_between_extremes": true,
    "rejects_invalid_width_and_minimum_types": true,
    "rejects_time_type_negative_or_backwards_order": true,
    "rejects_nonfinite_nonnumeric_or_outside_voltage": true,
    "unit_shape_and_empty_contract": true,
    "repeatable_tuple_and_list_without_mutation": true
  },
  "unique_checks": 18,
  "limitations": [
    "Every observation has equal rank weight, not duration weighting; duplicate times are separate observations.",
    "No padding, future values, NaN omission, resampling, physical sensor, clock or noise-removal guarantee.",
    "Maintained sorted-list insertion/deletion shifts elements; not O(log k) total updates, hard realtime or fixed-memory.",
    "A fixed time width does not bound arrival density; the finite input/output lists also consume memory.",
    "Floating-point midpoint rounding remains; original decimal measurements are not recovered."
  ]
}

میانه‌ها به ترتیب None، ۱٫۱۲۵، ۱٫۲۵، ۱٫۵، ۱٫۶۲۵، ۱٫۶۲۵، ۱٫۱۲۵ و None هستند. در زمان بیست‌ویک فقط مقدار زمان چهارده و مقدار جاری باقی‌اند؛ سه مشاهدهٔ آخر، مجموعهٔ دیگری می‌سازند. مقدار پنجِ انتخاب‌شده در این fixture رفتار رتبه را نشان می‌دهد، ولی خرابی حسگر یا نویز بودن آن ثابت نشده است. میانه به‌تنهایی تشخیص‌دهندهٔ خطا یا تضمین بهبود سیگنال نیست.

مرجع مستقل، هر پیشوند را کامل پیمایش و اعضای داخل بازه را با Fraction مرتب می‌کند؛ از deque، bisect یا تابع اصلی کمک نمی‌گیرد. روی ۴۸ دادهٔ محدود با مقدارهای ربع‌ولتی مقایسه انجام شده است؛ این‌ها یک گروه از هجده گروه بررسی‌اند، نه ۴۸ آزمون نام‌دار اضافه. محدودهٔ ورودی از سرریز جمع زوج مرکزی جلوگیری می‌کند، ولی گردکردن float یا دقت فیزیکی را حذف نمی‌کند.

آزمون RMS ولتاژ دربارهٔ دامنهٔ مؤثر است؛ جای رتبهٔ مرکزی را نمی‌گیرد. زمان‌بند قطعی رویداد نیز انتخاب ترتیب را بررسی می‌کند. اینجا پنجرهٔ مقدارها موضوع اصلی است. کل ورودی و خروجی در حافظه‌اند؛ این درس یک دریافت‌کنندهٔ زنده با تأخیر یا حافظهٔ تضمین‌شده نیست.

۵. چهار پرسش پاسخ‌دار برای عیب‌یابی

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

پرسش دوم: چرا دو خروجی با زمان ده فرق دارند؟ پاسخ: یکی پس از مشاهدهٔ اول و دیگری پس از مشاهدهٔ دوم ساخته شده است. شمار اعضا سه و چهار و میانه‌ها ۱٫۵ و ۱٫۶۲۵ ولت‌اند؛ زمان برابر، پیشوند برابر نمی‌سازد.

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

پرسش چهارم: هر دو مقدار یک ولت را هنگام انقضای یکی حذف کنیم؟ پاسخ: خیر؛ فقط همان رخداد خارج‌شده حذف می‌شود. نمونهٔ هم‌مقدارِ جدیدتر ممکن است هنوز عضو پنجره باشد. منابع در ۲۴ مهر ۱۴۰۵ خوانده شدند؛ نتایج هجده بررسی، تأیید حسگر واقعی یا اثبات همهٔ ورودی‌ها نیست.

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

• Python 3.14؛ تعریف میانهٔ عددی و زوج مرکزی
• pandas؛ پهنای زمانی پنجره و مرزهای باز و بسته
• Python 3.14؛ bisect و هزینهٔ درج در فهرست
• Python 3.14؛ صف دوطرفه deque

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