ثبت دادهٔ حسگر در SQLite با پایتون؛ آزمون تراکنش و بسته‌های تکراری

۱. قرارداد دادهٔ حسگر را پیش از جدول بنویسید

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

هر بسته چهار فیلد دارد: شناسهٔ دستگاه، شمارهٔ ترتیبی، زمان UTC برحسب میلی‌ثانیه از مبدأ یونیکس و ولتاژ برحسب میلی‌ولت. برای نمونه، ۲۴۰۰۰ میلی‌ولت معادل ۲۴ ولت است. انتخاب عدد صحیح برای میلی‌ولت، مقایسهٔ دقیق بسته‌های تکراری این تمرین را ساده می‌کند؛ به معنی دقت واقعی یک میلی‌ولت در سخت‌افزار نیست. بازهٔ مجاز صفر تا ۶۰۰۰۰ میلی‌ولت را صرفاً قرارداد این نمونه قرار داده‌ایم. دامنهٔ حسگر واقعی و سیاست نگهداری اندازه‌گیری خارج از محدوده باید جدا مشخص شوند.

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

۲. کلید مرکب، رابطهٔ دستگاه و نوع داده

جدول device دستگاه‌های شناخته‌شده و جدول sample مشاهده‌ها را نگه می‌دارد. کلید مرکب مانع دو ردیف با هویت یکسان می‌شود و کلید خارجی، وجود دستگاه مرجع را الزام می‌کند. مطابق مستندات رسمی کلید خارجی SQLite، فعال‌سازی و بررسی این قابلیت باید برای هر اتصال انجام شود. برنامه آن را پیش از شروع تراکنش فعال می‌کند و مقدار برگشتی را می‌خواند. وجود نام دستگاه در جدول، احراز هویت سخت‌افزار یا اثبات اصالت بسته نیست؛ فقط سازگاری رابطهٔ داده را کنترل می‌کند.

از جدول STRICT استفاده کرده‌ایم که طبق مرجع رسمی STRICT از SQLite 3.37.0 پشتیبانی می‌شود. STRICT به معنای رد هر تبدیل نوع نیست؛ مقدار متنیِ عددی ممکن است بدون از دست رفتن اطلاعات به عدد تبدیل شود. به همین دلیل تابع Python قرارداد ورودی را نیز بررسی می‌کند: شماره، زمان و میلی‌ولت باید دقیقاً int باشند و حتی bool به جای آن پذیرفته نمی‌شود. سپس محدودیت‌های NOT NULL و CHECK در خود پایگاه داده، لایهٔ دیگری برای سازگاری مقدار فراهم می‌کنند.

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

۳. کد کامل ثبت دسته‌ای و آزمون بازگشت

برنامه با کتابخانهٔ استاندارد Python اجرا می‌شود و بستهٔ خارجی لازم ندارد. نسخهٔ اجراشده Python 3.14.5 و کتابخانهٔ SQLite متصل به آن 3.50.4 است؛ نسخهٔ SQLite الزاماً از روی شمارهٔ Python معلوم نمی‌شود. فایل را با نام sqlite_sensor.py ذخیره و با دستور python sqlite_sensor.py اجرا کنید. پایگاه داده در پوشهٔ موقت مخصوص همین آزمون ساخته می‌شود؛ هیچ فایل موجودی به‌عنوان پایگاه داده باز یا بازنویسی نمی‌شود.

کنترل تراکنش را عمداً صریح نوشته‌ایم: اتصال autocommit=True دارد و تابع با SQL، دستورات BEGIN IMMEDIATE، COMMIT و در صورت خطا ROLLBACK را اجرا می‌کند. مستندات sqlite3 در شاخهٔ Python 3.14 رفتار حالت اتصال و پارامترهای پرس‌وجو را توضیح می‌دهد. در این حالت نباید con.commit را جایگزین بی‌بررسی COMMIT صریح کرد. تمام مقدارها با جای‌نگهدار به SQL متصل می‌شوند و شناسهٔ دستگاه به متن دستور چسبانده نمی‌شود.

بر اساس مرجع تراکنش SQLite، BEGIN IMMEDIATE می‌تواند هنگام وجود نویسندهٔ دیگر با خطای مشغول‌بودن روبه‌رو شود. مثال ما یک اتصال نویسنده دارد و سیاست تلاش دوباره در رقابت چند فرایند را پیاده نمی‌کند. اگر خطای یکی از رکوردها درون دسته رخ دهد، تابع وضعیت تراکنش را بررسی و تغییرات همان دسته را برمی‌گرداند. رکوردهای موفق دسته‌های قبلی باقی می‌مانند. این قرارداد «همه یا هیچ» در محدودهٔ هر فراخوانی است.

