Лабораторна робота №05. Python-модуль дискретного сенсора

Мета: спроєктувати уніфікований Python API дискретного сенсора; реалізувати нормалізацію активного рівня, фільтрацію брязкоту послідовними вибірками, очікування стану з тайм-аутом і діагностику невизначеного значення; створити конфігурацію поза кодом, mock-джерело та відтворювані позитивні, негативні й граничні тести без обладнання.

Результати навчання та передумови

Після виконання студент уміє:

1. описати паспорт сенсорного модуля: призначення, API, стани, конфігурацію, safe-state і межі перевірки;
2. перетворювати сире булеве значення за параметром active_low;
3. повертати UNKNOWN, доки не отримано потрібну кількість послідовних однакових вибірок;
4. очікувати цільову подію лише до заданого тайм-ауту;
5. відрізнити SensorTimeout від SensorFault і не продовжувати фізичний цикл після жодної з них;
6. тестувати модуль через mock-послідовності без GPIO.

Передумови: ЛР-03–04; базові класи, Enum, протоколи, винятки та JSON у Python.

Середовище виконання

Необхідні знання та матеріали

Знання

• active-high/active-low, None як недостовірне спостереження;
• брязкіт/короткі завади та компроміс між стійкістю й затримкою;
• монотонний час і тайм-аут;
• dependency injection: логіка залежить від інтерфейсу, а не від GPIO-бібліотеки.

Обладнання, ПЗ та вхідні дані

• фізичний сенсор і Raspberry Pi не потрібні;
• Python зі стандартною бібліотекою і текстовий редактор;
• сценарії відмов як контекст очікуваної fail-closed реакції;
• апаратна частина, програмне середовище, електричні прогалини і політика допуску.

Ризики та правила безпеки

Ризики

Заборонені дії

• підключати сенсор, змінювати COM/проводку або подавати напругу;
• підбирати GPIO, active_low чи таймінги методом фізичного експерименту;
• імпортувати апаратну GPIO-бібліотеку або запускати з sudo;
• перехоплювати SensorTimeout/SensorFault і мовчки продовжувати рух.

Умови негайної зупинки

Зупинити програмний сценарій і повідомити оператора за фізичного руху, доступу до GPIO, невідомого сирого типу, повторюваного UNKNOWN, тайм-ауту, суперечності документації, пошкодження кабелю, нагріву або запаху.

Безпечний стан

Сенсор сам не має виконавчого виходу. Безпечна реакція споживача API: не формувати нову команду виконавцю; зупинити активні виходи через їхні модулі; записати UNKNOWN/помилку; вимагати встановлення причини й підтвердження оператора.

Передпусковий чекліст

• mode — тільки mock.
• Конфігурація не містить GPIO, IP, паролів або токенів.
• stable_samples, poll_interval_s, timeout_s валідовані.
• None не нормалізується в False.
• Тайм-аут обов'язковий і скінченний.
• Усі тести виконуються без фізичної бібліотеки.
• Споживач API документує реакцію на обидва винятки.

Методичні вказівки й теоретичні відомості

1. Контракт модуля

Фільтр N послідовних вибірок відхиляє коротший імпульс, але додає затримку. За періоду опитування T підтвердження з'явиться не раніше N-ї вибірки; реальна верхня межа додатково залежить від планувальника та драйвера. Саме тому N і T є конфігурацією, а не «магічними» константами.

2. Потік станів

Текстовий опис рисунка: mock-джерело подає True, False або None; нормалізація враховує active_low; дебаунсер накопичує N послідовних значень і видає ACTIVE/INACTIVE, тоді як None повертає UNKNOWN; очікувач завершується цільовим станом або тайм-аутом, після якого споживач блокує виконавчі команди.

