capture
This commit is contained in:
@@ -80,47 +80,6 @@ class VideoCaptureInterface(abc.ABC):
|
|||||||
)
|
)
|
||||||
self._last_detection = None # résultat du dernier alignement
|
self._last_detection = None # résultat du dernier alignement
|
||||||
|
|
||||||
# calibrage ou lecture réelle
|
|
||||||
#
|
|
||||||
|
|
||||||
def align_on_well_arrival(self, frame: bytes, cnc_controller, tube_diameter: float = 16.0) -> dict:
|
|
||||||
"""
|
|
||||||
Appelé UNE FOIS à l'arrivée sur un nouveau puits.
|
|
||||||
Détecte le tube, décide l'action, exécute la correction.
|
|
||||||
|
|
||||||
:param frame: Frame JPEG bytes capturée après déplacement CNC
|
|
||||||
:param grbl_send_func: Callable(gcode: str) → envoie le G-code au GRBL
|
|
||||||
:return: dict résultat de la détection
|
|
||||||
"""
|
|
||||||
nparr = np.frombuffer(frame, np.uint8)
|
|
||||||
img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
|
|
||||||
detection = self._aligner.detect_tube(img, tube_diameter=tube_diameter)
|
|
||||||
|
|
||||||
# Stockage pour process_frame
|
|
||||||
self._last_detection = detection
|
|
||||||
|
|
||||||
if not detection["detected"]:
|
|
||||||
logger.warning("align_on_well_arrival: tube non détecté")
|
|
||||||
return detection
|
|
||||||
|
|
||||||
action = detection["action"]
|
|
||||||
if action == "grbl":
|
|
||||||
dx_mm = detection["offset_x_mm"]
|
|
||||||
dy_mm = detection["offset_y_mm"]
|
|
||||||
|
|
||||||
msg = f"align_on_well_arrival: correction CNC move_relative(dx={dx_mm:.3f}, dy={dy_mm:.3f})"
|
|
||||||
cnc_controller.move_relative(dx=dx_mm, dy=dy_mm, feed=150)
|
|
||||||
|
|
||||||
self._tracker.reset()
|
|
||||||
self._last_detection["action"] = "none"
|
|
||||||
|
|
||||||
elif action == "crop":
|
|
||||||
msg = f"align_on_well_arrival: recadrage logiciel ({detection['offset_x_px']:.1f}px, {detection['offset_y_px']:.1f}px)"
|
|
||||||
|
|
||||||
logger.info(msg)
|
|
||||||
self.display(state='detect_tube', msg=msg)
|
|
||||||
return detection
|
|
||||||
|
|
||||||
|
|
||||||
def on_well_change(self):
|
def on_well_change(self):
|
||||||
"""
|
"""
|
||||||
@@ -281,9 +240,6 @@ class VideoCaptureInterface(abc.ABC):
|
|||||||
annotated = self._last_detection.get('frame_annotated')
|
annotated = self._last_detection.get('frame_annotated')
|
||||||
frame = annotated if annotated is not None else frame
|
frame = annotated if annotated is not None else frame
|
||||||
|
|
||||||
#if (self._last_detection.get("action") == "crop" and self._last_detection.get("detected")):
|
|
||||||
# frame = self._aligner.crop_to_tube(frame, self._last_detection)
|
|
||||||
|
|
||||||
# mode racking
|
# mode racking
|
||||||
if self.use_tracking:
|
if self.use_tracking:
|
||||||
ts = datetime.now(timezone.utc).timestamp()
|
ts = datetime.now(timezone.utc).timestamp()
|
||||||
|
|||||||
@@ -7,8 +7,6 @@ Created on 20 avr. 2026
|
|||||||
|
|
||||||
@author: denis
|
@author: denis
|
||||||
'''
|
'''
|
||||||
import cv2
|
|
||||||
import numpy as np
|
|
||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from threading import Thread, Event
|
from threading import Thread, Event
|
||||||
@@ -17,6 +15,8 @@ from django.utils import timezone
|
|||||||
from django.utils.html import mark_safe
|
from django.utils.html import mark_safe
|
||||||
from . import models
|
from . import models
|
||||||
|
|
||||||
|
CALIBRATION_AUTO_DURATION = 45.0
|
||||||
|
CALIBRATION_AUTO_TIMEOUT = 3.0
|
||||||
|
|
||||||
logging.basicConfig(level=logging.INFO)
|
logging.basicConfig(level=logging.INFO)
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -229,7 +229,7 @@ class MultiWellManager:
|
|||||||
self.stop_playing = Event()
|
self.stop_playing = Event()
|
||||||
self.process.data.tube_diameter = self.multiwell.diameter
|
self.process.data.tube_diameter = self.multiwell.diameter
|
||||||
cam = self.process.cam
|
cam = self.process.cam
|
||||||
duration = self.duration if not auto else 45.0
|
duration = self.duration if not auto else CALIBRATION_AUTO_DURATION
|
||||||
start_test = time.monotonic()
|
start_test = time.monotonic()
|
||||||
|
|
||||||
for w in self.well_iterator:
|
for w in self.well_iterator:
|
||||||
@@ -245,14 +245,14 @@ class MultiWellManager:
|
|||||||
break
|
break
|
||||||
if auto and cam._last_detection:
|
if auto and cam._last_detection:
|
||||||
if cam._last_detection.get('action')=="grbl":
|
if cam._last_detection.get('action')=="grbl":
|
||||||
self.cnc_controller.wait_for(5.0)
|
self.cnc_controller.wait_for(CALIBRATION_AUTO_TIMEOUT)
|
||||||
dx_mm = cam._last_detection["offset_x_mm"]
|
dx_mm = cam._last_detection["offset_x_mm"]
|
||||||
dy_mm = cam._last_detection["offset_y_mm"]
|
dy_mm = cam._last_detection["offset_y_mm"]
|
||||||
|
|
||||||
self.cnc_controller.move_to(self.cnc_controller.x + dx_mm, self.cnc_controller.y + dy_mm, feed=150)
|
self.cnc_controller.move_to(self.cnc_controller.x + dx_mm, self.cnc_controller.y + dy_mm, feed=150)
|
||||||
msg = f"Correction CNC move_relative(dx={dx_mm:.3f}, dy={dy_mm:.3f})"
|
msg = f"Correction CNC move_relative(dx={dx_mm:.3f}, dy={dy_mm:.3f})"
|
||||||
self.process._send(state='center', msg=msg)
|
self.process._send(state='center', msg=msg)
|
||||||
elif cam._last_detection.get('action') in ['none', 'crop']:
|
elif cam._last_detection.get('action') in ['none',]:
|
||||||
msg = f"ok {w.multiwell.position}-{w.well.name}: ({self.cnc_controller.x}, {self.cnc_controller.y})"
|
msg = f"ok {w.multiwell.position}-{w.well.name}: ({self.cnc_controller.x}, {self.cnc_controller.y})"
|
||||||
logger.info(msg)
|
logger.info(msg)
|
||||||
self.process._send(state='save', msg=msg)
|
self.process._send(state='save', msg=msg)
|
||||||
|
|||||||
Reference in New Issue
Block a user