Caso práctico: control NFC con Raspberry Pi 4

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import json
import os
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, List, Optional


@dataclass
class AccessEvent:
    timestamp: str
    uid: str
    authorized: bool
    source: str
    alarm_active: bool


class RTCAdapter:
    def now_iso(self) -> str:
        raise NotImplementedError


class SystemRTC(RTCAdapter):
    def now_iso(self) -> str:
        return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


class MockDS3231RTC(RTCAdapter):
    def __init__(self, fixed_time: Optional[str] = None) -> None:
        self.fixed_time = fixed_time

    def now_iso(self) -> str:
        if self.fixed_time:
            return self.fixed_time
        return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")


class NFCReaderAdapter:
    def poll_uid(self) -> Optional[str]:
        raise NotImplementedError


class MockPN532Reader(NFCReaderAdapter):
    def __init__(self, sequence: Iterable[str], repeat: bool = False) -> None:
        self._sequence = list(sequence)
        self._repeat = repeat
        self._index = 0

    def poll_uid(self) -> Optional[str]:
        if not self._sequence:
            return None
        if self._index >= len(self._sequence):
            if self._repeat:
                self._index = 0
            else:
                return None
        uid = self._sequence[self._index]
        self._index += 1
        time.sleep(0.2)
        return uid


class AlarmAdapter:
    def set_alarm(self, active: bool) -> None:
        raise NotImplementedError


class ConsoleAlarm(AlarmAdapter):
    def __init__(self) -> None:
        self.state = False

    def set_alarm(self, active: bool) -> None:
        if active != self.state:
            self.state = active
            print(f"[ALARM] state={'ON' if active else 'OFF'}")


class AccessController:
    def __init__(
        self,
        rtc: RTCAdapter,
        reader: NFCReaderAdapter,
        alarm: AlarmAdapter,
        allowed_uids: List[str],
        log_path: Path,
    ) -> None:
        self.rtc = rtc
        self.reader = reader
        self.alarm = alarm
        self.allowed_uids = {uid.strip().upper() for uid in allowed_uids if uid.strip()}
        self.log_path = log_path
        self.alarm_active = False

    def handle_uid(self, uid: str, source: str = "nfc") -> AccessEvent:
        normalized = uid.strip().upper()
        authorized = normalized in self.allowed_uids
        self.alarm_active = not authorized
        self.alarm.set_alarm(self.alarm_active)

        event = AccessEvent(
            timestamp=self.rtc.now_iso(),
            uid=normalized,
            authorized=authorized,
            source=source,
            alarm_active=self.alarm_active,
        )
        self._append_log(event)
        if authorized:
            print(f"{event.timestamp} ACCESS GRANTED uid={event.uid}")
        else:
            print(f"{event.timestamp} ACCESS DENIED uid={event.uid}")
        return event

    def _append_log(self, event: AccessEvent) -> None:
        file_exists = self.log_path.exists()
        with self.log_path.open("a", newline="", encoding="utf-8") as f:
            writer = csv.writer(f)
            if not file_exists:
                writer.writerow(["timestamp", "uid", "authorized", "source", "alarm_active"])
            writer.writerow([
                event.timestamp,
                event.uid,
                int(event.authorized),
                event.source,
                int(event.alarm_active),
            ])

    def run(self, max_reads: int = 0, poll_delay: float = 0.5) -> int:
        reads = 0
        while True:
            uid = self.reader.poll_uid()
            if uid:
                self.handle_uid(uid)
                reads += 1
            else:
                time.sleep(poll_delay)

            if max_reads > 0 and reads >= max_reads:
                break
        return 0


def load_allowed_uids(path: Path) -> List[str]:
    with path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict) or "allowed_uids" not in data:
        raise ValueError("allowlist file must contain a JSON object with key 'allowed_uids'")
    items = data["allowed_uids"]
    if not isinstance(items, list):
        raise ValueError("'allowed_uids' must be a list")
    return [str(x) for x in items]


def build_mock_sequence(args: argparse.Namespace) -> List[str]:
    if args.mock_sequence:
        return [x.strip().upper() for x in args.mock_sequence.split(",") if x.strip()]
    return [
        "04A1B2C3D4",
        "1122334455",
        "04A1B2C3D4",
    ]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="NFC door access alarm prototype")
    parser.add_argument("--allowlist", default="allowlist.json", help="Path to JSON allowlist")
    parser.add_argument("--log", default="access_log.csv", help="Path to CSV log file")
    parser.add_argument("--mock", action="store_true", help="Use mock NFC and RTC adapters")
    parser.add_argument("--mock-sequence", default="", help="Comma-separated UID sequence for mock mode")
    parser.add_argument("--max-reads", type=int, default=3, help="Stop after this many successful reads, 0=forever")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    allowlist_path = Path(args.allowlist)
    log_path = Path(args.log)

    allowed_uids = load_allowed_uids(allowlist_path)

    if args.mock:
        rtc: RTCAdapter = MockDS3231RTC()
        reader: NFCReaderAdapter = MockPN532Reader(build_mock_sequence(args), repeat=False)
    else:
        # real-hardware adapter integration is intentionally not auto-imported here.
        # If no mock mode is selected, use system clock and require explicit future extension
        # for PN532 hardware reading.
        rtc = SystemRTC()
        reader = MockPN532Reader([], repeat=False)

    alarm = ConsoleAlarm()
    controller = AccessController(
        rtc=rtc,
        reader=reader,
        alarm=alarm,
        allowed_uids=allowed_uids,
        log_path=log_path,
    )
    return controller.run(max_reads=args.max_reads)


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
from __future__ import annotations

import csv
import subprocess
import sys
from pathlib import Path


def main() -> int:
    project_dir = Path(__file__).resolve().parent
    log_path = project_dir / "test_access_log.csv"

    if log_path.exists():
        log_path.unlink()

    cmd = [
        sys.executable,
        str(project_dir / "access_controller.py"),
        "--mock",
        "--allowlist",
        str(project_dir / "allowlist.json"),
        "--log",
        str(log_path),
        "--mock-sequence",
        "04A1B2C3D4,DEADBEEF01",
        "--max-reads",
        "2",
    ]

    result = subprocess.run(cmd, capture_output=True, text=True, check=False)
    print(result.stdout)
    if result.returncode != 0:
        print(result.stderr)
        return result.returncode

    if not log_path.exists():
        print("ERROR: log file was not created")
        return 1

    with log_path.open("r", encoding="utf-8", newline="") as f:
        rows = list(csv.DictReader(f))

    if len(rows) != 2:
        print(f"ERROR: expected 2 log entries, got {len(rows)}")
        return 1

    if rows[0]["authorized"] != "1":
        print("ERROR: first UID should be authorized")
        return 1

    if rows[1]["authorized"] != "0":
        print("ERROR: second UID should be unauthorized")
        return 1

    if rows[1]["alarm_active"] != "1":
        print("ERROR: alarm should be active for unauthorized UID")
        return 1

    print("Dry-run validation passed.")
    return 0


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