Лабораторна робота №06. Python-модуль конвеєра

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

Після роботи студент уміє:

1. описати паспорт конвеєрного модуля: призначення, інтерфейси, стани, конфігурацію, safe-state і відновлення;
2. відокремити нормалізовану команду швидкості 0.0…1.0 від непідтвердженої фізичної швидкості/електричного сигналу;
3. не вважати команду виконаною без окремого feedback;
4. завершити transfer лише після сигналу приймання й до тайм-ауту;
5. зупинити mock-вихід за заклинювання або втрати деталі та явно позначити зупинку непідтвердженою за втрати зв'язку;
6. довести тестами відсутність нескінченного очікування й автоматичного restart.

Передумови: ЛР-03–05; класи, Enum, протоколи, винятки, JSON; розуміння того, що процесна команда stop() не є автоматично safety-функцією.

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

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

Знання

• команда, feedback і технологічне підтвердження є різними сигналами;
• тайм-аут обмежує очікування, але його значення потребує обґрунтування;
• втрата зв'язку означає невідомий фізичний стан, а не підтверджену зупинку;
• нормалізована швидкість — програмна абстракція, яку майбутній adapter має перевести в підтверджені одиниці/сигнал.

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

• фізичний конвеєр і сенсор не потрібні;
• Python зі стандартною бібліотекою та текстовий редактор;
• технічна специфікація — декларує два регульовані конвеєри, але не визначає перевірений інтерфейс керування;
• апаратна частина, архітектура, експлуатація й безпека, програма ЛР і сценарій conveyor_jam.

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

Ризики

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

• запускати конвеєр, торкатися стрічки або утримувати деталь;
• створювати фізичне заклинювання чи втрату деталі;
• підключати мотор/драйвер до GPIO, реле або MG400 I/O;
• підставляти значення з технічної специфікації без фактичного зіставлення команди та швидкості;
• після fault автоматично викликати start, скидати захист або продовжувати цикл.

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

Негайно припинити роботу й повідомити оператора за будь-якого фізичного руху, втрати feedback/зв'язку, заклинювання, падіння деталі, суперечливого сенсора, пошкодження, нагріву, запаху чи неочікуваного доступу до I/O.

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

У mock: команда швидкості 0.0, feedback running=false, нові transfer заблоковані після fault до створення нового перевіреного екземпляра. У реальній системі безпечний стан не можна вважати досягнутим лише після мережевої команди stop: потрібне підтвердження приводу або незалежна функція, визначена оцінюванням ризику. За втрати зв'язку стан UNKNOWN/FAULT, а stop_confirmed=false ескалується оператору.

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

• mode=mock; hardware backend відсутній.
• Мінімум/максимум нормалізованої команди та всі tick-тайм-аути валідовані.
• Початковий стан STOPPED, команда 0.0.
• Є окремі running feedback, part-present і accept-event у mock.
• finally завжди викликає safe_stop().
• За втрати зв'язку немає хибного stop_confirmed=true.
• Jam, lost part, connection lost і межі швидкості протестовані.
• Фізичний restart не передбачено.

[!WARNING] [ПОТРЕБУЄ ДОПРАЦЮВАННЯ: МОДЕЛЬ, СИЛОВА СХЕМА Й ІНТЕРФЕЙС КОНВЕЄРА] Не підтверджено: виробник/модель/ревізія конвеєра, двигуна й драйвера, живлення, струми, захист, COM, тип входів/виходів, активні рівні, спосіб регулювання, feedback, реле/оптопари, схема знеструмлення та фактичний безпечний стан; тому hardware adapter відсутній. Що потрібно додати: фото шильдиків, клем і проводки, офіційні datasheet/manual усіх компонентів, затверджену електричну схему MERC-I5, карту сигналів і вимірювання кваліфікованого фахівця. Як завершити: звірити джерела живлення, струмові шляхи, пускові струми, захист індуктивного навантаження, розв'язку, активні рівні й поведінку при втраті живлення; після підписаного протоколу реалізувати окремий adapter і hardware-in-the-loop тести.

[!WARNING] [ПОТРЕБУЄ ДОПРАЦЮВАННЯ: КАЛІБРУВАННЯ ШВИДКОСТІ, FEEDBACK І ТАЙМ-АУТІВ] Не підтверджено: відповідність нормалізованої команди фізичній швидкості, похибка/стабільність, мінімальна робоча команда, час запуску/зупинки, місця сенсорів, очікуваний час перенесення й критерій заклинювання; mock-ticks не є секундами реального стенда. Що потрібно додати: методику безпечного вимірювання, датовані пари «команда — швидкість», сирі feedback/сенсорні журнали для вантажів і граничних умов, допустимі похибки й затверджені тайм-аути. Як завершити: оператор проводить контрольовану серію на мінімальній затвердженій швидкості після повної передпускової перевірки, фахівець оцінює розкид і найгірший час, а параметри проходять повторні інтеграційні тести; тоді mock-профіль доповнюють стендовим.

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