flowchart LR
    M["MockRawInput"] --> N["Нормалізація active_low"]
    N --> D["N послідовних вибірок"]
    D -->|"підтверджено"| S["ACTIVE / INACTIVE"]
    D -->|"None або нестабільно"| U["UNKNOWN"]
    S --> W["wait_for(target, timeout)"]
    U --> W
    W -->|"ціль до тайм-ауту"| O["Підтверджена подія"]
    W -->|"тайм-аут / fault"| E["Блокувати команди й журналювати"]

Рис. 1. Ілюстративний потік нормалізації, дебаунсу та обмеженого очікування сенсора

3. Чому mock є частиною API

Mock має відтворювати не лише «ідеальні» 0/1, а брязкіт, None, довгу відсутність події та некоректний тип. Він дає детермінований тест оркестрації, але не вимірює реальну електричну заваду, затримку ОС або геометрію детекції.

Хід виконання роботи

1. Описати паспорт і контракт сенсора.
2. Створити несекретну mock-конфігурацію.
3. Реалізувати модуль і mock.
4. Перевірити брязкіт, інверсію, тайм-аут та граничні значення.
5. Підготувати результати й правила інтеграції.

Виконання лабораторної роботи

Крок 1. Специфікувати API й safe-state

Заповніть таблицю контракту власними словами та наведіть приклад споживача: конвеєр не запускає наступний етап, поки сенсор не повернув підтверджений ACTIVE.

Код/команда: не потрібні.

Очікуваний результат: описано всі три стани, два винятки, тайм-аут, закриття й реакцію споживача.

Критерій правильності: UNKNOWN не дорівнює INACTIVE; fault/timeout не ведуть до автоматичного продовження.

Якщо результат не отримано: поверніться до таблиці контракту й позначте невизначене; не додавайте hardware API.

Крок 2. Створити зовнішню конфігурацію

Збережіть як lab05-sensor.json. Числа є демонстраційними для mock, не параметрами MERC-I5.

{
  "mode": "mock",
  "active_low": false,
  "stable_samples": 3,
  "poll_interval_s": 0.01,
  "timeout_s": 0.05
}

python -m json.tool lab05-sensor.json

Очікуваний результат: валідний несекретний JSON без номера піна.

Критерій правильності: stable_samples — додатне ціле; часи додатні; mode=mock.

Якщо результат не отримано: виправте синтаксис; не підставляйте фактичний GPIO без схеми.

Крок 3. Реалізувати SensorModule і mock

Збережіть як lab05_sensor.py. Без аргументів код виконує скінченний smoke-тест без сну, мережі й обладнання.

# safe-smoke
from __future__ import annotations

import json
import math
from collections import deque
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Callable, Iterable, Mapping, Protocol


class SensorState(str, Enum):
    UNKNOWN = "UNKNOWN"
    INACTIVE = "INACTIVE"
    ACTIVE = "ACTIVE"


class SensorFault(RuntimeError):
    pass


class SensorTimeout(TimeoutError):
    pass


class RawInput(Protocol):
    def read(self) -> bool | None: ...
    def close(self) -> None: ...


class MockRawInput:
    def __init__(self, samples: Iterable[bool | None]) -> None:
        values = list(samples)
        if not values:
            raise ValueError("at least one mock sample is required")
        self._samples = deque(values)
        self._last = values[-1]
        self.closed = False

    def read(self) -> bool | None:
        if self.closed:
            raise SensorFault("input is closed")
        if self._samples:
            self._last = self._samples.popleft()
        return self._last

    def close(self) -> None:
        self.closed = True


@dataclass(frozen=True)
class SensorConfig:
    mode: str
    active_low: bool
    stable_samples: int
    poll_interval_s: float
    timeout_s: float


