flt/ai-titration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source 
Former-commit-id: 1b34dc488f5306fcb5d20746c217255bb9ead3c5
This commit is contained in:
flt6
2025-06-06 23:02:15 +08:00
parent 8184179a17
commit c1f33df66d
7 changed files with 112 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
main.py
View File

@ -6,6 +6,7 @@ import ch340
import atexit
import utils
from utils import State, History
from scipy.signal import find_peaks
class MAT:
@ -39,8 +40,8 @@ class MAT:
self.control_logger.info('完成抽取')
def ch340_init(self):
self.ch340.push(speed=1,t=1)
self.ch340.pull(speed=1.2,vol=1)
# self.ch340.push(speed=1,t=1)
self.ch340.pull(speed=1.2,vol=1.8)
self.control_logger.info('电极位置已初始化')
def ch340_push(self, speed=0.1):
@ -93,12 +94,17 @@ class MAT:
if self.state.mode == State.Mode.CRAZY:
self._display_status(im, ret, rate, val)
return ret
# === 状态进入逻辑 ===
if now - self._start_time<10:
self._display_status(im, ret, rate, val)
return ret
if not self.edited and self.history.base is not None and rate>self.history.base * 2:
self.edited = True
self.speeds[0] /= 2
# 1. middle: predictor返回middle立即进入slow状态
if ret == "middle":
if self.state.is_fast_mode():
@ -141,8 +147,8 @@ class MAT:
if self.state.mode == State.Mode.ABOUT and self.state.about_check:
h = self.history.get_recent_records(self.about_time/3,now)
ratio = self.history.get_state_ratio("about", h)
self.history.about_history.append(ratio<0.3)
ratio = self.history.get_state_ratio("transport", h)
self.history.about_history.append(ratio == 1)
while len(self.history.about_history) > 5:
self.history.about_history.pop(0)
if len(self.history.about_history) == 5:
@ -150,51 +156,96 @@ class MAT:
if rate > 0.8:
self.state.exit_about()
self.control_logger.info(f"about比例{ratio:.2%}<80%，退出about检查，返回middle模式")
self.end_check()
# print(f"end_history: {self.history.end_history}",peaks)
# if len(self.history.end_history) >= 10:
# colored_ratio=sum(self.history.end_history)/len(self.history.end_history)
# if colored_ratio > 0.7:
# self.endpoint_logger.info(f"colored比例{colored_ratio:.2%}>=70%，确认滴定终点")
# self.endpoint_logger.info(f"最终体积: {self.colored_volume:.2f} ml")
# self.running = False
# self.ch340.stop()
# if self.colored_im is not None:
# cv2.imwrite(f"colored_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg", self.colored_im)
# return "colored"
# while len(self.history.end_history) > 10:
# self.history.end_history.pop(0)
self.state.about_check = False
# end检查: 进入end之后的end_bounce_time，如果end比例<80%，则重置；否则终止实验
if self.state.should_check_end_result(now):
colored_ratio = self.history.get_state_ratio("colored", self.history.get_recent_records(self.state.end_bounce_time, now))
# if self.state.should_check_end_result(now):
# colored_ratio = self.history.get_state_ratio("colored", self.history.get_recent_records(self.state.end_bounce_time, now))
if colored_ratio < 0.8:
# end比例<80%，从history中找到第二个end并继续check逻辑
self.endpoint_logger.warning(f"colored比例{colored_ratio:.2%}<80%，寻找下一个colored点")
# if colored_ratio < 0.8:
# # end比例<80%，从history中找到第二个end并继续check逻辑
# self.endpoint_logger.warning(f"colored比例{colored_ratio:.2%}<80%，寻找下一个colored点")
# 寻找历史中倒数第二个colored状态
colored_times = self.history.get_states_by_type("colored")
if len(colored_times) >= 2:
# 使用倒数第二个colored时间重新开始检查
second_last_colored_time = colored_times[1]
self.state.end_detected_time = second_last_colored_time
# # 寻找历史中倒数第二个colored状态
# colored_times = self.history.get_states_by_type("colored")
# if len(colored_times) >= 2:
# # 使用倒数第二个colored时间重新开始检查
# second_last_colored_time = colored_times[1]
# self.state.end_detected_time = second_last_colored_time
# 更新colored记录为对应的体积
record = self.history.find_record_by_timestamp(second_last_colored_time)
if record:
self.colored_volume = record.volume
self.colored_time = record.timestamp
self.colored_im = record.image.copy()
# # 更新colored记录为对应的体积
# record = self.history.find_record_by_timestamp(second_last_colored_time)
# if record:
# self.colored_volume = record.volume
# self.colored_time = record.timestamp
# self.colored_im = record.image.copy()
self.endpoint_logger.info(f"重置到第二个colored点: {self.colored_volume:.2f} ml")
else:
# 没有足够的colored点，重置end检查
self.state.reset_end_check()
self.colored_volume = None
self.colored_time = None
self.endpoint_logger.info("没有足够的colored点，重置end检查")
else: # end比例>=80%，确认终点，终止实验
self.endpoint_logger.info(f"colored比例{colored_ratio:.2%}>=80%，确认滴定终点")
self.endpoint_logger.info(f"最终体积: {self.colored_volume:.2f} ml")
self.running = False
self.ch340.stop()
if self.colored_im is not None:
cv2.imwrite(f"colored_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg", self.colored_im)
return "colored"
# self.endpoint_logger.info(f"重置到第二个colored点: {self.colored_volume:.2f} ml")
# else:
# # 没有足够的colored点，重置end检查
# self.state.reset_end_check()
# self.colored_volume = None
# self.colored_time = None
# self.endpoint_logger.info("没有足够的colored点，重置end检查")
# else: # end比例>=80%，确认终点，终止实验
# self.endpoint_logger.info(f"colored比例{colored_ratio:.2%}>=80%，确认滴定终点")
# self.endpoint_logger.info(f"最终体积: {self.colored_volume:.2f} ml")
# self.running = False
# self.ch340.stop()
# if self.colored_im is not None:
# cv2.imwrite(f"colored_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg", self.colored_im)
# return "colored"
# 显示状态信息
self._display_status(im, ret, rate, val)
return ret
def end_check(self,dep=0):
if dep > 5:
# self.endpoint_logger.info(f"colored比例{.2%}>=70%，确认滴定终点")
self.endpoint_logger.info(f"最终体积: {self.colored_volume:.2f} ml")
self.running = False
self.ch340.stop()
if self.colored_im is not None:
cv2.imwrite(f"colored_{datetime.now().strftime('%Y%m%d_%H%M%S')}.jpg", self.colored_im)
return "colored"
suc,im = self.cap.read()
hsv = cv2.cvtColor(im, cv2.COLOR_BGR2HSV)
s = hsv[:, :, 0]
s = s[s > 0]
hist = cv2.calcHist([s], [0], None, [256], [0, 256])
hist = hist.flatten() # 转换为一维数组
# 峰值检测 - 找到直方图中的峰值
peaks, properties = find_peaks(hist,
height=np.max(hist) * 0.2, # 峰值高度至少是最大值的10%
distance=10, # 峰值之间的最小距离
prominence=np.max(hist) * 0.01) # 峰值的突出度
if np.any(peaks>130):
self.history.end_history.append(True)
time.sleep(2)
self.control_logger.info(f"检测到colored状态，end_check at {dep}")
self.end_check(dep+1)
else:
self.history.end_history.append(False)
def _display_status(self, im, detection_result, rate, volume):
"""显示状态信息到图像上"""
mode_color = {
@ -223,7 +274,7 @@ class MAT:
rate = val/tot
if self.history.base is not None:
base = self.history.base
thr = (min(0.05,base*5), min(0.4,base*9), 0.75)
thr = (min(0.05,base*5), min(0.4,base*13), 0.5)
if rate < thr[0]:
return "transport",rate
elif rate <thr[1]:
@ -246,8 +297,9 @@ class MAT:
def run(self, quick_speed=0.2, slow_speed=0.05, end_speed=0.02, mid_time=0.5, about_time=1, cap_dir="Videos"):
self.running = True
self.start_time = time.time()
self.speeds = (quick_speed, slow_speed, end_speed, 1.0)
self.speeds = [quick_speed, slow_speed, end_speed, 1.0]
self.need_check = False
self.edited = False
if cap_dir is not None:
vid_name = f"{cap_dir}/{datetime.now().strftime('%Y%m%d_%H%M%S')}.mkv"
@ -363,7 +415,7 @@ if __name__ == "__main__":
mat.run(
slow_speed = 0.05,
quick_speed = 0.15,
quick_speed = 0.45,
about_time=3,
# cap_dir=None
)