"""Offline synthetic voltage packets: transaction and duplicate-policy lesson."""
import hashlib
import json
import platform
import sqlite3
import tempfile
from pathlib import Path

SCHEMA = """
CREATE TABLE device(id TEXT PRIMARY KEY) STRICT;
CREATE TABLE sample(
    device_id TEXT NOT NULL REFERENCES device(id),
    seq INTEGER NOT NULL CHECK(seq >= 0),
    utc_ms INTEGER NOT NULL CHECK(utc_ms >= 0),
    voltage_mv INTEGER NOT NULL CHECK(voltage_mv BETWEEN 0 AND 60000),
    PRIMARY KEY(device_id, seq)
) STRICT;
CREATE INDEX sample_device_time ON sample(device_id, utc_ms);
"""


def open_db(path):
    if sqlite3.sqlite_version_info < (3, 37, 0):
        raise RuntimeError('SQLite 3.37+ is required for STRICT tables')
    con = sqlite3.connect(path, autocommit=True)
    con.execute('PRAGMA foreign_keys = ON')
    if con.execute('PRAGMA foreign_keys').fetchone() != (1,):
        con.close()
        raise RuntimeError('Foreign-key enforcement is unavailable')
    return con


def ingest(con, packets):
    """All-or-nothing batch; exact replays are no-ops, altered replays fail."""
    inserted, duplicates = 0, 0
    con.execute('BEGIN IMMEDIATE')
    try:
        for packet in packets:
            if len(packet) != 4:
                raise ValueError('Expected device, sequence, UTC ms, voltage mV')
            dev, seq, utc, mv = packet
            if not isinstance(dev, str) or not dev:
                raise ValueError('Nonempty device identifier required')
            if not all(type(v) is int for v in (seq, utc, mv)):
                raise ValueError('Sequence, UTC milliseconds and mV must be ints')
            old = con.execute(
                'SELECT utc_ms, voltage_mv FROM sample WHERE device_id=? AND seq=?',
                (dev, seq)).fetchone()
            if old is not None:
                if old != (utc, mv):
                    raise ValueError('Conflicting replay: same ID, different payload')
                duplicates += 1
                continue
            con.execute('INSERT INTO sample VALUES(?,?,?,?)', packet)
            inserted += 1
        con.execute('COMMIT')  # SQL explicit transaction; autocommit=True
    except Exception:
        if con.in_transaction:
            con.execute('ROLLBACK')
        raise
    return {'inserted': inserted, 'duplicates': duplicates}


def snapshot(con):
    return con.execute('SELECT * FROM sample ORDER BY device_id, seq').fetchall()


def main():
    checks = {}
    epoch = 1792022400000  # authored UTC epoch-ms reference, not live sensor time
    packets = [('supply-A', 1, epoch, 24000),
               ('supply-A', 2, epoch + 1000, 24100),
               ('supply-A', 3, epoch + 2000, 23900)]
    with tempfile.TemporaryDirectory(prefix='sensor-sqlite-lesson-') as folder:
        path = Path(folder) / 'fixture.db'
        con = open_db(path)
        try:
            con.executescript(SCHEMA)
            con.execute('INSERT INTO device VALUES(?)', ('supply-A',))
            first = ingest(con, packets)
            assert first == {'inserted': 3, 'duplicates': 0}
            checks['first_batch_committed'] = True
            replay = ingest(con, packets)
            assert replay == {'inserted': 0, 'duplicates': 3}
            baseline = snapshot(con)
            checks['identical_replay_is_idempotent'] = True
            failures = [
                ('conflicting_replay_rolls_back_batch', ValueError,
                 [('supply-A', 4, epoch + 3000, 24200),
                  ('supply-A', 2, epoch + 1000, 24500)]),
                ('unknown_device_rejected', sqlite3.IntegrityError,
                 [('unknown', 1, epoch, 24000)]),
                ('voltage_limit_rolls_back_batch', sqlite3.IntegrityError,
                 [('supply-A', 4, epoch + 3000, 24200),
                  ('supply-A', 5, epoch + 4000, 61000)]),
                ('noninteger_timestamp_rejected', ValueError,
                 [('supply-A', 4, '1792022403000', 24200)])]
            for name, expected, batch in failures:
                try:
                    ingest(con, batch)
                except expected:
                    assert snapshot(con) == baseline and not con.in_transaction
                    checks[name] = True
                else:
                    raise AssertionError(name + ' did not fail')
            window = con.execute(
                'SELECT seq, voltage_mv FROM sample '
                'WHERE device_id=? AND utc_ms>=? AND utc_ms<? ORDER BY utc_ms, seq',
                ('supply-A', epoch + 1000, epoch + 2000)).fetchall()
            assert window == [(2, 24100)]
            checks['half_open_window_boundaries'] = True
            injection = "supply-A' OR 1=1 --"
            assert con.execute('SELECT count(*) FROM sample WHERE device_id=?',
                               (injection,)).fetchone() == (0,)
            checks['identifier_is_bound_as_data'] = True
            logical_hash = hashlib.sha256(json.dumps(baseline,
                separators=(',', ':')).encode()).hexdigest()
        finally:
            con.close()
        con = open_db(path)
        try:
            assert snapshot(con) == baseline
            assert con.execute('PRAGMA foreign_keys').fetchone() == (1,)
            checks['reopen_preserves_rows_and_enforces_fk'] = True
        finally:
            con.close()
    report = {'versions': {'python': platform.python_version(),
                           'sqlite_runtime': sqlite3.sqlite_version},
              'first_batch': first, 'exact_replay': replay,
              'rows_after_rejected_batches': baseline,
              'half_open_window_result': window,
              'logical_rows_sha256': logical_hash,
              'checks': checks,
              'limitations': ['One writer, tiny authored dataset; no throughput benchmark',
                              'Close/reopen tested, not power-loss recovery or storage wear',
                              'UTC timestamps supplied by fixture; no clock synchronization',
                              'SQL constraints do not prove sensor calibration or trustworthy hardware']}
    print(json.dumps(report, indent=2))


