فهرست مقاله
دو کانال داده ممکن است جداگانه سالم به نظر برسند، اما هنگام جفت‌کردن، نمونهٔ انتهایی یکی از آن‌ها بی‌صدا حذف شود. در بازبینیِ این مسیر، برابر بودن طول و یکسان بودن هویت نمونه دو مسئولیت متفاوت‌اند. این درس با دو دنبالهٔ ساختگی، zip سخت‌گیرانه و یک مرجع جایگاهی نشان می‌دهد چه چیزی را می‌توان پیش از تحویل نتیجه بررسی کرد. هیچ حسگر، ارتباط شبکه یا ذخیره‌سازی واقعی در آزمایش وجود ندارد.
۱. دو کانال باید همان نمونه را نشان دهند
هر رکوردِ این مدل دو عضو دارد: شناسهٔ صحیح نمونه و مقدار صحیح بر حسب میلی‌ولت. شناسه‌ها صفر، یک و دو هستند. کانال چپ مقادیر ۱۰۰۰،۱۰۱۰،۱۰۲۰ و کانال راست ۹۹۰،۱۰۰۵،۱۰۱۰ میلی‌ولت دارد. انتظار داریم رکوردهای دارای همان شناسه با هم مقایسه شوند؛ اختلاف را مقدار چپ منهای راست می‌گیریم و علامت آن بخشی از قرارداد است.
خروجی دستی سه ردیف با اختلاف‌های ۱۰،۵،۱۰ میلی‌ولت است. جمع این اختلاف‌ها ۲۵ میلی‌ولت می‌شود؛ این جمع صرفاً کنترل حسابیِ fixture است، نه انتگرال زمانی، خطای کالیبراسیون یا ولتاژ یک قطعهٔ مشترک. زمان نمونه‌برداری تعریف نکرده‌ایم و از فاصلهٔ شناسه‌ها، نرخ دریافت یا هم‌زمانی فیزیکی نتیجه نمی‌گیریم. شناسه در اینجا کلید تطبیق نرم‌افزاری است.
در قرارداد ما هر کانال حداکثر ۶۴ رکورد دارد. شناسه دقیقاً int، در بازهٔ صفر تا یک‌میلیون و درون هر کانال اکیداً افزایشی است. مقدار نیز دقیقاً int و در بازهٔ منفی ده‌هزار تا ده‌هزار میلی‌ولت است. bool، float، رشتهٔ عددی، شناسهٔ تکراری، ترتیب کاهشی و رکورد با تعداد عضو نادرست رد می‌شوند. این حدود آموزشی‌اند و مشخصات تجهیزات نیستند.
دو کانال با طول برابر می‌توانند شناسه‌های متفاوت داشته باشند؛ مثلاً آخرین شناسهٔ سمت راست سه باشد. ابزار جفت‌کردن به‌تنهایی مفهوم شناسه را نمی‌داند. تابع ما پس از تشکیل هر جفت، برابری کلید را هم بررسی می‌کند. اگر سیاست کاربرد تطبیق زمانیِ تقریبی یا نگهداریِ دادهٔ گمشده باشد، باید قرارداد دیگری نوشت؛ این برنامه درون‌یابی یا اتصال بر اساس نزدیک‌ترین زمان انجام نمی‌دهد.
۲. برابری طول، تنبلی و زمانِ آشکارشدن خطا
مرجع رسمی zip توضیح می‌دهد که حالت معمول در پایان کوتاه‌ترین ورودی متوقف می‌شود. اگر آخرین رکورد راست حذف شود، جفت‌کردن معمول فقط دو ردیف می‌دهد و نمونهٔ سوم چپ وارد خروجی نمی‌شود. این رفتار برای بعضی مسائل مطلوب است، اما با فرضِ «دو کانال هم‌طول‌اند» سازگار نیست؛ فرض را باید به کنترل قابل مشاهده تبدیل کرد.
گزینهٔ strict=True هنگام تمام‌شدن نامساویِ ورودی‌ها ValueError می‌دهد. این گزینه از Python 3.10 وجود دارد و قابلیت تازهٔ نسخهٔ اجرای امروز نیست. با این حال zip تنبل است؛ ساختن آن کل ورودی را نمی‌خواند. در شاهد ما، دو جفت سالم از zip خام گرفته می‌شود و درخواستِ بعدی خطای کوتاهی را آشکار می‌کند. پس strict تضمینِ «هیچ پیشوندی قبلاً تحویل نشده» نیست.
پیشنهاد پذیرفته‌شدهٔ PEP 618 انگیزهٔ جلوگیری از کوتاه‌شدن بی‌صدای داده و زمان بروز خطا را شرح می‌دهد. مقالهٔ حاضر آن متن را ترجمه نمی‌کند؛ مسئله، داده، تابع و آزمون‌هایش جدا تألیف شده‌اند. strict نیز برابری مقدارها یا شناسه‌ها را بررسی نمی‌کند؛ سه جفت با شناسهٔ آخر متفاوت می‌توانند از خود zip سخت‌گیرانه عبور کنند.
در دسته‌بندی مولد حسگر، مسئله حفظ آخرین دسته و گذر دوم بود. اینجا دو ورودی مستقل از نظر معنایی را در کنار هم می‌خوانیم. بازبینی عمر گروه‌های groupby نیز اشتراک منبع را بررسی می‌کرد؛ شباهت ابزار تکرارگر نباید تفاوتِ قرارداد این سه مسئله را پنهان کند.
۳. کد کامل و تحویل پس از اعتبارسنجی
فایل strict_zip_sensor_channels.py با Python 3.14.5 و فقط کتابخانهٔ استاندارد اجرا شده است. checked_channel اعتبارسنجی را هنگام پیمایش انجام می‌دهد. ورودی می‌تواند list، tuple یا تکرارگر باشد؛ پشتیبانی از هر شیء صرفاً iterable وعده داده نشده است. پس رشته، مجموعه یا منبع سفارشیِ بدون رابط تکرارگر، جای دو کانالِ این قرارداد قرار نمی‌گیرد.
تابع paired_channels ردیف‌های پذیرفته‌شده را در pending نگه می‌دارد و فقط پس از پایان موفق، tuple کامل برمی‌گرداند. خروجی هر ردیف شامل شناسه، مقدار چپ، مقدار راست و اختلاف است. خطای دیررس باعث می‌شود نتیجهٔ موفقی از تابع برنگردد. نمونهٔ محل تحویل نیز فقط پس از برگشت تابع، نتیجه را به فهرست خروجی اضافه می‌کند؛ ثبت بیرونیِ واقعی در این مثال نداریم.
"""Strict pairing of two bounded, synthetic channel streams; no hardware I/O."""
import json
import platform
def checked_channel(source):
if type(source) not in (list, tuple) and not hasattr(source, "__next__"):
raise TypeError("list, tuple or iterator required")
previous = -1
for count, record in enumerate(source, 1):
if count > 64:
raise ValueError("at most 64 records per channel")
if type(record) not in (list, tuple) or len(record) != 2:
raise ValueError("record must contain sample id and millivolts")
sample_id, millivolts = record
if type(sample_id) is not int or not 0 <= sample_id <= 1_000_000:
raise ValueError("actual bounded integer sample id required")
if sample_id <= previous:
raise ValueError("strictly increasing ids within each channel required")
if type(millivolts) is not int or not -10_000 <= millivolts <= 10_000:
raise ValueError("actual integer millivolts in declared range required")
previous = sample_id
yield sample_id, millivolts
def paired_channels(left, right):
"""Return a complete immutable staged result, or raise without returning one."""
pending = []
for a, b in zip(checked_channel(left), checked_channel(right), strict=True):
if a[0] != b[0]:
raise ValueError("paired sample identities differ")
pending.append((a[0], a[1], b[1], a[1] - b[1]))
return tuple(pending)
def independent_reference(left, right):
"""Already validated small fixture lists; positional indexing, no zip/helpers."""
if len(left) != len(right):
raise ValueError("fixture lengths differ")
answer = []
for index in range(len(left)):
if left[index][0] != right[index][0]:
raise ValueError("fixture identities differ")
answer.append((left[index][0], left[index][1], right[index][1],
left[index][1] - right[index][1]))
return tuple(answer)
def tracked(records, label, trace):
for record in records:
trace.append((label, record[0]))
yield record
def rejects(error_type, function, *args):
try:
function(*args)
except error_type:
return True
return False
def main():
left = [(0, 1000), (1, 1010), (2, 1020)]
right = [(0, 990), (1, 1005), (2, 1010)]
hand = ((0, 1000, 990, 10), (1, 1010, 1005, 5), (2, 1020, 1010, 10))
result = paired_channels(left, right)
checks = {"hand_pairs_and_signed_differences": result == hand}
checks["independent_positional_oracle"] = result == independent_reference(left, right)
short = right[:-1]
default_pairs = list(zip(left, short))
checks["default_zip_silently_keeps_only_shorter_prefix"] = len(default_pairs) == 2
checks["strict_rejects_left_longer_and_right_longer"] = (
rejects(ValueError, paired_channels, left, short)
and rejects(ValueError, paired_channels, left[:-1], right))
raw = zip(left, short, strict=True)
raw_prefix = [next(raw), next(raw)]
checks["strict_error_is_late_after_valid_prefix"] = (
len(raw_prefix) == 2 and rejects(ValueError, next, raw))
checks["equal_length_different_id_is_not_solved_by_strict"] = (
len(list(zip(left, [(0, 990), (1, 1005), (3, 1010)], strict=True))) == 3
and rejects(ValueError, paired_channels, left,
[(0, 990), (1, 1005), (3, 1010)]))
checks["both_empty_allowed_one_empty_rejected"] = (
paired_channels([], []) == ()
and rejects(ValueError, paired_channels, [], right)
and rejects(ValueError, paired_channels, left, []))
consumed = []
source_left = tracked(left, "L", consumed)
source_right = tracked(right, "R", consumed)
checks["constructing_generators_has_no_sample_consumption"] = consumed == []
checks["one_pass_sources_and_left_to_right_observed_trace"] = (
paired_channels(source_left, source_right) == hand
and consumed == [("L", 0), ("R", 0), ("L", 1), ("R", 1), ("L", 2), ("R", 2)]
and list(source_left) == [] and list(source_right) == [])
after_error = []
error_left = tracked(left, "L", after_error)
error_right = tracked(short, "R", after_error)
published = ["previous-result"]
try:
ready = paired_channels(error_left, error_right)
except ValueError:
pass
else:
published.append(ready)
checks["staging_does_not_publish_prefix_or_undo_source_consumption"] = (
published == ["previous-result"] and bool(after_error))
checks["duplicate_and_decreasing_ids_rejected"] = all(
rejects(ValueError, paired_channels, records, records)
for records in [[(1, 0), (1, 1)], [(2, 0), (1, 1)]])
checks["record_type_and_integer_profile_rejected"] = all(
rejects(ValueError, paired_channels, [record], [(0, 0)])
for record in [(True, 1), (0, True), (0, 1.0), (0, "1"),
(-1, 0), (0, 10001), (0,), "01"])
checks["finite_bound_accepts_64_rejects_65"] = (
len(paired_channels([(i, 0) for i in range(64)],
[(i, 1) for i in range(64)])) == 64
and rejects(ValueError, paired_channels,
[(i, 0) for i in range(65)], [(i, 0) for i in range(65)]))
shared = iter(left + right)
shared_pairs = list(zip(shared, shared, strict=True))
checks["same_iterator_alias_pairs_successive_not_parallel_records"] = (
shared_pairs == [(left[0], left[1]), (left[2], right[0]), (right[1], right[2])])
checks["inputs_preserved_result_immutable_and_detached"] = (
left == [(0, 1000), (1, 1010), (2, 1020)] and type(result) is tuple
and all(type(row) is tuple for row in result))
mutation_left = [[0, 1000]]
detached = paired_channels(mutation_left, [[0, 990]])
mutation_left[0][1] = 999
checks["later_caller_edit_does_not_change_returned_integer_values"] = (
detached == ((0, 1000, 990, 10),))
checks = {name: bool(value) for name, value in checks.items()}
if not all(checks.values()):
raise RuntimeError(checks)
return {
"python_version": platform.python_version(), "data_origin": "original synthetic two-channel sample records",
"voltage_unit": "mV", "left": left, "right": right, "paired_result": result,
"difference_sum_mv": sum(row[3] for row in result),
"raw_strict_valid_prefix_count_before_error": len(raw_prefix),
"successful_consumption_trace": consumed,
"failed_consumption_trace_observed_this_runtime": after_error,
"checks": checks, "unique_check_groups": len(checks),
}
if __name__ == "__main__":
print(json.dumps(main(), ensure_ascii=False, indent=2, allow_nan=False))
این نگهداری موقت محدود، برگشتِ مصرف ورودی را تضمین نمی‌کند. مولدهای منبع ممکن است تا زمان کشف خطا پیش رفته باشند؛ اثر خارجیِ احتمالیِ تولید رکورد هم با استثنا ناپدید نمی‌شود. همچنین تابع تمام نتیجهٔ محدود را نگه می‌دارد و ادعای حافظهٔ ثابت ندارد. حد ۶۴ برای جلوگیری از پیمایش بی‌پایانِ موفق است، نه جایگزین timeout، مدیریت سوکت یا تراکنش پایگاه داده.
رکوردهای برگشتی از اعداد صحیحِ مستقل و tuple ساخته می‌شوند. تغییر بعدیِ فهرست ورودی، مقدارِ ردیفِ برگشتی را تغییر نمی‌دهد. این ویژگی را برای همین ساختار ساده می‌آزماییم؛ به معنی کپی عمیق هر نوع شیء یا محافظت در برابر دست‌کاری هم‌زمانِ منبع نیست. اگر نوع محتوا به اشیای تغییرپذیر گسترش یابد، قراردادِ نگهداری و آزمون استقلال نیز باید تغییر کند.
۴. مرجع مستقل، ردِ مصرف و نتیجهٔ اجرا
مرجع دوم، روی فهرست‌های کوچکِ معتبر از طول و اندیس استفاده می‌کند؛ zip، checked_channel یا تابع اصلی را دوباره صدا نمی‌زند. طول‌ها و شناسه‌های متناظر را کنترل و ردیف مورد انتظار را جایگاهی می‌سازد. مرجع مستقل، الزام یکسان بودن تعریف مسئله را حذف نمی‌کند؛ استقلال اینجا در روش ساخت جواب است و چهار عدد هر ردیف نیز با شاهد دستی مقایسه می‌شوند.
شانزده گروه بررسی واقعی گذشتند. کوتاهی در هر دو جهت، خطای دیررس، شناسهٔ متفاوت با طول برابر، ورودی خالی، نوع و حدود رکورد، منبع یک‌بارمصرف و استقلال خروجی بررسی شدند. مرز ۶۴ و ۶۵ رکورد نیز اجرا شد. این تعداد، نام گروه‌هاست؛ تک‌تک ورودی‌های نامعتبرِ داخل یک گروه را مقاله یا آزمایش مستقل نمی‌شماریم.
{
"python_version": "3.14.5",
"data_origin": "original synthetic two-channel sample records",
"voltage_unit": "mV",
"left": [
[
0,
1000
],
[
1,
1010
],
[
2,
1020
]
],
"right": [
[
0,
990
],
[
1,
1005
],
[
2,
1010
]
],
"paired_result": [
[
0,
1000,
990,
10
],
[
1,
1010,
1005,
5
],
[
2,
1020,
1010,
10
]
],
"difference_sum_mv": 25,
"raw_strict_valid_prefix_count_before_error": 2,
"successful_consumption_trace": [
[
"L",
0
],
[
"R",
0
],
[
"L",
1
],
[
"R",
1
],
[
"L",
2
],
[
"R",
2
]
],
"failed_consumption_trace_observed_this_runtime": [
[
"L",
0
],
[
"R",
0
],
[
"L",
1
],
[
"R",
1
],
[
"L",
2
]
],
"checks": {
"hand_pairs_and_signed_differences": true,
"independent_positional_oracle": true,
"default_zip_silently_keeps_only_shorter_prefix": true,
"strict_rejects_left_longer_and_right_longer": true,
"strict_error_is_late_after_valid_prefix": true,
"equal_length_different_id_is_not_solved_by_strict": true,
"both_empty_allowed_one_empty_rejected": true,
"constructing_generators_has_no_sample_consumption": true,
"one_pass_sources_and_left_to_right_observed_trace": true,
"staging_does_not_publish_prefix_or_undo_source_consumption": true,
"duplicate_and_decreasing_ids_rejected": true,
"record_type_and_integer_profile_rejected": true,
"finite_bound_accepts_64_rejects_65": true,
"same_iterator_alias_pairs_successive_not_parallel_records": true,
"inputs_preserved_result_immutable_and_detached": true,
"later_caller_edit_does_not_change_returned_integer_values": true
},
"unique_check_groups": 16
}مولدهای دارای ردِ مصرف، هنگام ساخت هیچ نمونه‌ای تحویل نمی‌دهند. اجرای موفق در این محیط، ترتیب چپ صفر، راست صفر، چپ یک، راست یک، چپ دو، راست دو را ثبت می‌کند. در خطای کوتاهیِ راست، چپ دو نیز پیش از خطا مصرف شده است. این رد، مشاهدهٔ همین برنامه و نسخه است؛ شمار خواندن از درایور، بافر سیستم‌عامل یا شبکهٔ واقعی اندازه‌گیری نشده است.
اگر یک تکرارگر واحد را دو بار به zip بدهیم، دو کانال مستقل نمی‌سازیم. در شاهد خامِ شش رکورد، اعضای پیاپیِ همان منبع دوتادوتا کنار هم قرار می‌گیرند. حتی strict می‌تواند این سه جفت را بدون خطا بسازد، چون تعداد اعضای منبع زوج است. هویت ورودی‌ها، برابری طول و معنای جفت‌ها سه موضوع جدا هستند؛ شکل ظاهریِ خروجی کافی نیست.
۵. چهار پرسش با پاسخ کامل
پرسش اول: حذف رکورد آخرِ راست چه تفاوتی ایجاد می‌کند؟ پاسخ: zip معمول دو جفت می‌دهد؛ zip سخت‌گیرانه همان پیشوند را می‌تواند بدهد و سپس خطا کند. تابعِ ما نتیجهٔ کامل موفق برنمی‌گرداند، چون تحویل را تا پایان اعتبارسنجی عقب انداخته‌ایم. نمونه‌های مصرف‌شدهٔ منبع خودکار بازگردانده نمی‌شوند.
پرسش دوم: اگر طول‌ها برابر و آخرین شناسهٔ راست سه باشد چه می‌شود؟ پاسخ: خود zip سخت‌گیرانه سه جفت می‌سازد؛ کنترل شناسهٔ تابع، جفت سوم را رد می‌کند. strict کنترل محتوای زوج‌ها نیست. این مثال عمداً کلید دقیق می‌خواهد و سیاست تحمل اختلاف زمان یا ترمیم شناسه پیاده نکرده است.
پرسش سوم: دو کانال خالی پذیرفته‌اند؟ پاسخ: بله؛ خروجی tuple تهی است و این سیاستِ همین درس است. خالی بودن فقط یکی، خطا دارد. اگر کاربرد حداقل یک مشاهده بخواهد، باید شرط جدا و آزمون روشن به مرز تحویل اضافه شود؛ تهی را نباید بی‌دلیل با خرابی دریافت یکی دانست.
پرسش چهارم: آیا استثنا نوشته‌های بیرونیِ قبلی را پس می‌گیرد؟ پاسخ: خیر. مثالِ تحویل فقط نشان می‌دهد تابعِ کامل‌نشده به فهرست مقصد اضافه نشده؛ تراکنش یا rollback واقعی وجود ندارد. منابع 3.14.8 و PEP در ۲۵ مهر ۱۴۰۵ خوانده شده‌اند و نسخهٔ اجرا جدا ثبت است. این آموزش پایه، خبر تازه یا ارزیابی دستگاه واقعی نیست.
منابع و مطالعهٔ بیشتر
بررسی منابع: ۲۵ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.
