93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
'''
|
||
# @Time : 2025/12/12 11:05
|
||
# @Author : reenrr
|
||
# @File : main_control.py
|
||
# @Desc : 主控程序
|
||
'''
|
||
import multiprocessing # 多进程模块
|
||
import threading
|
||
from threading import Event
|
||
import time
|
||
from EMV.EMV_test import press_sensors_triggered, control_solenoid, global_relay
|
||
from visual_algorithm.visual_algorithm import flaw_detection
|
||
import logging
|
||
import os
|
||
|
||
# ------------ 日志+参数配置 ------------
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
log_file_path = os.path.join(script_dir, "main_control.log")
|
||
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format='[%(asctime)s.%(msecs)03d] [%(levelname)s] %(message)s',
|
||
datefmt='%Y-%m-%d %H:%M:%S',
|
||
handlers=[
|
||
logging.StreamHandler(),
|
||
logging.FileHandler(log_file_path, encoding='utf-8')
|
||
]
|
||
)
|
||
|
||
# ------------全局事件-------------
|
||
manger = multiprocessing.Manager()
|
||
conveyor_start_event = manger.Event()
|
||
|
||
def start_thread():
|
||
"""开启各种线程"""
|
||
global_relay.start_press_sensors_monitor() # 双压传感器监听线程
|
||
|
||
def stop_thread():
|
||
"""关闭各种线程"""
|
||
global_relay.stop_press_sensors_monitor() # 双压传感器监听线程
|
||
|
||
|
||
def quality_testing():
|
||
logging.info("线条开始质量检测：")
|
||
|
||
# 执行质量检测
|
||
result = flaw_detection({"line_id": "L001", "straightness": 0.95, "noise_ratio": 0.08})
|
||
if result == "qualified":
|
||
result = "合格"
|
||
logging.info("该线条是否合格：", result)
|
||
logging.info("等待线条落到传送带（双压传感器触发）...")
|
||
# 等待时间触发，超时时间设为10秒（避免无限等待）
|
||
if press_sensors_triggered.wait(timeout=10):
|
||
logging.info("线条已落到传送带，控制两个传送带向前移动")
|
||
# 触发传送带启动事件
|
||
conveyor_start_event.set()
|
||
else:
|
||
logging.info("超时警告：线条未落到传送带，请检查")
|
||
elif result == "unqualified":
|
||
result = "不合格"
|
||
logging.info("该线条是否合格：", result)
|
||
logging.info("等待线条落到传送带上")
|
||
# 等待时间触发，超时时间设为10秒（避免无限等待）
|
||
if press_sensors_triggered.wait(timeout=10):
|
||
logging.info("线条已落到传送带")
|
||
logging.info("进入NG动作")
|
||
control_solenoid() # 执行NG动作,控制电磁阀
|
||
logging.info("NG动作结束")
|
||
# logging.info("判断NG线条是否落入肥料区：")
|
||
|
||
# -----------对外接口-------------
|
||
def main_control():
|
||
logging.info("开启各种线程")
|
||
start_thread()
|
||
|
||
logging.info("开始摆放线条")
|
||
|
||
# 质量检测
|
||
quality_testing()
|
||
|
||
while True: # 防止跳出循环
|
||
time.sleep(1)
|
||
|
||
# ------------测试接口-------------
|
||
if __name__ == '__main__':
|
||
main_control()
|
||
|
||
|
||
|
||
|