[Python] Issue with inconsistent variable state inside button thread loop (Raspberry Pi +...

Discussão em 'Python' iniciado por Stack, Abril 11, 2025.

  1. Stack

    Stack Membro Participativo

    OBS: If more context is needed I will send it and sorry for any kind of English or code error as the question was translated by AI

    I'm running a Flask-based API on a Raspberry Pi that controls a turnstile using a servo motor. It integrates with a facial recognition system, and its main flow is:

    Receives facial recognition data (via external API).

    If access is approved, it opens the turnstile.

    When the person goes through and presses a button, it confirms their passage via another external API using their

    I recently changed some logic from using global variables and utility functions to using a CatracaState class to manage state, but the issue I’m facing still persists.

    Problem:

    The thread responsible for monitoring the button behaves inconsistently. Here's what happens:

    On the first press, all the values (card_id, angle, etc.) come through as None or 0, even though I know they're set earlier.

    On the second press, the values are correct and everything works fine.

    The IS_HOLDING variable (used to debounce button presses) doesn't seem to behave properly — even if it’s set to True, the thread sometimes continues into the logic block.

    This issue appears only in the first iteration. After that, the logic works as expected.

    If I remove time.sleep after the servo action, everything seems to work (values are valid and confirmed), but I need that delay to allow people to pass through the turnstile.

    What I've tried:

    Moving from global variables to class-based state management.

    Making IS_HOLDING global, local, and class-bound — all had similar behavior.

    Checked the flow of data and prints confirm values are set, but the thread sees them as None/0.

    Suspect it's a thread visibility or race condition issue, but unsure how to fix this properly.

    Console Output (trimmed for clarity):

    #-> (this is not a print) Current angle is taken from the env so it should have # # value, card id is set...
    Button pressed: False
    Button state: True

    Current angle: 0
    check_student_pass started with card_id: None
    ➡️ Moving servo to 0°
    ⚠️ No card read. Skipping request.
    Then on the second try:



    #-> (this is not a print) here is the second attempt, even though the button that # should be true remains as false, as if it had not changed
    Current angle: 85
    check_student_pass started with card_id: cpf2
    ✅ Passage confirmed
    ➡️ Moving servo to 65°
    Sending confirmation to server: cpf2
    Additional Context:



    I'm using RPi.GPIO, a custom class (CatracaState) to track card reads and repetitions, and a threading.Thread to run the button monitor.

    I cannot use await for the face recognition process, because another person might show up before the previous one is confirmed — the system needs to stay responsive.

    ❓ Question:

    Why are my variables (card_id, angle, etc.) showing None or 0 on the first loop iteration of the button thread, even though they’re clearly set beforehand? How can I make sure the thread sees the correct updated state right away?

    This error already occurred before the check function, but it worked because it does not send a request with a null card id


    Button monitor

    import RPi.GPIO as GPIO
    import time
    import threading
    import requests
    import os
    from datetime import datetime
    from servo_controller import set_angle, get_current_angle
    from api.global_state import catraca_state

    BUTTON_PIN = 23 # Button GPIO pin
    CONFIRM_PASS_URL = "https://your.api.endpoint/turnstile/pass" # Replace with real endpoint
    TURNSTILE_NAME = os.getenv("TURNSTILE_ID", 1)
    TOKEN = "REPLACE_WITH_REAL_TOKEN"
    headers = {
    "Authorization": f"Bearer {TOKEN}"
    }
    DEBUG = True

    # Setup GPIO pin for the button
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP) # Using PULL-UP configuration

    def _monitor_button():
    """Infinite loop that checks if the button is pressed and sends a confirmation to the server."""
    is_holding = False

    try:
    while True:
    if GPIO.input(BUTTON_PIN) != GPIO.HIGH:
    print(" Button pressed. is_holding:", is_holding)
    if is_holding:
    time.sleep(0.01)
    continue
    is_holding = True

    print("Proceeding to check angle")
    current_angle = get_current_angle()
    print("Current angle:", current_angle)
    [ANGLE_OPEN, ANGLE_CLOSE] = catraca_state.get_catraca_angles()

    if catraca_state.get_always_open():
    print(" 'Always open' mode is ON. Skipping motor and API.")
    time.sleep(0.01)
    continue

    last_card_id = catraca_state.get_last_card_id()

    card_id = check_student_pass(last_card_id)

    # Give time for the person to pass
    time.sleep(1)
    print(f"➡️ Moving servo to {ANGLE_CLOSE}°")
    set_angle(ANGLE_CLOSE)

    if card_id is not None:
    last_card_id = card_id

    if last_card_id is None:
    print("⚠️ No card ID detected. Request will not be sent.")
    else:
    try:
    print(f" Sending confirmation to server: {last_card_id}")
    response = requests.put(
    CONFIRM_PASS_URL,
    json={
    "CPF": last_card_id,
    "turnstileName": TURNSTILE_NAME
    },
    headers=headers,
    timeout=5
    )
    print(f" Server response: {response.status_code} - {response.text}")

    if response.status_code == 200:
    print("✅ Passage successfully confirmed!")
    else:
    print("⚠️ Failed to confirm passage.")

    catraca_state.reset_last_card_id()

    except requests.exceptions.RequestException as e:
    print(f"❌ Error communicating with server: {e}")
    else:
    is_holding = False

    time.sleep(0.01)
    except KeyboardInterrupt:
    GPIO.cleanup()
    print(" Button monitoring stopped.")

    def start_button_monitor():
    """Starts the thread to monitor the button press."""
    GPIO.setmode(GPIO.BCM)
    GPIO.setup(BUTTON_PIN, GPIO.IN, pull_up_down=GPIO.PUD_UP)
    t = threading.Thread(target=_monitor_button, daemon=True)
    t.start()

    def check_student_pass(last_card_id):
    """
    Logic to verify if the same student has passed twice in a row.
    If true, returns the card_id of the student who should have passed instead.

    This prevents someone from passing with another person's facial recognition.
    """
    print(f" check_student_pass started with card_id: {last_card_id}")

    if last_card_id is None:
    return None

    catraca_state.update_student_pass(last_card_id, True)
    print(f"✅ Updated pass count for card_id: {last_card_id}")

    students_pass_list = catraca_state.get_students_pass()
    print(f" Total students in memory: {len(students_pass_list)}")

    flag_double_pass = False
    double_pass_student = None

    student_passing = next((s for s in students_pass_list if s['card_id'] == last_card_id), None)

    if student_passing is None:
    print(f"⚠️ No student found with card_id: {last_card_id}")
    return None

    print(f" Student found: {student_passing['card_id']}, group: {student_passing['group_id']}, count: {student_passing['pass_count']}")

    is_not_whitelisted = student_passing['group_id'] != 2
    print(f" Not in whitelist? {is_not_whitelisted}")

    for student in students_pass_list:
    if is_not_whitelisted and student['pass_count'] == 0 and student['group_id'] == 2:
    print(f" Case 1: Whitelisted student who hasn't passed yet found: {student['card_id']}")
    return student['card_id']

    if student_passing['pass_count'] >= 2:
    flag_double_pass = True
    double_pass_student = student_passing
    print(f"⚠️ Student {student_passing['card_id']} passed more than once: {student_passing['pass_count']}")

    if flag_double_pass:
    t1 = double_pass_student['identification_time']
    print(f" Identification time of double-pass student: {t1}")

    for student in students_pass_list:
    if student['card_id'] == double_pass_student['card_id']:
    print(f"⏭️ Skipping self comparison for: {student['card_id']}")
    continue

    t2 = student['identification_time']
    print(f" Comparing with: {student['card_id']}, group: {student['group_id']}, count: {student['pass_count']}, time: {t2}")

    if student['pass_count'] == 0 and t1 > t2 and student['group_id'] == 2:
    print(f"✅ Case 2: Whitelisted student who was identified earlier and hasn't passed: {student['card_id']}")
    student['pass_count'] += 1
    catraca_state.update_student_pass(student['card_id'], True)
    double_pass_student['pass_count'] -= 1
    catraca_state.update_student_pass(double_pass_student['card_id'], False)
    print(f" Adjusted counts: {student['card_id']}={student['pass_count']}, {double_pass_student['card_id']}={double_pass_student['pass_count']}")
    return student['card_id']

    print(f"❌ No special case detected. Returning None")
    return None

    Continue reading...

Compartilhe esta Página