diff --git a/cloud/app.py b/cloud/app.py
index d1bade8..4eeb5d4 100644
--- a/cloud/app.py
+++ b/cloud/app.py
@@ -6,6 +6,7 @@ import base64
import asyncio
import fcntl
import atexit
+import re
from datetime import datetime, timedelta, timezone
from flask import Flask, Response, request, jsonify, current_app, abort, g
from pymongo import MongoClient, ASCENDING, DESCENDING
@@ -45,6 +46,7 @@ alert_collection = db["alert_data"]
webex_tokens_collection = db["webex_tokens"]
device_ip_collection = db["device_ips"]
cve_audit_results = db["cve_audit_results"]
+rfid_cards_collection = db["rfid_cards"]
# Initialize Database Indexes
def init_db_indexes():
@@ -490,6 +492,52 @@ def run_cve_audit():
"timestamp": datetime.now(timezone.utc).isoformat()
}), status_code
+@app.route("/rfid-card", methods=["GET"])
+def get_rfid_cards():
+ """OData compliant endpoint for technician RFID card validation."""
+ filter_str = request.args.get("$filter", "")
+ card_id = request.args.get("card_id")
+
+ mongo_query = {}
+
+ # Extract card_id from OData $filter string if not supplied directly
+ if not card_id and filter_str:
+ match = re.search(r"(?:card_id|cardId)\s+eq\s+['\"]([^'\"]+)['\"]", filter_str, re.IGNORECASE)
+ if match:
+ card_id = match.group(1)
+ else:
+ # Generic fallback to extract quoted strings
+ generic_match = re.search(r"['\"]([a-zA-Z0-9]+)['\"]", filter_str)
+ if generic_match:
+ card_id = generic_match.group(1)
+
+ if card_id:
+ mongo_query["card_id"] = card_id
+
+ # Filter for active/valid cards only
+ mongo_query["valid"] = True
+
+ try:
+ cards = list(rfid_cards_collection.find(mongo_query))
+
+ result_value = [
+ {
+ "id": str(card["_id"]),
+ "card_id": card.get("card_id"),
+ "valid": card.get("valid", False)
+ }
+ for card in cards
+ ]
+
+ return jsonify({
+ "@odata.context": f"{request.host_url.rstrip('/')}/$metadata#RfidCards",
+ "value": result_value
+ }), 200
+
+ except Exception as e:
+ current_app.logger.exception("Error querying rfid_cards collection")
+ return jsonify({"@odata.error": {"code": "500", "message": "Database query error"}}), 500
+
# ---------------------------------------------------------
# Debug Endpoints (Protected)
# ---------------------------------------------------------
@@ -570,11 +618,18 @@ def odata_metadata():
+
+
+
+
+
+
+
diff --git a/orchestrateur/README.md b/orchestrateur/README.md
index d32e0a7..88f4ffc 100644
--- a/orchestrateur/README.md
+++ b/orchestrateur/README.md
@@ -2,4 +2,11 @@
# IP Adresses
- AP (dynamic) : `10.110.0.190`
-- Generated Wifi : `192.168.50.1`
\ No newline at end of file
+- Generated Wifi : `192.168.50.1`
+
+# RFID
+
+## Cards
+
+- 2700423026
+- 270042302B
\ No newline at end of file
diff --git a/orchestrateur/main.py b/orchestrateur/main.py
index 4daedec..52bbae3 100644
--- a/orchestrateur/main.py
+++ b/orchestrateur/main.py
@@ -5,13 +5,12 @@ import asyncio
import requests
from shared.microwave_state import MicrowaveState, MicrowaveStateFields
-from orchestrateur.sensors import gps
from shared import get_lora, get_mqtt_client, deviceTypes, config, payloads, db
from shared.logging import log
from shared.cookingState import CookingStates
from shared.lora_device import LoraCommands
from shared.alerts import AlertManager, Alert, AlertType
-from sensors import ultrasonicRanger, temp_hum, button, camera, buzzer, rgb_lcd
+from sensors import ultrasonicRanger, temp_hum, button, camera, buzzer, rgb_lcd, gps, rfid_reader
from lib.systemd_logs import get_systemd_logs
from external_display_manager import DisplayManager
from rgb_lcd_manager import RGBLCDManager
@@ -21,6 +20,7 @@ rgb_lcd.setText("SmartWave\nOrchestrateur")
# --- CONFIGURATION CONSTANTS ---
DISPLAY_DEBOUNCE_SECONDS = 3 # Seconds to wait after first change before broadcasting
+TECHNICIAN_MODE = False
# --- DB SETUP ---
DB_PATH = "orchestrateur/db.sqlite"
@@ -145,6 +145,9 @@ mqtt_client.connect()
mqtt_client.subscribe(config.MQTT_TOPIC_SENSOR, qos=config.MQTT_QOS)
mqtt_client.subscribe(config.MQTT_TOPIC_HELLO, qos=config.MQTT_QOS)
+# RFID reader
+rfid = rfid_reader.RFIDReader(rx_pin=17)
+
# DHT sensor warmup
time.sleep(2) # Allow GrovePi MCU and DHT sensor circuits to settle post-service start
temp_hum.get_temperature_and_humidity() # Dummy read to flush initial NaN state
@@ -174,7 +177,7 @@ if hasattr(cloud_mqtt_client._client, "loop_start"):
print("[MQTT Cloud] Paho background loop started.")
-# --- BACKGROUND TASKS (PRODUCERS) ---
+# --- BACKGROUND TASKS ---
async def lora_listener_task():
"""Polls LoRa and pushes to the async queue."""
print("[LoRa] Async listener started.")
@@ -230,32 +233,47 @@ async def cloud_mqtt_listener_task():
})
await asyncio.sleep(0.1)
-# === LCD DISPLAY ===
-def on_screens_change(screens, is_removed: bool, modified_name: str):
- """Callback for when screens are added or removed."""
- log(f"[DisplayManager] Screens changed. Current screens: {list(screens.keys())}")
- rgb_lcd_manager.set_external_screen_count(len(screens))
+async def rfid_listener_task():
+ """Polls RFID reader, verifies cards via OData API, and toggles TECHNICIAN_MODE."""
+ global TECHNICIAN_MODE
+ print("[RFID] Async task started on GPIO 17.")
- # Add Or remove from database of connected components
- # 1. Extract id from name (assuming format "smartwave-epaper-._displaytcp._tcp.local")
- display_id = modified_name.split("-")[-1].split(".")[0]
- if is_removed:
- remove_connected_component(display_id)
- else:
- save_connected_component(display_id, "external_display")
-
-display_manager.set_on_screens_change_callback(on_screens_change)
+ last_scanned_tag = None
+ last_scan_time = 0
-def update_lcd_microwave_count():
- """Updates the LCD with the current number of connected microwaves."""
- global microwave_states
- count = len(microwave_states)
- rgb_lcd_manager.set_microwave_count(count)
-
-def update_lcd_cloud_alert():
- """Updates the LCD with the current cloud connectivity status."""
- global cloud_alert
- rgb_lcd_manager.set_cloud_alert(cloud_alert)
+ while True:
+ try:
+ # Quick non-blocking read via pigpio IPC
+ tag_id = await asyncio.to_thread(rfid.read_tag)
+ current_time = time.time()
+
+ if tag_id:
+ # Debounce same tag within 3 seconds to avoid spamming the API
+ if tag_id == last_scanned_tag and (current_time - last_scan_time) < 3.0:
+ await asyncio.sleep(0.1)
+ continue
+
+ last_scanned_tag = tag_id
+ last_scan_time = current_time
+ print(f"[RFID] Tag detected: {tag_id}. Verifying with API...")
+
+ is_valid = await check_rfid_card_api(tag_id)
+
+ if is_valid:
+ TECHNICIAN_MODE = True
+ print(f"[RFID] Card {tag_id} VALIDATED! -> TECHNICIAN_MODE = TRUE")
+ # Optional: Sound positive feedback beep
+ buzzer.buzzer_siren(beatsNb=1)
+ else:
+ print(f"[RFID] Card {tag_id} REJECTED / Invalid.")
+ # Optional: Sound error feedback beep
+ buzzer.buzzer_siren(beatsNb=2)
+
+ except Exception as e:
+ print(f"[RFID Task] Error reading RFID tag: {e}")
+
+ # Sleep 100ms between checks to keep CPU usage near zero
+ await asyncio.sleep(0.1)
async def display_broadcast_worker_task():
"""
@@ -288,7 +306,137 @@ async def display_broadcast_worker_task():
microwave_states=microwave_states,
cloud_alert=cloud_alert
)
+
+async def process_messages_task():
+ """Consumes the unified event queue."""
+ # Helper to normalize config topics to str
+ def to_str(val):
+ return val.decode('utf-8') if isinstance(val, bytes) else val
+
+ while True:
+ msg = await async_event_queue.get()
+ source = msg["source"]
+ data = msg["data"]
+
+ if source == "LoRa":
+ if "new_cooking_state" in data.get("data", {}):
+ mw_id = data["data"].get("id")
+ n_state = data["data"].get("new_cooking_state")
+ print(f"[LoRa] Microwave {mw_id} state changed to: {n_state}")
+
+ microwave_states[mw_id].set_cooking_state(n_state)
+ update_lcd_microwave_count() # Update the LCD with new microwave count
+
+ if n_state in (CookingStates.DONE, CookingStates.STIRRING_REQUIRED, CookingStates.ALERT):
+ beats = 5 if n_state == CookingStates.ALERT else (1 if n_state == CookingStates.STIRRING_REQUIRED else 3)
+ buzzer.buzzer_siren(beatsNb=beats)
+ if n_state == CookingStates.IDLE and microwave_states.get(mw_id).state == MicrowaveState.COOKING:
+ microwave_states[mw_id].set_state(MicrowaveState.DONE)
+ print(f"[{mw_id}] Cooking finished. Waiting for user to remove dish.")
+ if "action" in data.get("data", {}):
+ if type(data["data"]) != dict:
+ print(f"[LoRa] Invalid data format received: {data['data']}")
+ continue
+ action = data["data"]["action"]
+ if action == LoraCommands.COOKING_UPDATE:
+ mw_id = data["data"].get("id")
+ if mw_id in microwave_states:
+ state_info = data["data"]
+ microwave_states[mw_id].set_cooking_state(state_info.get("cooking_state", microwave_states[mw_id].cooking_state))
+ microwave_states[mw_id].set_cooking_estimated_remaining_time(state_info.get("estimated_remaining_time", microwave_states[mw_id].cooking_estimated_remaining_time))
+ microwave_states[mw_id].set_paused(state_info.get("paused", microwave_states[mw_id].paused))
+ elif action == LoraCommands.NEW_ALERT:
+ alert_type = data["data"].get("alert_type")
+ alert_message = data["data"].get("alert_message")
+ alert = Alert(alert_type, alert_message, redistributed=True)
+ alert_manager.add_alert(alert)
+ print(f"[LoRa] New alert received: {alert.to_dict()}")
+ else:
+ print(f"[LoRa] Unhandled action received: {action} with data: {data['data']}")
+ elif source == "MQTT":
+ topic = msg["topic"]
+ hello_topic = to_str(config.MQTT_TOPIC_HELLO)
+ sensor_topic = to_str(config.MQTT_TOPIC_SENSOR)
+
+ if topic == hello_topic:
+ if data.get("id_orchestrator") != DEVICE_ID:
+ component_id = data.get("id_microwave")
+ component_type = data.get("type", deviceTypes.DEVICE_TYPES["MICROWAVE"])
+
+ if component_id:
+ print(f"[MQTT] Hello received from '{component_id}' ({component_type}). Updating DB & sending ACK.")
+ # Offload DB insertion to async thread execution pool
+ await asyncio.to_thread(save_connected_component, component_id, component_type)
+
+ # Pass change callback to newly connected microwave
+ microwave_states[component_id] = MicrowaveState(
+ component_id,
+ on_change_callback=on_microwave_change
+ )
+ notify_display_update() # Update the external screens
+ update_lcd_microwave_count() # Update the LCD with new microwave count
+
+ mqtt_client.publish(
+ config.MQTT_TOPIC_HELLO,
+ payloads.mqtt_hello_ack(DEVICE_ID, component_id),
+ qos=config.MQTT_QOS
+ )
+
+ elif topic == sensor_topic:
+ mw_id = str(data.get("id_microwave"))
+ print(f"[MQTT] Sensor data received for microwave {mw_id}: {data}")
+
+ # Store IR data and notify the waiting dish handler
+ ir_data_cache[mw_id] = data
+ if mw_id in ir_data_events:
+ ir_data_events[mw_id].set()
+
+ elif source == "CloudMQTT":
+ topic = msg["topic"]
+ print(f"[CloudMQTT] Command received on '{topic}': {data}")
+
+ action = data.get("action")
+
+ if action == "request_telemetry":
+ endpoint = data.get("endpoint")
+ if endpoint:
+ # Schedule as a background task so processing the queue isn't stalled
+ do_timeout = topic == "cmd/all"
+ asyncio.create_task(handle_telemetry_request(endpoint, do_timeout=do_timeout))
+ else:
+ print("[CloudMQTT] Received 'request_telemetry' but missing 'endpoint' field.")
+ else:
+ print(f"[CloudMQTT] Unknown action: {action}")
+
+# === LCD DISPLAY ===
+def on_screens_change(screens, is_removed: bool, modified_name: str):
+ """Callback for when screens are added or removed."""
+ log(f"[DisplayManager] Screens changed. Current screens: {list(screens.keys())}")
+ rgb_lcd_manager.set_external_screen_count(len(screens))
+
+ # Add Or remove from database of connected components
+ # 1. Extract id from name (assuming format "smartwave-epaper-._displaytcp._tcp.local")
+ display_id = modified_name.split("-")[-1].split(".")[0]
+ if is_removed:
+ remove_connected_component(display_id)
+ else:
+ save_connected_component(display_id, "external_display")
+
+display_manager.set_on_screens_change_callback(on_screens_change)
+
+def update_lcd_microwave_count():
+ """Updates the LCD with the current number of connected microwaves."""
+ global microwave_states
+ count = len(microwave_states)
+ rgb_lcd_manager.set_microwave_count(count)
+
+def update_lcd_cloud_alert():
+ """Updates the LCD with the current cloud connectivity status."""
+ global cloud_alert
+ rgb_lcd_manager.set_cloud_alert(cloud_alert)
+
+# --- Button ---
def button_callback():
"""Button physical interrupt callback."""
global button_state
@@ -428,6 +576,44 @@ async def handle_new_dish(microwave_id, detected_height):
finally:
ir_data_events.pop(microwave_id, None)
+# === API CALLS ===
+
+async def check_rfid_card_api(card_id: str) -> bool:
+ """Queries the OData API asynchronously to verify if card_id is valid."""
+ url = "https://smartwave.matthiasg.dev/rfid-card"
+
+ params = {
+ "$filter": f"card_id eq '{card_id}'"
+ }
+ headers = {"Content-Type": "application/json"}
+ if API_KEY:
+ headers["X-API-Key"] = API_KEY
+
+ try:
+ response = await asyncio.to_thread(
+ requests.get,
+ url,
+ params=params,
+ headers=headers,
+ timeout=5
+ )
+ if response.status_code == 200:
+ set_cloud_alert(False) # Clear cloud alert on success
+ data = response.json()
+
+ if isinstance(data, dict):
+ items = data.get("value", data.get("data", []))
+ return len(items) > 0 if isinstance(items, list) else bool(data)
+ elif isinstance(data, list):
+ return len(data) > 0
+
+ return False
+
+ except Exception as e:
+ print(f"[RFID Task] OData API check error: {e}")
+ set_cloud_alert(True) # Set cloud alert on API connection failure
+ return False
+
async def request_cloud_cooking_plan(microwave_id, sensors_data):
"""Sends all data to the cloud with up to 3 retries (330s timeout for AI generation)."""
global cloud_alert, alert_manager
@@ -535,107 +721,6 @@ async def request_cloud_cooking_plan(microwave_id, sensors_data):
microwave_states[microwave_id].set_state(MicrowaveState.DONE)
# --- MAIN LOGIC TASKS ---
-async def process_messages_task():
- """Consumes the unified event queue."""
- # Helper to normalize config topics to str
- def to_str(val):
- return val.decode('utf-8') if isinstance(val, bytes) else val
-
- while True:
- msg = await async_event_queue.get()
- source = msg["source"]
- data = msg["data"]
-
- if source == "LoRa":
- if "new_cooking_state" in data.get("data", {}):
- mw_id = data["data"].get("id")
- n_state = data["data"].get("new_cooking_state")
- print(f"[LoRa] Microwave {mw_id} state changed to: {n_state}")
-
- microwave_states[mw_id].set_cooking_state(n_state)
- update_lcd_microwave_count() # Update the LCD with new microwave count
-
- if n_state in (CookingStates.DONE, CookingStates.STIRRING_REQUIRED, CookingStates.ALERT):
- beats = 5 if n_state == CookingStates.ALERT else (1 if n_state == CookingStates.STIRRING_REQUIRED else 3)
- buzzer.buzzer_siren(beatsNb=beats)
- if n_state == CookingStates.IDLE and microwave_states.get(mw_id).state == MicrowaveState.COOKING:
- microwave_states[mw_id].set_state(MicrowaveState.DONE)
- print(f"[{mw_id}] Cooking finished. Waiting for user to remove dish.")
- if "action" in data.get("data", {}):
- if type(data["data"]) != dict:
- print(f"[LoRa] Invalid data format received: {data['data']}")
- continue
- action = data["data"]["action"]
- if action == LoraCommands.COOKING_UPDATE:
- mw_id = data["data"].get("id")
- if mw_id in microwave_states:
- state_info = data["data"]
- microwave_states[mw_id].set_cooking_state(state_info.get("cooking_state", microwave_states[mw_id].cooking_state))
- microwave_states[mw_id].set_cooking_estimated_remaining_time(state_info.get("estimated_remaining_time", microwave_states[mw_id].cooking_estimated_remaining_time))
- microwave_states[mw_id].set_paused(state_info.get("paused", microwave_states[mw_id].paused))
- elif action == LoraCommands.NEW_ALERT:
- alert_type = data["data"].get("alert_type")
- alert_message = data["data"].get("alert_message")
- alert = Alert(alert_type, alert_message, redistributed=True)
- alert_manager.add_alert(alert)
- print(f"[LoRa] New alert received: {alert.to_dict()}")
- else:
- print(f"[LoRa] Unhandled action received: {action} with data: {data['data']}")
- elif source == "MQTT":
- topic = msg["topic"]
-
- hello_topic = to_str(config.MQTT_TOPIC_HELLO)
- sensor_topic = to_str(config.MQTT_TOPIC_SENSOR)
-
- if topic == hello_topic:
- if data.get("id_orchestrator") != DEVICE_ID:
- component_id = data.get("id_microwave")
- component_type = data.get("type", deviceTypes.DEVICE_TYPES["MICROWAVE"])
-
- if component_id:
- print(f"[MQTT] Hello received from '{component_id}' ({component_type}). Updating DB & sending ACK.")
- # Offload DB insertion to async thread execution pool
- await asyncio.to_thread(save_connected_component, component_id, component_type)
-
- # Pass change callback to newly connected microwave
- microwave_states[component_id] = MicrowaveState(
- component_id,
- on_change_callback=on_microwave_change
- )
- notify_display_update() # Update the external screens
- update_lcd_microwave_count() # Update the LCD with new microwave count
-
- mqtt_client.publish(
- config.MQTT_TOPIC_HELLO,
- payloads.mqtt_hello_ack(DEVICE_ID, component_id),
- qos=config.MQTT_QOS
- )
-
- elif topic == sensor_topic:
- mw_id = str(data.get("id_microwave"))
- print(f"[MQTT] Sensor data received for microwave {mw_id}: {data}")
-
- # Store IR data and notify the waiting dish handler
- ir_data_cache[mw_id] = data
- if mw_id in ir_data_events:
- ir_data_events[mw_id].set()
-
- elif source == "CloudMQTT":
- topic = msg["topic"]
- print(f"[CloudMQTT] Command received on '{topic}': {data}")
-
- action = data.get("action")
-
- if action == "request_telemetry":
- endpoint = data.get("endpoint")
- if endpoint:
- # Schedule as a background task so processing the queue isn't stalled
- do_timeout = topic == "cmd/all"
- asyncio.create_task(handle_telemetry_request(endpoint, do_timeout=do_timeout))
- else:
- print("[CloudMQTT] Received 'request_telemetry' but missing 'endpoint' field.")
- else:
- print(f"[CloudMQTT] Unknown action: {action}")
async def handle_telemetry_request(endpoint: str, do_timeout=True):
"""
@@ -787,7 +872,8 @@ async def main():
cloud_mqtt_listener_task(),
process_messages_task(),
monitor_dish_height_task(),
- display_broadcast_worker_task()
+ display_broadcast_worker_task(),
+ rfid_listener_task() # <-- ADDED HERE
)
if __name__ == "__main__":
@@ -796,6 +882,7 @@ if __name__ == "__main__":
except KeyboardInterrupt:
print("\nArrêt manuel.")
finally:
+ rfid.close() # <-- ADDED CLEANUP HERE
asyncio.run(display_manager.stop())
if hasattr(mqtt_client._client, "loop_stop"):
mqtt_client._client.loop_stop()
diff --git a/orchestrateur/requirements.txt b/orchestrateur/requirements.txt
index 66d4003..d067341 100644
--- a/orchestrateur/requirements.txt
+++ b/orchestrateur/requirements.txt
@@ -4,4 +4,6 @@ pyserial>=3.5,<4
# OpenCV
# sudo apt install -y python3-opencv
# sudo apt install -y opencv-data
-zeroconf>=0.131.0
\ No newline at end of file
+zeroconf>=0.131.0
+# sudo apt install pigpio python3-pigpio
+# sudo systemctl enable pigpiod --now
\ No newline at end of file
diff --git a/orchestrateur/sensors/__init__.py b/orchestrateur/sensors/__init__.py
index 4f3405b..7bd69b2 100644
--- a/orchestrateur/sensors/__init__.py
+++ b/orchestrateur/sensors/__init__.py
@@ -6,4 +6,5 @@ import sensors.button as button
import sensors.gps as gps
import sensors.camera as camera
import sensors.buzzer as buzzer
-import sensors.rgb_lcd as rgb_lcd
\ No newline at end of file
+import sensors.rgb_lcd as rgb_lcd
+import sensors.rfid_reader as rfid_reader
\ No newline at end of file
diff --git a/orchestrateur/sensors/rfid_reader.py b/orchestrateur/sensors/rfid_reader.py
new file mode 100644
index 0000000..2cc035f
--- /dev/null
+++ b/orchestrateur/sensors/rfid_reader.py
@@ -0,0 +1,95 @@
+import time
+import pigpio
+from typing import Optional
+from sensors.lock import serial_lock
+
+
+class RFIDReader:
+ """
+ Grove 125kHz RFID Reader using bit-banged software serial on GPIO 17 (Pin 11).
+ Parses 14-byte frame: [0x02 STX] + [10 ASCII ID] + [2 ASCII Checksum] + [0x03 ETX]
+ """
+
+ def __init__(self, rx_pin: int = 17, baudrate: int = 9600):
+ self.rx_pin = rx_pin
+ self.baudrate = baudrate
+ self.buffer = bytearray()
+ self.pi = pigpio.pi()
+
+ if not self.pi.connected:
+ raise RuntimeError("pigpio daemon is not running. Run 'sudo systemctl start pigpiod'.")
+
+ with serial_lock:
+ self.pi.set_mode(self.rx_pin, pigpio.INPUT)
+
+ # Clean up lingering serial sessions on this pin
+ try:
+ self.pi.bb_serial_read_close(self.rx_pin)
+ except pigpio.error:
+ pass
+
+ self.pi.bb_serial_read_open(self.rx_pin, self.baudrate, 8)
+
+ def read_tag(self) -> Optional[str]:
+ """
+ Reads and accumulates bytes, returning the 10-digit Tag ID.
+ """
+ with serial_lock:
+ if not self.pi or not self.pi.connected:
+ return None
+
+ count, data = self.pi.bb_serial_read(self.rx_pin)
+ if count > 0:
+ self.buffer.extend(data)
+
+ # Look for Start-of-Text (0x02)
+ stx_idx = self.buffer.find(b"\x02")
+
+ if stx_idx != -1:
+ # Discard noise prior to 0x02
+ if stx_idx > 0:
+ self.buffer = self.buffer[stx_idx:]
+
+ # Wait for full 14-byte payload
+ if len(self.buffer) >= 14:
+ raw_frame = self.buffer[:14]
+ self.buffer = self.buffer[14:] # Flush parsed frame
+
+ # Verify End-of-Text (0x03)
+ if raw_frame[-1] == 0x03:
+ try:
+ # Extract 10-character ID (indices 1 through 10)
+ return raw_frame[1:11].decode("ascii")
+ except UnicodeDecodeError:
+ return None
+ else:
+ # Flush buffer if filled with noise without STX marker
+ if len(self.buffer) > 64:
+ self.buffer.clear()
+
+ return None
+
+ def close(self):
+ with serial_lock:
+ if self.pi and self.pi.connected:
+ try:
+ self.pi.bb_serial_read_close(self.rx_pin)
+ except pigpio.error:
+ pass
+ self.pi.stop()
+
+
+if __name__ == "__main__":
+ reader = RFIDReader(rx_pin=17)
+ print("RFID Reader active on GPIO 17 (Pin 11). Swipe a tag...")
+
+ try:
+ while True:
+ tag = reader.read_tag()
+ if tag:
+ print(f"Scanned Tag ID: {tag}")
+ time.sleep(0.05)
+ except KeyboardInterrupt:
+ print("\nStopping reader.")
+ finally:
+ reader.close()
\ No newline at end of file