def parse_config(raw: Mapping[str, object]) -> SensorConfig:
    if raw.get("mode") != "mock":
        raise ValueError("only mode=mock is permitted")
    active_low = raw.get("active_low")
    stable = raw.get("stable_samples")
    poll = raw.get("poll_interval_s")
    timeout = raw.get("timeout_s")
    if not isinstance(active_low, bool):
        raise ValueError("active_low must be boolean")
    if not isinstance(stable, int) or isinstance(stable, bool) or not 1 <= stable <= 100:
        raise ValueError("stable_samples must be 1..100")
    if not isinstance(poll, (int, float)) or isinstance(poll, bool) or not 0 < float(poll) <= 1:
        raise ValueError("poll_interval_s must be >0 and <=1")
    if not isinstance(timeout, (int, float)) or isinstance(timeout, bool) or not 0 < float(timeout) <= 60:
        raise ValueError("timeout_s must be >0 and <=60")
    return SensorConfig("mock", active_low, stable, float(poll), float(timeout))


class SensorModule:
    def __init__(self, source: RawInput, config: SensorConfig, sleeper: Callable[[float], None]) -> None:
        self.source = source
        self.config = config
        self.sleeper = sleeper
        self._candidate: SensorState | None = None
        self._count = 0
        self._stable = SensorState.UNKNOWN

    def read_state(self) -> SensorState:
        raw = self.source.read()
        if raw is None:
            self._candidate, self._count = None, 0
            self._stable = SensorState.UNKNOWN
            return self._stable
        if type(raw) is not bool:
            raise SensorFault(f"invalid raw type: {type(raw).__name__}")
        logical = raw ^ self.config.active_low
        candidate = SensorState.ACTIVE if logical else SensorState.INACTIVE
        if candidate == self._candidate:
            self._count += 1
        else:
            self._candidate, self._count = candidate, 1
        if self._count >= self.config.stable_samples:
            self._stable = candidate
        return self._stable

    def wait_for(self, target: SensorState, timeout_s: float | None = None) -> SensorState:
        if target is SensorState.UNKNOWN:
            raise ValueError("UNKNOWN cannot be a successful target")
        timeout = self.config.timeout_s if timeout_s is None else timeout_s
        if not isinstance(timeout, (int, float)) or isinstance(timeout, bool) or timeout <= 0:
            raise ValueError("timeout must be positive")
        attempts = max(1, math.ceil(float(timeout) / self.config.poll_interval_s))
        for index in range(attempts):
            if self.read_state() is target:
                return target
            if index + 1 < attempts:
                self.sleeper(self.config.poll_interval_s)
        raise SensorTimeout(f"{target.value} not reached in {timeout}s")

    def close(self) -> None:
        self.source.close()


def smoke_test() -> dict[str, object]:
    config = parse_config({
        "mode": "mock", "active_low": False, "stable_samples": 3,
        "poll_interval_s": 0.01, "timeout_s": 0.06,
    })
    sensor = SensorModule(MockRawInput([False, True, False, True, True, True]), config, lambda _: None)
    assert sensor.wait_for(SensorState.ACTIVE) is SensorState.ACTIVE
    sensor.close()

    inverse = SensorModule(
        MockRawInput([False]),
        SensorConfig("mock", True, 1, 0.01, 0.01),
        lambda _: None,
    )
    assert inverse.wait_for(SensorState.ACTIVE) is SensorState.ACTIVE

    unknown = SensorModule(MockRawInput([None]), config, lambda _: None)
    try:
        unknown.wait_for(SensorState.ACTIVE, 0.02)
    except SensorTimeout:
        timeout_ok = True
    else:
        raise AssertionError("UNKNOWN must not satisfy ACTIVE")

    try:
        parse_config({"mode": "physical", "active_low": False, "stable_samples": 1,
                      "poll_interval_s": 0.01, "timeout_s": 1.0})
    except ValueError:
        physical_rejected = True
    else:
        raise AssertionError("physical mode must be rejected")
    return {
        "bounce_filtered": True,
        "active_low_checked": True,
        "timeout_checked": timeout_ok,
        "physical_mode_rejected": physical_rejected,
        "hardware_tested": False,
    }