if __name__ == '__main__':
    main()

۴. خروجی ثبت‌شده و پاسخ تمرین‌ها

اولین دسته سه ردیف درج می‌کند و ارسال دوبارهٔ همان داده، صفر درج و سه تکرار گزارش می‌دهد. در آزمون تعارض، ابتدا شمارهٔ ۴ معتبر به دسته اضافه می‌شود و سپس نسخهٔ تغییرکردهٔ شمارهٔ ۲ می‌رسد؛ خطا باعث می‌شود شمارهٔ ۴ نیز از همان دسته باقی نماند. آزمون مقدار ۶۱۰۰۰ میلی‌ولت هم بازگشت دسته را کنترل می‌کند. پس از تمام خطاهای عمدی، تنها همان سه ردیف اولیه با مقادیر ۲۴۰۰۰، ۲۴۱۰۰ و ۲۳۹۰۰ میلی‌ولت وجود دارند.

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

{
  "versions": {
    "python": "3.14.5",
    "sqlite_runtime": "3.50.4"
  },
  "first_batch": {
    "inserted": 3,
    "duplicates": 0
  },
  "exact_replay": {
    "inserted": 0,
    "duplicates": 3
  },
  "rows_after_rejected_batches": [
    [
      "supply-A",
      1,
      1792022400000,
      24000
    ],
    [
      "supply-A",
      2,
      1792022401000,
      24100
    ],
    [
      "supply-A",
      3,
      1792022402000,
      23900
    ]
  ],
  "half_open_window_result": [
    [
      2,
      24100
    ]
  ],
  "logical_rows_sha256": "4f0968d5c6ff356de408281da992444517666577ae7ee0420aaebe099c3ceb42",
  "checks": {
    "first_batch_committed": true,
    "identical_replay_is_idempotent": true,
    "conflicting_replay_rolls_back_batch": true,
    "unknown_device_rejected": true,
    "voltage_limit_rolls_back_batch": true,
    "noninteger_timestamp_rejected": true,
    "half_open_window_boundaries": true,
    "identifier_is_bound_as_data": true,
    "reopen_preserves_rows_and_enforces_fk": true
  },
  "limitations": [
    "One writer, tiny authored dataset; no throughput benchmark",
    "Close/reopen tested, not power-loss recovery or storage wear",
    "UTC timestamps supplied by fixture; no clock synchronization",
    "SQL constraints do not prove sensor calibration or trustworthy hardware"
  ]
}

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

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

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

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

۵. از ثبت قابل‌آزمون تا دادهٔ تحلیل و هوش مصنوعی

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

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

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

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

• کنترل تراکنش و اتصال پارامترها در sqlite3 پایتون ۳٫۱۴
• فعال‌سازی و بررسی کلیدهای خارجی SQLite
• نوع داده و تبدیل بدون اتلاف در جدول‌های STRICT
• تراکنش صریح و رفتار BEGIN IMMEDIATE در SQLite

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