1. Команда не дорівнює стану

Команду запуску треба перевірити feedback у скінченному вікні. Подія приймання завершує transfer. Відсутність події до тайм-ауту — не доказ конкретно механічного заклинювання: це клас transfer_timeout_or_jam, який потребує діагностики.

2. Автомат станів

Текстовий опис рисунка: зі STOPPED валідована команда переходить у STARTING; підтвердження руху веде в RUNNING; подія приймання спричиняє команду stop та повернення до STOPPED; тайм-аут, втрата деталі/зв'язку або непідтверджений старт ведуть у FAULT, де виконується спроба safe stop і потрібне ручне рішення.

stateDiagram-v2
    [*] --> STOPPED
    STOPPED --> STARTING: валідована команда > 0
    STARTING --> RUNNING: running feedback
    STARTING --> FAULT: немає feedback / connection lost
    RUNNING --> STOPPING: accept event
    RUNNING --> FAULT: timeout / lost part / connection lost
    STOPPING --> STOPPED: stop confirmed
    STOPPING --> FAULT: stop unconfirmed
    FAULT --> STOPPED: лише новий mock або ручна процедура поза цією ЛР

Рис. 1. Ілюстративний автомат станів конвеєрного transfer без автоматичного відновлення

3. Python API

Нормалізована команда зручна для API, однак майбутній adapter має документувати одиниці, нелінійність, насичення та те, що означає 0.0 для реального приводу.

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

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

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

Крок 1. Специфікувати паспорт і стани

Опишіть призначення, входи/виходи, стани STOPPED/STARTING/RUNNING/STOPPING/FAULT, нормалізовану швидкість, feedback, тайм-аути, safe-state та процедуру відновлення.

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

Очікуваний результат: паспорт не містить фізичних пінів чи швидкостей; кожен перехід має тригер, тайм-аут і помилку.

Критерій правильності: FAULT не переходить автоматично в RUNNING; connection lost не вважається підтвердженим stop.

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

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

Збережіть як lab06-conveyor.json. Tick-параметри — дискретні кроки симуляції, не реальний час.

{
  "mode": "mock",
  "min_speed": 0.1,
  "max_speed": 1.0,
  "start_confirm_ticks": 2,
  "stop_confirm_ticks": 2,
  "transfer_timeout_ticks": 4
}

python -m json.tool lab06-conveyor.json

Очікуваний результат: валідний несекретний JSON без GPIO, напруги чи фізичних одиниць швидкості.

Критерій правильності: 0 < min_speed <= max_speed <= 1; усі tick-параметри — додатні цілі.

Якщо результат не отримано: виправте формат/межі; не замінюйте tick реальними секундами без даних.

Крок 3. Реалізувати mock-конвеєр і контрольований transfer

Збережіть як lab06_conveyor.py. Без аргументів код перевіряє нормальний цикл і три відмови; обладнання й мережа не використовуються.

# safe-smoke
from __future__ import annotations

import json
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Mapping


class ConveyorState(str, Enum):
    STOPPED = "STOPPED"
    STARTING = "STARTING"
    RUNNING = "RUNNING"
    STOPPING = "STOPPING"
    FAULT = "FAULT"


class ConveyorFault(RuntimeError):
    def __init__(self, reason: str, stop_confirmed: bool | None = None) -> None:
        super().__init__(reason)
        self.reason = reason
        self.stop_confirmed = stop_confirmed


@dataclass(frozen=True)
class Config:
    mode: str
    min_speed: float
    max_speed: float
    start_confirm_ticks: int
    stop_confirm_ticks: int
    transfer_timeout_ticks: int


def parse_config(raw: Mapping[str, object]) -> Config:
    if raw.get("mode") != "mock":
        raise ValueError("only mode=mock is permitted")
    min_speed, max_speed = raw.get("min_speed"), raw.get("max_speed")
    if any(isinstance(v, bool) or not isinstance(v, (int, float)) for v in (min_speed, max_speed)):
        raise ValueError("speed limits must be numbers")
    if not 0 < float(min_speed) <= float(max_speed) <= 1:
        raise ValueError("require 0 < min_speed <= max_speed <= 1")
    tick_names = ("start_confirm_ticks", "stop_confirm_ticks", "transfer_timeout_ticks")
    ticks = []
    for name in tick_names:
        value = raw.get(name)
        if not isinstance(value, int) or isinstance(value, bool) or not 1 <= value <= 1000:
            raise ValueError(f"{name} must be an integer from 1 to 1000")
        ticks.append(value)
    return Config("mock", float(min_speed), float(max_speed), *ticks)