def main() -> int:
    # Необов'язкове читання профілю показує відокремлення конфігурації;
    # без аргументів smoke-тест повністю самодостатній.
    import sys
    if len(sys.argv) == 3 and sys.argv[1] == "--config":
        parse_config(json.loads(Path(sys.argv[2]).read_text(encoding="utf-8")))
    elif len(sys.argv) != 1:
        raise SystemExit("usage: lab05_sensor.py [--config FILE]")
    print(json.dumps(smoke_test(), ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python lab05_sensor.py
python lab05_sensor.py --config lab05-sensor.json

Очікуваний результат: JSON із чотирма true і hardware_tested: false; обидві команди завершуються з кодом 0.

Критерій правильності: брязкіт не дає ранній ACTIVE; три стабільні True підтверджують стан; active_low перевірено; None завершується тайм-аутом; physical mode відхилено.

Якщо результат не отримано: виведіть послідовність станів у копії, перевірте тип bool і параметри; не підключайте GPIO.

Крок 4. Виконати негативні й граничні тести

Додайте локальні тести: вічний False; None; нестабільне True/False; закрите джерело; stable_samples=0/101; timeout=0; межа stable_samples=1; active_low=True для сирого False.

Код/команда: python lab05_sensor.py після кожної групи assert.

Очікуваний результат: timeout/fault/валідаційний виняток виникає перед будь-якою наступною дією; обидві валідні межі працюють.

Критерій правильності: 8/8 тестів; процес скінченний; не існує шляху, де UNKNOWN стає підтвердженим ACTIVE.

Якщо результат не отримано: мінімізуйте послідовність, скиньте внутрішній стан новим екземпляром і не приховуйте виняток.

Крок 5. Оформити паспорт та приклад інтеграції

Опишіть призначення, сирий контракт, нормалізацію, стани, винятки, параметри, mock, safe-state споживача, запуск/зупинку й межі апаратної перевірки. Додайте псевдопослідовність «wait — підтвердження — наступна дія», але без керування приводом.

Код/команда: не потрібні.

Очікуваний результат: інший студент може замінити mock-послідовність і відтворити тести, не знаючи GPIO.

Критерій правильності: жодного pinout; параметри позначені демонстраційними; fault і timeout ведуть до блокування команд.

Якщо результат не отримано: зіставте паспорт із публічними методами та тестами; невідоме винесіть у перелік потрібних доказів.

Сценарії перевірки

Таблиці вимірювань і метрики

Трасування вибірок

Підсумок

Вимоги до звіту

Дотримуйтеся загальних вимог. Додатково подайте паспорт модуля, JSON, код, трасування bounce-послідовності, вивід smoke-тесту, вісім тестів, аналіз затримки N-вибіркового фільтра, реакцію споживача на fault/timeout та фразу «Модель, електрична схема, GPIO і фізичний сенсор не перевірялися».

Критерії оцінювання

Критична помилка: фізичне підключення, вигаданий pinout/таймінг, нескінченне очікування або продовження після fault.

Контрольні питання

1. Чому UNKNOWN є окремим станом?
2. Як працює raw XOR active_low?
3. Який компроміс створює збільшення stable_samples?
4. Чому timeout має бути конфігурацією?
5. Чим SensorFault відрізняється від SensorTimeout?
6. Чому mock має утримувати останнє значення після завершення послідовності?
7. Навіщо закривати джерело і тестувати читання після закриття?
8. Що повинен зробити конвеєр після невідомого стану сенсора?
9. Які дані потрібні для hardware adapter?
10. Чого не доводить успішний mock-тест?

Висновки

У висновку наведіть результати bounce, інверсії, timeout і граничних тестів; поясніть, як API відмовляє за замовчуванням і що робить споживач. Окремо зазначте, що демонстраційні часові параметри, модель сенсора, pinout і апаратна працездатність MERC-I5 не підтверджені.