在我的项目中，我有一个 USB 监视器正在运行，但我发现它会干扰其他功能，因此我想在必要时启动/停止它。在我的代码中，启动功能正常，但停止功能不行，我无法让它停止。 这是代码：
#lots of import stuff#
# USB Monitoring Process (global variables to manage lifecycle,
# there are many due to the experiments I made trying to solve the issue)
usb_process = None
running = False
usb_monitor_thread = None
usb_monitor_started = False
def device_event_monitor(action, device):
if device.subsystem == 'usb':
if action == 'add' and device.device_type == 'usb_device':
# my code does stuff and get some info from the connected device
# especially gets some values to generate an hash
# to insert in a sql database
device_hash = hash_object.hexdigest()
if device_hash in hash_list:
cursor.execute(
"UPDATE devices SET connection_time = ? WHERE hash = ?",
(start_time, device_hash),
)
else:
# This is the part I can't get working: I want to stop monitor and
# start a device registration process through a webpage, but
# I see no message, the monitor does not stop, the page is not rendered
stop_usb_monitor()
print(f"STOP MONITOR!!")
with app.app_context():
return render_template("new_device.html")
def monitor_usb():
global running
context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by('usb')
monitor.enable_receiving()
observer = pyudev.MonitorObserver(monitor, device_event_monitor)
observer.start()
try:
while running:
time.sleep(0.1)
except KeyboardInterrupt:
print("\nMonitor manually stopped.")
finally:
observer.stop()
print("Monitor USB stopped.")
def start_usb_monitor():
global running, usb_monitor_thread
if running:
print("Already running!")
return
running = True
usb_monitor_thread = threading.Thread(target=monitor_usb, daemon=True)
usb_monitor_thread.start()
print("\nMonitor started!")
def stop_usb_monitor():
global running, usb_monitor_thread
if not running:
print("Monitor not running!")
return
running = False
if usb_monitor_thread:
usb_monitor_thread.join() # Aspetta la terminazione del thread
print("Monitor USB stopped!")
@app.route("/")
def index():
return render_template("index.html")
@app.route('/start', methods=['POST'])
def start():
start_usb_monitor()
return render_template("table.html")
@app.route("/new_device", methods=["POST"])
def new_device():
device_name = request.form.get("owner_name")
if __name__ == "__main__":
# Start Flask app
app.run(debug=True)`
如果我的代码混乱，请抱歉，我是初学者，我尝试了很多变量来启动/停止监视器。我发现的一件事是，我的html页面中的JavaScript在很多方面进行了干扰，因此我将页面保留为纯html并获得了一些更好的结果（主要是为了控制监控过程的启动，一开始我将其设置为自动启动）。
我觉得你想得太复杂了。
pyudev.MonitorObserver以下示例向您展示了通过网站启动和停止监控的一种方法。所有过滤的事件都通过套接字报告给站点并显示。根据要求，您应该调整和修改代码以适应具体情况。
如果你想在事件处理程序中停止观察者，你应该使用
MonitorObserver.send_stop()'''
Install the requirements:
% pip install pyudev flask flask-socketio
'''
from flask import (
Flask,
render_template_string,
)
from flask_socketio import SocketIO
from threading import Lock
import pyudev
class MyObserver:
def __init__(self):
self.context = pyudev.Context()
self.monitor = pyudev.Monitor.from_netlink(self.context)
self.monitor.filter_by('usb')
self.monitor.enable_receiving()
self.observer = None
self.thread_lock = Lock()
def log_event(self, action, device):
data = {
'action': action,
'device_path': device.get('DEVPATH'),
'device_name': device.get('DEVNAME'),
'device_vendor': device.get('ID_VENDOR_FROM_DATABASE'),
}
sio.emit('event', data)
def start(self):
with self.thread_lock:
if self.observer is None:
self.observer = pyudev.MonitorObserver(self.monitor, self.log_event)
self.observer.start()
def stop(self):
with self.thread_lock:
if self.observer is not None:
self.observer.stop()
self.observer = None
app = Flask(__name__)
sio = SocketIO(app)
observer = MyObserver()
@app.route('/')
def index():
return render_template_string('''
<form name="start">
<button type="submit">Start</button>
</form>
<form name="stop">
<button type="submit">Stop</button>
</form>
<div id="device-infos"></div>
<script
src="https://cdn.socket.io/4.8.1/socket.io.min.js"
integrity="sha384-mkQ3/7FUtcGyoppY6bz/PORYoGqOl7/aSUMn2ymDOJcapfS6PHqxhRTMh1RR0Q6+"
crossorigin="anonymous"></script>
<script>
(function() {
const handleEvent = (data) => {
const devInfoEl = document.createElement('div');
devInfoEl.innerHTML = `
<table width="100%" border="1">
<thead>
<tr>
<td>Action</td>
<td>Path</td>
<td>Name</td>
<td>Vendor</td>
</tr>
</thead>
<tbody>
<tr>
<td>${data.action}</td>
<td>${data.device_path}</td>
<td>${data.device_name}</td>
<td>${data.device_vendor}</td>
</tr>
</tbody>
</table>
`;
const devInfosEl = document.getElementById('device-infos');
devInfosEl.append(devInfoEl);
};
const handleStart = (event) => {
event.preventDefault();
fetch('/start', { method: 'post' });
};
const handleStop = (event) => {
event.preventDefault();
fetch('/stop', { method: 'post' });
};
sock = io();
sock.on('event', handleEvent);
document.querySelector('form[name="start"]').addEventListener('submit', handleStart);
document.querySelector('form[name="stop"]').addEventListener('submit', handleStop);
})();
</script>
''')
@app.post('/start')
def start():
observer.start()
return ''
@app.post('/stop')
def stop():
observer.stop()
return ''