如何在Flask外部重复运行脚本:Raspberry Pi距离检测自动化实现
Got it, let's break this down. You need a way to start a recurring distance measurement task (once per minute) when clicking a button on your Flask page, and make sure that task keeps running even if you navigate away or do other things on the site. The key here is to use a background task scheduler—because Flask's default request handling is synchronous, so you can't just loop and sleep in a request (it'll block the server and die when the request ends).
I'll use APScheduler for this—it's lightweight, easy to set up, and works great on Raspberry Pi. Here's the step-by-step solution:
1. Install Required Packages
First, grab the dependencies we need:
pip install flask apscheduler
2. Integrate Your Sensor Script with Flask
Let's assume your existing sensor measurement code looks something like this (swap this out with your actual sensor logic, e.g., HC-SR04):
# sensor.py import RPi.GPIO as GPIO import time def measure_distance(): # Example HC-SR04 measurement logic GPIO.setmode(GPIO.BCM) TRIG_PIN = 23 ECHO_PIN = 24 GPIO.setup(TRIG_PIN, GPIO.OUT) GPIO.setup(ECHO_PIN, GPIO.IN) # Trigger sensor GPIO.output(TRIG_PIN, False) time.sleep(0.2) GPIO.output(TRIG_PIN, True) time.sleep(0.00001) GPIO.output(TRIG_PIN, False) # Calculate distance while GPIO.input(ECHO_PIN) == 0: pulse_start = time.time() while GPIO.input(ECHO_PIN) == 1: pulse_end = time.time() pulse_duration = pulse_end - pulse_start distance = round(pulse_duration * 17150, 2) # Convert to cm GPIO.cleanup() return distance
Now build the Flask app with the scheduler. We'll use a global variable to store the latest distance reading, and manage the scheduler instance so we can start/stop it:
# app.py from flask import Flask, render_template, jsonify from apscheduler.schedulers.background import BackgroundScheduler import sensor import atexit app = Flask(__name__) latest_distance = None # Holds our most recent measurement scheduler = None # Track the scheduler instance def update_distance_reading(): """Run the sensor measurement and update the latest value""" global latest_distance try: latest_distance = sensor.measure_distance() print(f"Updated distance: {latest_distance} cm") except Exception as e: print(f"Measurement failed: {str(e)}") @app.route('/') def home(): return render_template('index.html') @app.route('/start-measurement') def start_measurement_task(): """API endpoint to start the 1-minute interval measurement""" global scheduler if scheduler and scheduler.running: return jsonify({"status": "already_running", "message": "Task is already active!"}) # Initialize background scheduler scheduler = BackgroundScheduler() # Add task: run every 60 seconds scheduler.add_job(update_distance_reading, 'interval', minutes=1) scheduler.start() # Make sure we shut down the scheduler cleanly when the app exits atexit.register(lambda: scheduler.shutdown()) # Run once immediately so we don't wait a minute for the first reading update_distance_reading() return jsonify({"status": "started", "initial_distance": latest_distance}) @app.route('/get-latest-distance') def get_latest_distance(): """API endpoint to fetch the most recent measurement""" return jsonify({"distance": latest_distance}) @app.route('/stop-measurement') def stop_measurement_task(): """API endpoint to stop the recurring task""" global scheduler if scheduler and scheduler.running: scheduler.shutdown() return jsonify({"status": "stopped", "message": "Task has been halted."}) return jsonify({"status": "not_running", "message": "No active task to stop."}) if __name__ == '__main__': # Important: Disable debug mode! It restarts the app and creates duplicate schedulers app.run(host='0.0.0.0', port=5000, debug=False)
3. Build the Frontend Web Page
Create a templates folder in your project root, then add index.html—this is the page with the start/stop buttons and live distance display:
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <title>Raspberry Pi Distance Monitor</title> <style> .container { max-width: 500px; margin: 3rem auto; text-align: center; font-family: Arial, sans-serif; } .distance-display { font-size: 2rem; margin: 2rem 0; padding: 1rem; border: 2px solid #333; border-radius: 8px; } .btn { padding: 0.8rem 1.5rem; margin: 0 0.5rem; font-size: 1rem; border: none; border-radius: 4px; cursor: pointer; } .start-btn { background-color: #4CAF50; color: white; } .stop-btn { background-color: #f44336; color: white; } </style> </head> <body> <div class="container"> <h1>Distance Sensor Monitor</h1> <div class="distance-display"> Current Distance: <span id="distance-value">--</span> cm </div> <button class="btn start-btn" onclick="startMeasurement()">Start 1-Minute Measurements</button> <button class="btn stop-btn" onclick="stopMeasurement()">Stop Measurements</button> </div> <script> let refreshTimer; function startMeasurement() { fetch('/start-measurement') .then(res => res.json()) .then(data => { if (data.status === 'started') { document.getElementById('distance-value').textContent = data.initial_distance; // Refresh the display every 5 seconds to get the latest reading refreshTimer = setInterval(() => { fetch('/get-latest-distance') .then(res => res.json()) .then(data => { document.getElementById('distance-value').textContent = data.distance || '--'; }); }, 5000); alert('Measurement task started! Check the console for logs.'); } else { alert(data.message); } }); } function stopMeasurement() { fetch('/stop-measurement') .then(res => res.json()) .then(data => { if (data.status === 'stopped') { clearInterval(refreshTimer); alert(data.message); } else { alert(data.message); } }); } </script> </body> </html>
Key Notes to Remember
- Background Task Independence: The
BackgroundSchedulerruns in a separate thread, so it won't be affected by web requests, page reloads, or navigating away from the site. It'll keep running until you call the stop endpoint or restart the Flask app. - Debug Mode Warning: Never run this with
debug=True—Flask's debug mode restarts the app automatically, which creates duplicate scheduler instances and can cause GPIO conflicts. - GPIO Cleanup: Always make sure your sensor function cleans up GPIO pins (like
GPIO.cleanup()in the example) to avoid pin locking issues on Raspberry Pi. - Frontend Refresh: The frontend fetches the latest reading every 5 seconds—this is more frequent than the 1-minute measurement interval, so you'll always see the most up-to-date value as soon as it's measured.
How to Test
- Organize your files like this:
your-project/ ├── app.py ├── sensor.py └── templates/ └── index.html - Run the app:
python app.py - Open your browser and go to
http://[your-rpi-ip]:5000(find your Pi's IP withhostname -I). - Click "Start 1-Minute Measurements"—you'll see the first reading immediately, and it'll update every minute. Even if you close the tab, the task keeps running in the background until you click stop or restart the app.
内容的提问来源于stack exchange,提问作者Mike Meyers