class MockConveyorBackend:
    """Детермінований backend; ticks не є реальними секундами."""

    def __init__(self, scenario: str, complete_after: int = 3) -> None:
        if scenario not in {"normal", "jam", "lost_part", "connection_lost"}:
            raise ValueError("unknown mock scenario")
        self.scenario = scenario
        self.complete_after = complete_after
        self.tick_count = 0
        self.connected = True
        self.requested_speed = 0.0

    def command(self, speed: float) -> None:
        if not self.connected:
            raise ConveyorFault("connection_lost")
        if not 0 <= speed <= 1:
            raise ValueError("backend speed must be 0..1")
        self.requested_speed = float(speed)

    def tick(self) -> dict[str, object]:
        self.tick_count += 1
        if self.scenario == "connection_lost" and self.tick_count >= 2:
            self.connected = False
        connected = self.connected
        running = self.requested_speed > 0 if connected else None
        part_present = not (self.scenario == "lost_part" and self.tick_count >= 2)
        accepted = (
            self.scenario == "normal"
            and self.requested_speed > 0
            and self.tick_count >= self.complete_after
        )
        return {
            "connected": connected,
            "running_feedback": running,
            "part_present": part_present,
            "accept_sensor_changed": accepted,
            "tick": self.tick_count,
        }


class ConveyorModule:
    def __init__(self, backend: MockConveyorBackend, config: Config) -> None:
        self.backend = backend
        self.config = config
        self.state = ConveyorState.STOPPED

    def _validate_speed(self, speed: float) -> float:
        if isinstance(speed, bool) or not isinstance(speed, (int, float)):
            raise ValueError("speed must be numeric")
        value = float(speed)
        if not self.config.min_speed <= value <= self.config.max_speed:
            raise ValueError("speed outside configured normalized range")
        return value

    def safe_stop(self) -> bool:
        self.state = ConveyorState.STOPPING
        try:
            self.backend.command(0.0)
        except ConveyorFault:
            self.state = ConveyorState.FAULT
            return False
        for _ in range(self.config.stop_confirm_ticks):
            snapshot = self.backend.tick()
            if snapshot["connected"] is True and snapshot["running_feedback"] is False:
                self.state = ConveyorState.STOPPED
                return True
        self.state = ConveyorState.FAULT
        return False

    def transfer(self, speed: float) -> dict[str, object]:
        if self.state is not ConveyorState.STOPPED:
            raise ConveyorFault("module_not_stopped")
        value = self._validate_speed(speed)
        reason: str | None = None
        completed_tick: int | None = None
        stop_confirmed = False
        try:
            self.state = ConveyorState.STARTING
            self.backend.command(value)
            started = False
            for _ in range(self.config.start_confirm_ticks):
                snapshot = self.backend.tick()
                if snapshot["connected"] is not True:
                    reason = "connection_lost"
                    break
                if snapshot["running_feedback"] is True:
                    started = True
                    self.state = ConveyorState.RUNNING
                    break
            if reason is None and not started:
                reason = "start_not_confirmed"

            if reason is None:
                for _ in range(self.config.transfer_timeout_ticks):
                    snapshot = self.backend.tick()
                    if snapshot["connected"] is not True:
                        reason = "connection_lost"
                        break
                    if snapshot["part_present"] is not True:
                        reason = "part_lost"
                        break
                    if snapshot["accept_sensor_changed"] is True:
                        completed_tick = int(snapshot["tick"])
                        break
                if reason is None and completed_tick is None:
                    reason = "transfer_timeout_or_jam"
        except ConveyorFault as exc:
            reason = exc.reason
        finally:
            stop_confirmed = self.safe_stop()

        if reason is not None:
            self.state = ConveyorState.FAULT
            raise ConveyorFault(reason, stop_confirmed)
        if not stop_confirmed:
            self.state = ConveyorState.FAULT
            raise ConveyorFault("stop_not_confirmed", False)
        return {
            "result": "COMPLETE",
            "completed_tick": completed_tick,
            "stop_confirmed": True,
            "final_state": self.state.value,
            "hardware_tested": False,
        }


def smoke_test() -> dict[str, object]:
    config = parse_config({
        "mode": "mock", "min_speed": 0.1, "max_speed": 1.0,
        "start_confirm_ticks": 2, "stop_confirm_ticks": 2,
        "transfer_timeout_ticks": 4,
    })
    normal = ConveyorModule(MockConveyorBackend("normal"), config)
    result = normal.transfer(0.5)
    assert result["final_state"] == "STOPPED"
    assert normal.backend.requested_speed == 0.0

    failures: dict[str, dict[str, object]] = {}
    for scenario, expected in (
        ("jam", "transfer_timeout_or_jam"),
        ("lost_part", "part_lost"),
        ("connection_lost", "connection_lost"),
    ):
        module = ConveyorModule(MockConveyorBackend(scenario), config)
        try:
            module.transfer(0.5)
        except ConveyorFault as exc:
            assert exc.reason == expected
            failures[scenario] = {"reason": exc.reason, "stop_confirmed": exc.stop_confirmed}
        else:
            raise AssertionError(f"{scenario} must fail")
    assert failures["jam"]["stop_confirmed"] is True
    assert failures["lost_part"]["stop_confirmed"] is True
    assert failures["connection_lost"]["stop_confirmed"] is False

    try:
        ConveyorModule(MockConveyorBackend("normal"), config).transfer(0.0)
    except ValueError:
        invalid_speed_rejected = True
    else:
        raise AssertionError("zero transfer speed must be rejected")
    return {"normal": result, "failures": failures,
            "invalid_speed_rejected": invalid_speed_rejected}


def main() -> int:
    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: lab06_conveyor.py [--config FILE]")
    print(json.dumps(smoke_test(), ensure_ascii=False))
    return 0


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

python lab06_conveyor.py
python lab06_conveyor.py --config lab06-conveyor.json

Очікуваний результат: нормальний сценарій має COMPLETE, STOPPED, stop_confirmed=true; jam і lost part завершуються fault із підтвердженим mock-stop; connection lost — stop_confirmed=false; нульова transfer-швидкість відхилена.

Критерій правильності: код завершення smoke-тесту 0; усі чотири сценарії та межа швидкості відповідають очікуванню; після сценаріїв із доступним зв'язком команда дорівнює 0.0.

Якщо результат не отримано: виведіть snapshots/ticks у копії, перевірте перший незбіжний reason; не додавайте sleep, GPIO або реальний привод.

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

Перевірте speed=min_speed, speed=max_speed, трохи нижче/вище межі, bool, start_confirm_ticks=0, невідомий сценарій, transfer після примусового FAULT і stop за втрати зв'язку.

Код/команда: python lab06_conveyor.py після додавання локальних assert.

Очікуваний результат: дві валідні межі проходять; шість некоректних/небезпечних варіантів відхиляються; stop за втрати зв'язку лишається непідтвердженим.

Критерій правильності: 8/8 тестів; немає автоматичного переходу FAULT→RUNNING; усі цикли скінченні.

Якщо результат не отримано: створюйте новий module/backend на тест, перевірте reason і не скидайте state вручну.

Крок 5. Оформити паспорт, журнал і метрики

Заповніть паспорт модуля, таблицю кожного snapshot для нормального й одного fault-сценарію, фінальний state, stop confirmation, кількість ticks і межі апаратної перевірки.

Код/команда: використайте JSON-вивід; команда фізичному обладнанню не потрібна.

Очікуваний результат: журнал дає змогу відтворити причину переходу; connection lost відрізнено від підтвердженого stop.

Критерій правильності: кожна подія має tick, команду, feedback, part/accept, стан і reason; немає фізичних секунд чи швидкостей без джерела.

Якщо результат не отримано: додайте структурований snapshot у mock; не реконструюйте відсутні дані «на око».

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

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

Трасування transfer

Підсумок

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

Дотримуйтеся загальних вимог. Додатково подайте паспорт модуля, JSON, код, автомат станів, snapshot-таблиці normal і fault, результати чотирьох базових та восьми розширених тестів, аналіз stop_confirmed=false, правила ручного відновлення й фразу «Конвеєр, електричний інтерфейс, фізичну швидкість і час зупинки не перевіряли».

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

Критична помилка: фізичний запуск/заклинювання, вигаданий pinout/таймінг, автоматичний restart або хибне stop_confirmed=true за втрати зв'язку.

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

1. Чому команда швидкості не є підтвердженням руху?
2. Яку роль має сигнал приймання деталі?
3. Чому timeout не доводить саме механічне заклинювання?
4. Навіщо нормалізувати швидкість і що має зробити hardware adapter?
5. Чому stop за втрати зв'язку вважається непідтвердженим?
6. Для чого finally викликає safe_stop()?
7. Чому FAULT не скидається автоматично?
8. Які незалежні засоби можуть бути потрібні для реального safe-state?
9. Які дані потрібні для калібрування команди у см/с?
10. Чого не доводять успішні mock-тести?

Висновки

У висновку наведіть результати normal/jam/lost part/connection lost, межі швидкості, кількість ticks і стан stop confirmation. Поясніть, як модуль відмовляє за замовчуванням і чому за втрати зв'язку потрібен оператор/незалежний захист. Не заявляйте фізичну швидкість, час зупинки, електричну сумісність або апаратну апробацію.