diff --git a/README.md b/README.md
index d20b200..019a254 100644
--- a/README.md
+++ b/README.md
@@ -173,6 +173,21 @@ PlanarianScanner/
---
+## Procédure de calibration en 4 étapes
+1. Activer "Debug détection" → voir le cercle et les zones sur le stream
+
+2. Positionner la CNC manuellement sur un point stable
+ → cliquer "Calib — Point A"
+ → mpos_A et centre tube A enregistrés
+
+3. Déplacer la CNC manuellement d'une distance connue (ex: 10mm en X)
+ → attendre stabilisation (la pause 2s est déjà là)
+ → cliquer "Calib — Point B"
+
+4. Résultat affiché :
+ "Calibration OK — 38.2000 px/mm (0.026178 mm/px) Δ=10.000mm / 382.0px"
+ → px_per_mm sauvegardé dans TubeAligner et persisté en base
+
## Contexte scientifique
Les **planaires** sont des vers plats dotés de remarquables capacités de
diff --git a/test_tube_scanner/home/settings.py b/test_tube_scanner/home/settings.py
index 9dc5041..b89a3b5 100644
--- a/test_tube_scanner/home/settings.py
+++ b/test_tube_scanner/home/settings.py
@@ -387,3 +387,10 @@ EXPORTS_LOCAL_PATH = config("EXPORTS_LOCAL_PATH")
EXPORT_REMOTE_PATH = config("EXPORT_REMOTE_PATH")
EXPORT_DESTINATIONS = ["local", "remote"]
+
+TEST_VIDEOFILE = False
+
+TRACKING = False
+TRACKER_TUBE_AXIS = "horizontal" #"vertical"
+TRACKER_MIN_AREA = 200
+
diff --git a/test_tube_scanner/modules/capture_interface.py b/test_tube_scanner/modules/capture_interface.py
index 0040459..2cd0ef5 100644
--- a/test_tube_scanner/modules/capture_interface.py
+++ b/test_tube_scanner/modules/capture_interface.py
@@ -22,7 +22,9 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Callable, TYPE_CHECKING
-from modules.planarian_tracker import PlanarianTracker
+from django.conf import settings
+from modules.planarian_tracker import PlanarianTracker
+from modules.tube_aligner import TubeAligner
if TYPE_CHECKING:
from .circular_crop import CircularCrop # Evite l'import circulaire au runtime
@@ -47,13 +49,15 @@ class VideoCaptureInterface(abc.ABC):
# Cadence par défaut en images par seconde
DEFAULT_FPS: float = 5.0
- def __init__(self, fps: float = DEFAULT_FPS):
+ def __init__(self, fps: float = DEFAULT_FPS, use_tracking: bool = False, px_per_mm: float = 2.15, display=None):
"""
Initialise l'interface de capture.
:param fps: Cadence cible en images par seconde
"""
self._fps: float = fps
+ self.use_tracking = use_tracking
+ self.display = display
self._interval: float = 1.0 / fps # Intervalle en secondes entre chaque capture
self._running: bool = False # Indique si la capture est en cours
self._thread: Optional[threading.Thread] = None
@@ -61,13 +65,61 @@ class VideoCaptureInterface(abc.ABC):
self._on_frame: Optional[Callable[[bytes, datetime], None]] = None # Callback image
self._circular_crop: Optional["CircularCrop"] = None # Recadrage circulaire optionnel
self._active_median = False
+ self._active_crop = False
self._error_occured = False
self._tracker = PlanarianTracker(
- tube_axis = "vertical", # à rendre configurable via settings
- min_area_px = 20,
+ tube_axis = settings.TRACKER_TUBE_AXIS,
+ min_area_px = settings.TRACKER_MIN_AREA,
)
+ self._aligner = TubeAligner(
+ px_per_mm = px_per_mm, # à calibrer selon la caméra
+ grbl_threshold_px = 20, # au-delà → correction GRBL
+ dead_zone_px = 5, # en-dessous → rien à faire
+ display = display,
+ )
+ self._last_detection = None # résultat du dernier alignement
+ # calibrage ou lecture réelle
+ #
+ def align_on_well_arrival(self, frame: bytes, cnc_controller) -> 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)
+
+ # 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):
"""
@@ -194,12 +246,15 @@ class VideoCaptureInterface(abc.ABC):
"""
self._circular_crop = crop
if crop is not None:
- logger.info(
- "%s : recadrage circulaire activé (R=%d, stratégie=%s)",
- self.__class__.__name__, crop.radius, crop.strategy.name,
- )
+ self._active_crop = True
+ msg = f"{self.__class__.__name__}: recadrage circulaire activé (R={crop.radius}, stratégie={crop.strategy.name})"
else:
- logger.info("%s : recadrage circulaire désactivé", self.__class__.__name__)
+ self._active_crop = False
+ msg= f"{self.__class__.__name__}: recadrage circulaire désactivé"
+
+ logger.info(msg)
+ self.display(state='circular_crop', msg=msg)
+
def process_frame(self, jpeg_bytes: bytes) -> bytes:
"""
@@ -211,26 +266,39 @@ class VideoCaptureInterface(abc.ABC):
:param jpeg_bytes: Image JPEG brute issue du capteur
:return: Image traitée (JPEG ou PNG selon la stratégie)
"""
+ metrics = {"detected": False}
if self._circular_crop is not None:
- jpeg = self._circular_crop.process(jpeg_bytes)
+ jpeg = self._circular_crop.process(jpeg_bytes)
+ nparr = np.frombuffer(jpeg, np.uint8)
+ frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+ if frame is None:
+ return jpeg, metrics
- # --- tracking ---
- nparr = np.frombuffer(jpeg, np.uint8)
- frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
- ts = datetime.now(timezone.utc).timestamp()
- #metrics = self._tracker.process(frame, ts) if frame is not None else {}
- if frame is not None:
- frame_annotated, metrics = self._tracker.process(frame, ts)
- # Ré-encodage JPEG de la frame annotée
- ok, buf = cv2.imencode(".jpg", frame_annotated, [cv2.IMWRITE_JPEG_QUALITY, 85])
- if ok:
- jpeg = buf.tobytes()
+ # Mode debug
+ if self._aligner.debug:
+ self._last_detection = detection = self._aligner.detect_tube(frame)
+ annotated = detection.get('frame_annotated')
+ frame = annotated if annotated is not None else frame
+ '''
else:
- metrics = {"detected": False}
-
+ detection = self._last_detection or {}
+
+ # --- Crop logiciel si nécessaire ---
+ if (detection.get("action") == "crop" and detection.get("detected") and not self._aligner.debug ):
+ frame = self._aligner.crop_to_tube(frame, detection)
+ '''
+
+ if self.use_tracking:
+ ts = datetime.now(timezone.utc).timestamp()
+ frame, metrics = self._tracker.process(frame, ts)
+
+ ok, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 85])
+ if ok:
+ jpeg = buf.tobytes()
+
return jpeg, metrics
- return jpeg_bytes, {"detected": False}
+ return jpeg_bytes, metrics
def save_frame(self, jpeg_bytes: bytes, directory: str = ".", prefix: str = "frame") -> Path:
"""
@@ -264,13 +332,8 @@ class VideoCaptureInterface(abc.ABC):
# ------------------------------------------------------------------
# tracer médianes
- # ------------------------------------------------------------------
- def set_median(self, is_median=False):
- """
- Active ou désactive les médianes
- """
- self._active_median = is_median
-
+ # ------------------------------------------------------------------
+
def display_median(self, jpeg):
if self._active_median:
nparr = np.frombuffer(jpeg, np.uint8)
diff --git a/test_tube_scanner/modules/grbl.py b/test_tube_scanner/modules/grbl.py
index 1c8e440..dc8465f 100644
--- a/test_tube_scanner/modules/grbl.py
+++ b/test_tube_scanner/modules/grbl.py
@@ -4,15 +4,6 @@ GCode pour piloter la L2544 Laser Engraving Machine
GRBLController:
Commande uniquement les mouvements (X, Y)
Le mode absolue est retenu
-
- GridScanner
- Balayage complet de la grille d'éprouvettes en mode serpentin
-
- Usage:
- grbl = GRBLController()
- scan = GridScanner(grbl, xbase=100, ybase=100, duration=5)
- scan.start()
-
Created on 25 mars 2026
@author: denis@miraceti.net
@@ -170,6 +161,7 @@ class GRBLController:
if "MPos" in status:
mpos = status.split("MPos:")[1].split("|")[0]
x, y, *_ = mpos.split(",")
+ self._state(state='Mpos', msg=f"pos >>> ({x}, {y})")
return float(x), float(y)
return None, None
@@ -188,13 +180,20 @@ class GRBLController:
break
self.wait_for(0.1)
+ def send_command(self, cmd):
+ self.send(cmd)
+ self.wait_idle()
+
def move_to(self, x, y, feed=1000):
x, y = self._clamp(x, y)
#cmd = f"G0 X{x:.2f} Y{y:.2f} F{feed}" # feed is not updated in G0 mode
cmd = f"G53 G1 X{x:.2f} Y{y:.2f} F{feed}"
- self.send(cmd)
- self.wait_idle()
-
+ self.send_command(cmd)
+
+ def move_relative(self, dx=0, dy=0, feed=1000):
+ x, y = self.get_mpos() # Position actuelle
+ self.move_to(x + dx, y + dy, feed=feed)
+
def move_relative_(self, dx=0, dy=0, feed=1000):
self.send("G91") # Mode relatif
cmd = f"G0 X{dx} Y{dy} F{feed}"
@@ -202,10 +201,6 @@ class GRBLController:
self.send("G90") # Retour en mode absolu
self.wait_idle()
- def move_relative(self, dx=0, dy=0, feed=1000):
- x, y = self.get_mpos() # Position actuelle
- self.move_to(x + dx, y + dy)
-
def go_origin(self, feed=1000):
self.move_to(0, 0, feed=feed)
self.wait_for(2.0)
@@ -230,131 +225,3 @@ class GRBLController:
def close(self):
self.ser.close()
-
-
-class GridScanner:
-
- def __init__(self, grbl, process=None, **config):
- '''
- xbase # Position X de départ (col 0) en mm
- ybase # Position Y de départ (row 0) en mm
- cols # Nombre de colonnes
- rows # Nombre de lignes
- dx # Pas entre colonnes en mm
- dy # Pas entre lignes en mm
- duration # Durée de filmage par éprouvette en secondes
- feed # Vitesse de déplacement entre éprouvettes (mm/min)
- '''
- self.grbl = grbl
- self.process = process
-
- self.position = config.get('position', 'HG')
- self.xbase = config.get('xbase', 50)
- self.ybase = config.get('ybase', 50)
- self.cols = config.get('cols', 6)
- self.rows = config.get('rows', 4)
- self.dx = config.get('dx', 20)
- self.dy = config.get('dy', 19)
- self.feed = config.get('feed', 1000)
- self.duration = config.get('duration', 120) # secondes
- self.xnext = config.get('xnext', 50)
- self.ynext = config.get('ynext', 50)
-
- row_to_char = config.get('row_to_char', 'D,C,B,A')
- self.row_to_char = row_to_char.split(',')
- self.stop_playing = None
-
-
- def halt(self):
- self.process.tag.record = False
- return self.stop_playing.set()
-
- def _capture(self, uuid: str, duration: float, stop_running: Optional[threading.Event]) -> None:
- """
- Déclenche la caméra ArduCam et attend la fin de l'acquisition.
- """
- print(f"# démarrer l'enregistrement {uuid}")
- self.process.cam.on_well_change()
-
- self.process.tag.uuid = uuid
- self.process.tag.record = True
-
- start = time.monotonic()
- while not stop_running.is_set():
- if time.monotonic() - start > duration:
- break
- self.grbl.wait_for(1.0)
-
- print("# arrêter l'enregistrement")
- self.process.tag.record = False
- self.process.tag.uuid = None
-
- def start(self, xnext=None, ynext=None, position=None):
- """
- Balayage complet de la grille d'éprouvettes en mode serpentin.
-
- Parcours :
- - Lignes paires (0, 2) : gauche → droite (col 0 → col 5)
- - Lignes impaires (1, 3) : droite → gauche (col 5 → col 0)
-
- Le déplacement entre éprouvettes se fait en mode absolu via move_to().
- La caméra filme pendant `` secondes sur chaque position.
-
- Grille : 6 colonnes × 4 lignes = 24 éprouvettes
- - x = XBASE + col * PAS_X
- - y = YBASE + row * PAS_Y
- """
- try:
- if xnext is None:
- xnext = self.xnext
- if ynext is None:
- ynext = self.ynext
- if position is None:
- position = self.position
-
- max_cells = self.cols * self.rows
- cell = 0
-
- logger.info("Début du scan serpentin : %d éprouvettes, %d s/éprouvette, durée totale estimée : %d min",
- max_cells,
- self.duration,
- (max_cells * self.duration) // 60,
- )
-
- self.stop_playing = threading.Event()
- for row in range(self.rows):
- if self.stop_playing.is_set():
- break
-
- # Ordre des colonnes selon la parité de la ligne (serpentin)
- if row % 2 == 0:
- # Ligne paire : gauche → droite
- cols = range(self.cols)
- else:
- # Ligne impaire : droite → gauche
- cols = range(self.cols - 1, -1, -1)
-
- for col in cols:
- if self.stop_playing.is_set():
- break
- # Calcul de la position absolue en mm
- x = self.xbase + col * self.dx
- y = self.ybase + row * self.dy
- cell += 1
-
- logger.info(
- "[%02d/%02d] row=%d col=%d → X=%.1f mm Y=%.1f mm",
- cell, max_cells, row, col, x, y,
- )
-
- self.grbl.move_to(x, y, feed=self.feed)
-
- uuid = f'{self.process.tag.session}-{position}-{self.row_to_char[row]}{col+1}'
- self._capture(uuid, self.duration, self.stop_playing)
-
- # Retour à nexr après le scan
- logger.info("Scan terminé — retour à l'origine (X=%.1f Y=%.1f)", xnext, ynext)
- self.grbl.move_to(xnext, ynext, feed=self.feed*2)
- except Exception as e:
- logger.error(f"scan error: {e}")
-
diff --git a/test_tube_scanner/modules/picamera2_capture.py b/test_tube_scanner/modules/picamera2_capture.py
index cc5354b..de5d09b 100644
--- a/test_tube_scanner/modules/picamera2_capture.py
+++ b/test_tube_scanner/modules/picamera2_capture.py
@@ -45,6 +45,10 @@ class PiCamera2Capture(VideoCaptureInterface):
jpeg_quality: int = 85,
camera_index: int = 0,
use_video_config: bool = True,
+ use_tracking: bool = False,
+ px_per_mm: float = 2.1,
+ display = None,
+
):
"""
:param fps: Cadence cible en images par seconde
@@ -55,7 +59,7 @@ class PiCamera2Capture(VideoCaptureInterface):
:param use_video_config: True = VideoConfiguration (flux continu, basse latence)
False = StillConfiguration (haute résolution, plus lent)
"""
- super().__init__(fps=fps)
+ super().__init__(fps=fps, use_tracking=use_tracking, px_per_mm=px_per_mm, display=display)
self._width: int = width
self._height: int = height
self._jpeg_quality: int = jpeg_quality
diff --git a/test_tube_scanner/modules/tube_aligner.py b/test_tube_scanner/modules/tube_aligner.py
new file mode 100644
index 0000000..3e176c2
--- /dev/null
+++ b/test_tube_scanner/modules/tube_aligner.py
@@ -0,0 +1,315 @@
+'''
+Created on 17 avr. 2026
+
+@author: denis
+'''
+# modules/tube_aligner.py
+
+import cv2
+import logging
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+class TubeAligner:
+
+ GRBL_THRESHOLD_PX = 20
+ DEAD_ZONE_PX = 5
+
+ def __init__(
+ self,
+ px_per_mm : float = 10.0,
+ grbl_threshold_px : int = 20,
+ dead_zone_px : int = 5,
+ debug : bool = False, # ← activable depuis la vue
+ display = None,
+ ):
+ self.TUBE_DIAMETER_MM = 16.0
+ self.grbl_threshold_px = grbl_threshold_px
+ self.dead_zone_px = dead_zone_px
+ self.debug = debug
+ self.display = display
+
+ # ------------------------------------------------------------------ #
+ # Détection principale
+ # ------------------------------------------------------------------ #
+
+ def detect_tube(self, frame: np.ndarray) -> dict:
+ h, w = frame.shape[:2]
+ cx_img = w // 2
+ cy_img = h // 2
+
+ result = {
+ "detected" : False,
+ "tube_cx" : None,
+ "tube_cy" : None,
+ "tube_radius" : None,
+ "offset_x_px" : 0,
+ "offset_y_px" : 0,
+ "offset_x_mm" : 0.0,
+ "offset_y_mm" : 0.0,
+ "action" : "none",
+ "frame_annotated": None,
+ }
+ frame_out = frame.copy()
+
+ gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+ blurred = cv2.GaussianBlur(gray, (15, 15), 3)
+
+ # 3 configurations légèrement différentes — vote majoritaire
+ # Fonctionne sur fond sombre ET fond clair
+ configs = [
+ dict(param1=50, param2=30, minRadius=int(min(w,h)*0.26), maxRadius=int(min(w,h)*0.36)),
+ dict(param1=60, param2=30, minRadius=int(min(w,h)*0.26), maxRadius=int(min(w,h)*0.37)),
+ dict(param1=50, param2=28, minRadius=int(min(w,h)*0.25), maxRadius=int(min(w,h)*0.365)),
+ ]
+
+ all_cx, all_cy, all_r = [], [], []
+ for cfg in configs:
+ circles = cv2.HoughCircles(
+ blurred,
+ cv2.HOUGH_GRADIENT,
+ dp=1.2,
+ minDist=min(w, h) // 2,
+ **cfg
+ )
+ if circles is not None:
+ c = np.round(circles[0]).astype(int)
+ best = min(c, key=lambda c: np.sqrt((c[0]-cx_img)**2 + (c[1]-cy_img)**2))
+ all_cx.append(int(best[0]))
+ all_cy.append(int(best[1]))
+ all_r.append(int(best[2]))
+
+ if not all_cx:
+ logger.warning("TubeAligner: aucun cercle détecté (%dx%d)", w, h)
+ if self.debug:
+ frame_out = self._draw_debug_no_detection(frame_out, cx_img, cy_img)
+ result["frame_annotated"] = frame_out
+ return result
+
+ # Moyenne des détections convergentes
+ tx = int(np.mean(all_cx))
+ ty = int(np.mean(all_cy))
+ tr = int(np.mean(all_r))
+ if tr > 0:
+ self.px_per_mm = (2 * tr) / 16.0
+
+ offset_x_px = tx - cx_img
+ offset_y_px = ty - cy_img
+ #offset_x_mm = offset_x_px / self.px_per_mm
+ #offset_y_mm = offset_y_px /self. px_per_mm
+
+ offset_x_mm = offset_y_px /self. px_per_mm # (X CNC = Y image)
+ offset_y_mm = -offset_x_px / self.px_per_mm # (Y CNC = -X image)
+
+ dist_px = float(np.sqrt(offset_x_px**2 + offset_y_px**2))
+
+ if dist_px <= self.dead_zone_px:
+ action = "none"
+ elif dist_px <= self.grbl_threshold_px:
+ action = "crop"
+ else:
+ action = "grbl"
+
+ if self.debug:
+ frame_out = self._draw_debug(
+ frame_out, cx_img, cy_img,
+ tx, ty, tr,
+ offset_x_px, offset_y_px,
+ offset_x_mm, offset_y_mm,
+ dist_px, action,
+ votes=len(all_cx), # ← affiche le nombre de configs ayant détecté
+ )
+
+ result.update({
+ "detected" : True,
+ "tube_cx" : tx,
+ "tube_cy" : ty,
+ "tube_radius" : tr,
+ "offset_x_px" : offset_x_px,
+ "offset_y_px" : offset_y_px,
+ "offset_x_mm" : round(offset_x_mm, 3),
+ "offset_y_mm" : round(offset_y_mm, 3),
+ "action" : action,
+ "frame_annotated": frame_out,
+ })
+ return result
+
+
+ def crop_to_tube(self, frame: np.ndarray, detection: dict) -> np.ndarray:
+ """
+ Recadrage logiciel : recentre l'image sur le tube détecté.
+ Utilisé quand action == "crop".
+ """
+ if not detection["detected"]:
+ return frame
+
+ tx = detection["tube_cx"]
+ ty = detection["tube_cy"]
+ tr = detection["tube_radius"]
+ h, w = frame.shape[:2]
+
+ # Fenêtre carrée autour du centre du tube
+ half = tr
+ x1 = max(tx - half, 0)
+ y1 = max(ty - half, 0)
+ x2 = min(tx + half, w)
+ y2 = min(ty + half, h)
+
+ cropped = frame[y1:y2, x1:x2]
+
+ # Redimensionne à la taille originale pour ne pas changer le pipeline
+ return cv2.resize(cropped, (w, h), interpolation=cv2.INTER_LINEAR)
+
+
+ def _detect_center_stable(
+ self,
+ capture_func, # callable() → frame bytes
+ n_samples: int = 5,
+ delay_s: float = 0.3,
+ ) -> tuple[float, float] | None:
+ """
+ Capture N frames et retourne le centre moyen du tube.
+ Réduit l'erreur de détection d'un facteur √N.
+
+ :param capture_func: callable sans argument → bytes JPEG
+ :param n_samples: nombre de captures à moyenner
+ :param delay_s: pause entre chaque capture
+ :return: (cx_mean, cy_mean) ou None si échec
+ """
+ import time
+ centers = []
+
+ for i in range(n_samples):
+ if i > 0:
+ time.sleep(delay_s)
+
+ frame_bytes = capture_func()
+ nparr = np.frombuffer(frame_bytes, np.uint8)
+ frame = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+
+ if frame is None:
+ continue
+
+ detection = self.detect_tube(frame)
+ if detection["detected"]:
+ centers.append((detection["tube_cx"], detection["tube_cy"]))
+ logger.debug(
+ "_detect_center_stable [%d/%d] : cx=%d cy=%d",
+ i+1, n_samples,
+ detection["tube_cx"], detection["tube_cy"],
+ )
+ else:
+ logger.warning("_detect_center_stable [%d/%d] : tube non détecté", i+1, n_samples)
+
+ if len(centers) < 3:
+ logger.error("_detect_center_stable : seulement %d détections valides", len(centers))
+ return None
+
+ # Filtre les valeurs aberrantes (écart > 2 sigma)
+ cx_arr = np.array([c[0] for c in centers], dtype=float)
+ cy_arr = np.array([c[1] for c in centers], dtype=float)
+
+ cx_mean, cx_std = np.mean(cx_arr), np.std(cx_arr)
+ cy_mean, cy_std = np.mean(cy_arr), np.std(cy_arr)
+
+ mask = (
+ (np.abs(cx_arr - cx_mean) <= 2 * cx_std) &
+ (np.abs(cy_arr - cy_mean) <= 2 * cy_std)
+ )
+ filtered = [(cx_arr[i], cy_arr[i]) for i in range(len(centers)) if mask[i]]
+
+ if not filtered:
+ filtered = centers # fallback si tout est filtré
+
+ cx_final = float(np.mean([c[0] for c in filtered]))
+ cy_final = float(np.mean([c[1] for c in filtered]))
+
+ logger.info(
+ "_detect_center_stable : %d/%d valides cx=%.1f±%.1f cy=%.1f±%.1f",
+ len(filtered), n_samples,
+ cx_final, cx_std, cy_final, cy_std,
+ )
+ return cx_final, cy_final
+
+
+
+ def calib_reset(self):
+ pass
+
+ # ------------------------------------------------------------------ #
+ # Dessin debug
+ # ------------------------------------------------------------------ #
+
+ def _draw_debug(
+ self, frame, cx_img, cy_img,
+ tx, ty, tr,
+ offset_x_px, offset_y_px,
+ offset_x_mm, offset_y_mm,
+ dist_px, action, votes: int = 3
+ ) -> np.ndarray:
+
+ # Couleur selon action
+ color = {
+ "none" : (0, 255, 0), # vert — centré
+ "crop" : (0, 200, 255), # orange — recadrage
+ "grbl" : (0, 0, 255), # rouge — correction CNC
+ }.get(action, (200, 200, 200))
+
+ # Cercle intérieur du tube
+ cv2.circle(frame, (tx, ty), tr, color, 2, cv2.LINE_AA)
+
+ # Rayon de zone morte (dead zone) en vert clair
+ cv2.circle(frame, (cx_img, cy_img), self.dead_zone_px,
+ (0, 255, 100), 1, cv2.LINE_AA)
+
+ # Rayon seuil GRBL en rouge pointillé (simulé par cercle fin)
+ cv2.circle(frame, (cx_img, cy_img), self.grbl_threshold_px,
+ (0, 80, 255), 1, cv2.LINE_AA)
+
+ # Croix centre image
+ cv2.drawMarker(frame, (cx_img, cy_img),
+ (255, 255, 255), cv2.MARKER_CROSS, 24, 1, cv2.LINE_AA)
+
+ # Centre tube
+ cv2.circle(frame, (tx, ty), 5, color, -1, cv2.LINE_AA)
+
+ # Vecteur offset centre image → centre tube
+ if dist_px > self.dead_zone_px:
+ cv2.arrowedLine(frame, (cx_img, cy_img), (tx, ty),
+ color, 2, cv2.LINE_AA, tipLength=0.2)
+
+ # Panneau info — fond semi-transparent
+ overlay = frame.copy()
+ cv2.rectangle(overlay, (8, 8), (400, 130), (0, 0, 0), -1)
+ cv2.addWeighted(overlay, 0.45, frame, 0.55, 0, frame)
+
+ lines = [
+ (f"Tube cx={tx} cy={ty} r={tr}px", (0, 255, 180)),
+ (f"Offset dx={offset_x_px:+d}px dy={offset_y_px:+d}px", color),
+ (f"Offset dx={offset_x_mm:+.3f}mm dy={offset_y_mm:+.3f}mm", color),
+ (f"Dist={dist_px:.1f}px action={action.upper()}", color),
+ (f"px/mm={self.px_per_mm:.4f} votes={votes}/3", (180, 180, 180)), # ← votes
+ ]
+
+ for i, (text, col) in enumerate(lines):
+ cv2.putText(frame, text, (14, 30 + i * 20),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.48, col, 1, cv2.LINE_AA)
+
+ # Légende zones
+ cv2.putText(frame, "dead zone", (cx_img + self.dead_zone_px + 3, cy_img - 3),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.35, (0, 255, 100), 1)
+ cv2.putText(frame, "GRBL threshold", (cx_img + self.grbl_threshold_px + 3, cy_img + 6),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.35, (0, 80, 255), 1)
+ return frame
+
+ def _draw_debug_no_detection(self, frame, cx_img, cy_img) -> np.ndarray:
+ cv2.drawMarker(frame, (cx_img, cy_img),
+ (255, 255, 255), cv2.MARKER_CROSS, 24, 1, cv2.LINE_AA)
+ cv2.putText(frame, "Tube non detecte", (14, 30),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2, cv2.LINE_AA)
+ return frame
+
+
+
\ No newline at end of file
diff --git a/test_tube_scanner/modules/tube_aligner_old.py b/test_tube_scanner/modules/tube_aligner_old.py
new file mode 100644
index 0000000..2d38fec
--- /dev/null
+++ b/test_tube_scanner/modules/tube_aligner_old.py
@@ -0,0 +1,230 @@
+'''
+Created on 17 avr. 2026
+
+@author: denis
+'''
+# modules/tube_aligner.py
+
+import cv2
+import logging
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+class TubeAligner:
+ """
+ Détecte le cercle du tube à essai dans une frame (vue par dessous,
+ éclairage par dessus → cercle clair sur fond sombre).
+ Calcule le décalage entre le centre du tube et le centre de l'image.
+ Décide d'une correction GRBL (grand écart) ou d'un recadrage (petit écart).
+ """
+
+ # Seuil en pixels : au-delà → correction GRBL, en-dessous → recadrage
+ GRBL_THRESHOLD_PX = 20
+ # Tolérance : en-dessous → pas de correction nécessaire
+ DEAD_ZONE_PX = 5
+
+ def __init__(
+ self,
+ px_per_mm: float = 10.0, # facteur d'échelle calibration (px/mm)
+ grbl_threshold_px: int = 20,
+ dead_zone_px: int = 5,
+ debug: bool = False, # ← activable depuis la vue
+ ):
+ self.px_per_mm = px_per_mm
+ self.grbl_threshold_px = grbl_threshold_px
+ self.dead_zone_px = dead_zone_px
+
+ def detect_tube(self, frame: np.ndarray) -> dict:
+ """
+ Détecte le cercle du tube et calcule le décalage par rapport au centre image.
+
+ :param frame: Frame BGR (numpy array)
+ :return: dict avec cercle détecté, décalage px et mm, action recommandée
+ """
+ h, w = frame.shape[:2]
+ cx_img = w // 2
+ cy_img = h // 2
+
+ result = {
+ "detected" : False,
+ "tube_cx" : None,
+ "tube_cy" : None,
+ "tube_radius" : None,
+ "offset_x_px" : None,
+ "offset_y_px" : None,
+ "offset_x_mm" : None,
+ "offset_y_mm" : None,
+ "action" : "none", # "none" | "crop" | "grbl"
+ "grbl_gcode" : None,
+ "frame_annotated": None,
+ }
+
+ # Prétraitement : niveaux de gris + flou
+ gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+ #blurred = cv2.GaussianBlur(gray, (9, 9), 2)
+ blurred = cv2.GaussianBlur(gray, (15, 15), 3)
+
+ # param1 : seuil Canny haut, param2 : seuil accumulation (plus bas = plus permissif)
+ min_radius = int(min(w, h) * 0.26) # ~260px sur 1000px
+ max_radius = int(min(w, h) * 0.36) # ~360px sur 1000px — bord intérieur du verre
+
+ circles = cv2.HoughCircles(
+ blurred,
+ cv2.HOUGH_GRADIENT,
+ dp = 1.2,
+ minDist = min(w, h) // 2, # un seul tube attendu
+ param1 = 50,
+ param2 = 30,
+ minRadius = min_radius,
+ maxRadius = max_radius,
+ )
+
+ if circles is None:
+ logger.warning("TubeAligner: aucun cercle détecté")
+ result["frame_annotated"] = self._annotate(frame.copy(), cx_img, cy_img, None)
+ return result
+
+ circles = np.round(circles[0, :]).astype(int)
+
+ # Prend le cercle le plus proche du centre image
+ best = min(
+ circles,
+ key=lambda c: np.sqrt((c[0] - cx_img)**2 + (c[1] - cy_img)**2)
+ )
+ tx, ty, tr = int(best[0]), int(best[1]), int(best[2])
+
+ # Décalage : positif = tube à droite/bas du centre image
+ offset_x_px = tx - cx_img
+ offset_y_px = ty - cy_img
+ offset_x_mm = offset_x_px / self.px_per_mm
+ offset_y_mm = offset_y_px / self.px_per_mm
+
+ dist_px = np.sqrt(offset_x_px**2 + offset_y_px**2)
+
+ # Décision d'action
+ if dist_px <= self.dead_zone_px:
+ action = "none"
+ grbl_gcode = None
+ elif dist_px <= self.grbl_threshold_px:
+ action = "crop"
+ grbl_gcode = None
+ else:
+ action = "grbl"
+ # G91 = coordonnées relatives, G0 = déplacement rapide
+ # Inversion du signe : si tube est à droite (+X image),
+ # la CNC doit reculer (-X GRBL) pour recentrer
+ #cmd = f"G53 G1 X{x:.2f} Y{y:.2f} F{feed}"
+
+ grbl_gcode = (
+ f"G91\n"
+ f"G1 X{-offset_x_mm:.3f} Y{-offset_y_mm:.3f}\n"
+ f"G90"
+ )
+ logger.info(
+ "TubeAligner: décalage %.1fpx (%.2fmm, %.2fmm) → GRBL: %s",
+ dist_px, offset_x_mm, offset_y_mm, grbl_gcode.replace('\n', ' | ')
+ )
+
+ result.update({
+ "detected" : True,
+ "tube_cx" : tx,
+ "tube_cy" : ty,
+ "tube_radius" : tr,
+ "offset_x_px" : offset_x_px,
+ "offset_y_px" : offset_y_px,
+ "offset_x_mm" : round(offset_x_mm, 3),
+ "offset_y_mm" : round(offset_y_mm, 3),
+ "action" : action,
+ "grbl_gcode" : None,
+ "frame_annotated": self._annotate(
+ frame.copy(), cx_img, cy_img, (tx, ty, tr), offset_x_px, offset_y_px
+ ),
+ })
+
+ return result
+
+ def crop_to_tube(self, frame: np.ndarray, detection: dict) -> np.ndarray:
+ """
+ Recadrage logiciel : recentre l'image sur le tube détecté.
+ Utilisé quand action == "crop".
+ """
+ if not detection["detected"]:
+ return frame
+
+ tx = detection["tube_cx"]
+ ty = detection["tube_cy"]
+ tr = detection["tube_radius"]
+ h, w = frame.shape[:2]
+
+ # Fenêtre carrée autour du centre du tube
+ half = tr
+ x1 = max(tx - half, 0)
+ y1 = max(ty - half, 0)
+ x2 = min(tx + half, w)
+ y2 = min(ty + half, h)
+
+ cropped = frame[y1:y2, x1:x2]
+
+ # Redimensionne à la taille originale pour ne pas changer le pipeline
+ return cv2.resize(cropped, (w, h), interpolation=cv2.INTER_LINEAR)
+
+ def _annotate(
+ self,
+ frame: np.ndarray,
+ cx_img: int,
+ cy_img: int,
+ circle: tuple | None,
+ offset_x: int = 0,
+ offset_y: int = 0,
+ ) -> np.ndarray:
+ """
+ Dessine le cercle détecté, le centre image et le vecteur de décalage.
+ """
+ # Croix centre image
+ cv2.drawMarker(
+ frame, (cx_img, cy_img),
+ (0, 255, 0), cv2.MARKER_CROSS, 20, 1, cv2.LINE_AA
+ )
+
+ if circle is None:
+ cv2.putText(frame, "Tube non detecte", (10, 30),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 1, cv2.LINE_AA)
+ return frame
+
+ tx, ty, tr = circle
+
+ # Cercle du tube en cyan
+ cv2.circle(frame, (tx, ty), tr, (255, 255, 0), 2, cv2.LINE_AA)
+
+ # Centre du tube en rouge
+ cv2.circle(frame, (tx, ty), 4, (0, 0, 255), -1, cv2.LINE_AA)
+
+ # Vecteur décalage (centre image → centre tube)
+ if abs(offset_x) > 2 or abs(offset_y) > 2:
+ cv2.arrowedLine(
+ frame,
+ (cx_img, cy_img),
+ (tx, ty),
+ (0, 100, 255), 2, cv2.LINE_AA, tipLength=0.2
+ )
+
+ # Texte décalage
+ dist = np.sqrt(offset_x**2 + offset_y**2)
+ cv2.putText(
+ frame,
+ f"dx={offset_x:+d}px dy={offset_y:+d}px dist={dist:.1f}px",
+ (10, 28),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 255, 255), 1, cv2.LINE_AA,
+ )
+ cv2.putText(
+ frame,
+ f"r={tr}px",
+ (10, 48),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.45, (200, 200, 200), 1, cv2.LINE_AA,
+ )
+
+ return frame
+
+
\ No newline at end of file
diff --git a/test_tube_scanner/modules/tube_aligner_old2.py b/test_tube_scanner/modules/tube_aligner_old2.py
new file mode 100644
index 0000000..a775b21
--- /dev/null
+++ b/test_tube_scanner/modules/tube_aligner_old2.py
@@ -0,0 +1,283 @@
+'''
+Created on 17 avr. 2026
+
+@author: denis
+'''
+# modules/tube_aligner.py
+
+import cv2
+import logging
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+class TubeAligner:
+
+ GRBL_THRESHOLD_PX = 20
+ DEAD_ZONE_PX = 5
+
+ def __init__(
+ self,
+ px_per_mm : float = 10.0,
+ grbl_threshold_px : int = 20,
+ dead_zone_px : int = 5,
+ debug : bool = False, # ← activable depuis la vue
+ ):
+ self.px_per_mm = px_per_mm
+ self.grbl_threshold_px = grbl_threshold_px
+ self.dead_zone_px = dead_zone_px
+ self.debug = debug
+
+ # Etat calibration
+ self._calib_step = 0 # 0=idle 1=point A enregistré
+ self._calib_pos_A_px = None # centre tube point A en px
+ self._calib_mpos_A = None # position CNC point A en mm
+
+ # ------------------------------------------------------------------ #
+ # Détection principale
+ # ------------------------------------------------------------------ #
+
+ def detect_tube(self, frame: np.ndarray) -> dict:
+ h, w = frame.shape[:2]
+ cx_img = w // 2
+ cy_img = h // 2
+
+ result = {
+ "detected" : False,
+ "tube_cx" : None,
+ "tube_cy" : None,
+ "tube_radius" : None,
+ "offset_x_px" : 0,
+ "offset_y_px" : 0,
+ "offset_x_mm" : 0.0,
+ "offset_y_mm" : 0.0,
+ "action" : "none",
+ "frame_annotated": None,
+ }
+
+ gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
+ blurred = cv2.GaussianBlur(gray, (15, 15), 3)
+
+ circles = cv2.HoughCircles(
+ blurred,
+ cv2.HOUGH_GRADIENT,
+ dp = 1.2,
+ minDist = min(w, h) // 2,
+ param1 = 50,
+ param2 = 30,
+ minRadius = int(min(w, h) * 0.26),
+ maxRadius = int(min(w, h) * 0.36),
+ )
+
+ frame_out = frame.copy()
+
+ if circles is None:
+ logger.warning("TubeAligner: aucun cercle détecté")
+ if self.debug:
+ frame_out = self._draw_debug_no_detection(frame_out, cx_img, cy_img)
+ result["frame_annotated"] = frame_out
+ return result
+
+ circles = np.round(circles[0, :]).astype(int)
+ best = min(circles, key=lambda c: np.sqrt((c[0]-cx_img)**2 + (c[1]-cy_img)**2))
+ tx, ty, tr = int(best[0]), int(best[1]), int(best[2])
+
+ offset_x_px = tx - cx_img
+ offset_y_px = ty - cy_img
+ offset_x_mm = offset_x_px / self.px_per_mm
+ offset_y_mm = offset_y_px / self.px_per_mm
+ dist_px = np.sqrt(offset_x_px**2 + offset_y_px**2)
+
+ if dist_px <= self.dead_zone_px:
+ action = "none"
+ elif dist_px <= self.grbl_threshold_px:
+ action = "crop"
+ else:
+ action = "grbl"
+
+ if self.debug:
+ frame_out = self._draw_debug(
+ frame_out, cx_img, cy_img,
+ tx, ty, tr,
+ offset_x_px, offset_y_px,
+ offset_x_mm, offset_y_mm,
+ dist_px, action,
+ )
+
+
+ result.update({
+ "detected" : True,
+ "tube_cx" : tx,
+ "tube_cy" : ty,
+ "tube_radius" : tr,
+ "offset_x_px" : offset_x_px,
+ "offset_y_px" : offset_y_px,
+ "offset_x_mm" : round(offset_x_mm, 3),
+ "offset_y_mm" : round(offset_y_mm, 3),
+ "action" : action,
+ "frame_annotated": frame_out,
+ })
+ return result
+
+ # ------------------------------------------------------------------ #
+ # Calibration px/mm — 2 points
+ # ------------------------------------------------------------------ #
+
+ def calib_record_point_A(self, detection: dict, mpos: tuple) -> bool:
+ """
+ Enregistre le point A (position CNC + centre tube en px).
+ Appeler quand la CNC est immobile sur le point A.
+
+ :param detection: résultat de detect_tube()
+ :param mpos: (x_mm, y_mm) retourné par cnc.get_mpos()
+ :return: True si enregistré
+ """
+ if not detection["detected"]:
+ logger.warning("calib_record_point_A: tube non détecté")
+ return False
+
+ self._calib_pos_A_px = (detection["tube_cx"], detection["tube_cy"])
+ self._calib_mpos_A = mpos
+ self._calib_step = 1
+ logger.info("Calibration point A : px=%s mpos=%s", self._calib_pos_A_px, mpos)
+ return True
+
+ def calib_record_point_B(self, detection: dict, mpos: tuple) -> dict | None:
+ """
+ Enregistre le point B et calcule px_per_mm.
+ Appeler après déplacement CNC manuel d'une distance connue.
+
+ :param detection: résultat de detect_tube()
+ :param mpos: (x_mm, y_mm) retourné par cnc.get_mpos()
+ :return: dict résultat calibration ou None si échec
+ """
+ if self._calib_step != 1:
+ logger.warning("calib_record_point_B: point A non enregistré")
+ return None
+
+ if not detection["detected"]:
+ logger.warning("calib_record_point_B: tube non détecté")
+ return None
+
+ pos_B_px = (detection["tube_cx"], detection["tube_cy"])
+ mpos_B = mpos
+
+ # Déplacement en px
+ dpx = np.sqrt(
+ (pos_B_px[0] - self._calib_pos_A_px[0])**2 +
+ (pos_B_px[1] - self._calib_pos_A_px[1])**2
+ )
+ # Déplacement en mm (distance euclidienne CNC)
+ dmm = np.sqrt(
+ (mpos_B[0] - self._calib_mpos_A[0])**2 +
+ (mpos_B[1] - self._calib_mpos_A[1])**2
+ )
+
+ if dmm < 0.1 or dpx < 2:
+ logger.warning("Déplacement trop faible : dpx=%.1f dmm=%.3f", dpx, dmm)
+ return None
+
+ px_per_mm_new = dpx / dmm
+ self.px_per_mm = px_per_mm_new
+ self._calib_step = 0
+
+ result = {
+ "px_per_mm" : round(px_per_mm_new, 4),
+ "mm_per_px" : round(dmm / dpx, 6),
+ "delta_px" : round(dpx, 2),
+ "delta_mm" : round(dmm, 3),
+ "point_A_px" : self._calib_pos_A_px,
+ "point_B_px" : pos_B_px,
+ "mpos_A" : self._calib_mpos_A,
+ "mpos_B" : mpos_B,
+ }
+ logger.info("Calibration OK : %.4f px/mm (%.6f mm/px)", px_per_mm_new, dmm/dpx)
+ return result
+
+ def calib_reset(self):
+ self._calib_step = 0
+ self._calib_pos_A_px = None
+ self._calib_mpos_A = None
+
+ # ------------------------------------------------------------------ #
+ # Dessin debug
+ # ------------------------------------------------------------------ #
+
+ def _draw_debug(
+ self, frame, cx_img, cy_img,
+ tx, ty, tr,
+ offset_x_px, offset_y_px,
+ offset_x_mm, offset_y_mm,
+ dist_px, action,
+ ) -> np.ndarray:
+
+ # Couleur selon action
+ color = {
+ "none" : (0, 255, 0), # vert — centré
+ "crop" : (0, 200, 255), # orange — recadrage
+ "grbl" : (0, 0, 255), # rouge — correction CNC
+ }.get(action, (200, 200, 200))
+
+ # Cercle intérieur du tube
+ cv2.circle(frame, (tx, ty), tr, color, 2, cv2.LINE_AA)
+
+ # Rayon de zone morte (dead zone) en vert clair
+ cv2.circle(frame, (cx_img, cy_img), self.dead_zone_px,
+ (0, 255, 100), 1, cv2.LINE_AA)
+
+ # Rayon seuil GRBL en rouge pointillé (simulé par cercle fin)
+ cv2.circle(frame, (cx_img, cy_img), self.grbl_threshold_px,
+ (0, 80, 255), 1, cv2.LINE_AA)
+
+ # Croix centre image
+ cv2.drawMarker(frame, (cx_img, cy_img),
+ (255, 255, 255), cv2.MARKER_CROSS, 24, 1, cv2.LINE_AA)
+
+ # Centre tube
+ cv2.circle(frame, (tx, ty), 5, color, -1, cv2.LINE_AA)
+
+ # Vecteur offset centre image → centre tube
+ if dist_px > self.dead_zone_px:
+ cv2.arrowedLine(frame, (cx_img, cy_img), (tx, ty),
+ color, 2, cv2.LINE_AA, tipLength=0.2)
+
+ # Panneau info — fond semi-transparent
+ overlay = frame.copy()
+ cv2.rectangle(overlay, (8, 8), (400, 130), (0, 0, 0), -1)
+ cv2.addWeighted(overlay, 0.45, frame, 0.55, 0, frame)
+
+ lines = [
+ (f"Tube cx={tx} cy={ty} r={tr}px", (0, 255, 180)),
+ (f"Offset dx={offset_x_px:+d}px dy={offset_y_px:+d}px", color),
+ (f"Offset dx={offset_x_mm:+.3f}mm dy={offset_y_mm:+.3f}mm", color),
+ (f"Dist={dist_px:.1f}px action={action.upper()}", color),
+ (f"px/mm={self.px_per_mm:.4f}", (180, 180, 180)),
+ ]
+ for i, (text, col) in enumerate(lines):
+ cv2.putText(frame, text, (14, 30 + i * 20),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.48, col, 1, cv2.LINE_AA)
+
+ # Légende zones
+ cv2.putText(frame, "dead zone", (cx_img + self.dead_zone_px + 3, cy_img - 3),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.35, (0, 255, 100), 1)
+ cv2.putText(frame, "GRBL threshold", (cx_img + self.grbl_threshold_px + 3, cy_img - 3),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.35, (0, 80, 255), 1)
+
+ # Indicateur calibration en cours
+ if self._calib_step == 1:
+ cv2.putText(frame, "CALIB — En attente point B",
+ (14, frame.shape[0] - 14),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.55, (0, 200, 255), 2, cv2.LINE_AA)
+
+ return frame
+
+ def _draw_debug_no_detection(self, frame, cx_img, cy_img) -> np.ndarray:
+ cv2.drawMarker(frame, (cx_img, cy_img),
+ (255, 255, 255), cv2.MARKER_CROSS, 24, 1, cv2.LINE_AA)
+ cv2.putText(frame, "Tube non detecte", (14, 30),
+ cv2.FONT_HERSHEY_SIMPLEX, 0.6, (0, 0, 255), 2, cv2.LINE_AA)
+ return frame
+
+
+
\ No newline at end of file
diff --git a/test_tube_scanner/modules/videofile_capture.py b/test_tube_scanner/modules/videofile_capture.py
index 5c9c29b..ee752a3 100644
--- a/test_tube_scanner/modules/videofile_capture.py
+++ b/test_tube_scanner/modules/videofile_capture.py
@@ -38,7 +38,10 @@ class VideoFileCapture(VideoCaptureInterface):
jpeg_quality: int = 85,
width: Optional[int] = None,
height: Optional[int] = None,
- video_lists = []
+ video_lists = [],
+ use_tracking: bool = False,
+ px_per_mm: float = 2.1,
+ display = None,
):
"""
:param video_file: fichier video
@@ -47,12 +50,13 @@ class VideoFileCapture(VideoCaptureInterface):
:param width: Largeur souhaitée (None = valeur par défaut du pilote)
:param height: Hauteur souhaitée (None = valeur par défaut du pilote)
"""
- super().__init__(fps=fps)
+ super().__init__(fps=fps, use_tracking=use_tracking, px_per_mm=px_per_mm, display=display)
self._video_file: str = video_file
self._jpeg_quality: int = jpeg_quality
self._width: Optional[int] = width
self._height: Optional[int] = height
self._video_lists = video_lists
+
self.ptf = 0
self._cap = None # Instance cv2.VideoCapture
diff --git a/test_tube_scanner/modules/webcam_capture.py b/test_tube_scanner/modules/webcam_capture.py
index 13e7e72..cd8d7bf 100644
--- a/test_tube_scanner/modules/webcam_capture.py
+++ b/test_tube_scanner/modules/webcam_capture.py
@@ -38,6 +38,9 @@ class WebcamCapture(VideoCaptureInterface):
jpeg_quality: int = 85,
width: Optional[int] = None,
height: Optional[int] = None,
+ use_tracking: bool = False,
+ px_per_mm: float = 2.1,
+ display = None,
):
"""
:param device_index: Index du périphérique V4L2 (0 = première webcam)
@@ -46,7 +49,7 @@ class WebcamCapture(VideoCaptureInterface):
:param width: Largeur souhaitée (None = valeur par défaut du pilote)
:param height: Hauteur souhaitée (None = valeur par défaut du pilote)
"""
- super().__init__(fps=fps)
+ super().__init__(fps=fps, use_tracking=use_tracking, px_per_mm=px_per_mm, display=display)
self._device_index: int = device_index
self._jpeg_quality: int = jpeg_quality
self._width: Optional[int] = width
diff --git a/test_tube_scanner/scanner/admin.py b/test_tube_scanner/scanner/admin.py
index fcb311f..036931c 100644
--- a/test_tube_scanner/scanner/admin.py
+++ b/test_tube_scanner/scanner/admin.py
@@ -7,11 +7,16 @@ class WellAdmin(admin.ModelAdmin):
list_display = ('name', 'author',)
class ConfigurationAdmin(admin.ModelAdmin):
- list_display = ('name', 'author', 'use_rpicam', 'video_width_capture', 'video_height_capture', 'video_frame_rate', 'active',)
+ list_display = ('name', 'author', 'use_rpicam', 'video_width_capture', 'video_height_capture', 'video_frame_rate', 'px_per_mm', 'active',)
class MultiWellAdmin(admin.ModelAdmin):
- list_filter = ('author',)
- list_display = ('label', 'position', 'author', 'order', 'xbase', 'ybase', 'duration', 'feed', 'active',)
+ list_filter = ('author', )
+ list_display = ('label', 'position', 'author', 'order', 'xbase', 'ybase', 'duration', 'feed', 'default', 'well_position', 'active',)
+
+class WellPositionAdmin(admin.ModelAdmin):
+ list_filter = ('author', 'multiwell')
+ list_display = ('multiwell__position', 'well__name', 'order', 'x', 'y', 'author',)
+
class ObservationMultiWellDetailInline(admin.TabularInline):
model = models.ObservationMultiWellDetail
@@ -60,5 +65,6 @@ class SessionAdmin(admin.ModelAdmin):
admin.site.register(models.Configuration, ConfigurationAdmin)
admin.site.register(models.Well, WellAdmin)
admin.site.register(models.MultiWell, MultiWellAdmin)
+admin.site.register(models.WellPostion, WellPositionAdmin)
admin.site.register(models.Observation, ObservationAdmin)
admin.site.register(models.Session, SessionAdmin)
diff --git a/test_tube_scanner/scanner/models.py b/test_tube_scanner/scanner/models.py
index ed6a63b..5f306c9 100644
--- a/test_tube_scanner/scanner/models.py
+++ b/test_tube_scanner/scanner/models.py
@@ -44,6 +44,8 @@ class Configuration(models.Model):
# Grbl configuration
grbl_xmax = models.FloatField(_("Grbl Xmax"), help_text=_("CNC Grbl Xmax en mm"), blank=False, default=350.0)
grbl_ymax = models.FloatField(_("Grbl Ymax"), help_text=_("CNC Grbl Ymax en mm"), blank=False, default=250.0)
+ px_per_mm = models.FloatField(_("Pixel / mm"), help_text=_('Rapport pixel / déplacement en pixel/mm'), blank=False, default=2.5)
+
# camera configuration
use_rpicam = models.BooleanField(_("Utiliser rpicam"), help_text=_("Par défaaut. Sinon USB webcam"), default=True)
webcam_device_index = models.PositiveSmallIntegerField(_("Index de la webcam"), help_text=_("Index de la webcam (0, 1, ...) si présente"), default=2)
@@ -57,7 +59,7 @@ class Configuration(models.Model):
calibration_default_multiwell = models.CharField(_("Multi-puits de calibration par défaut"), help_text=_("Position du multi-puits de calibration par défaut"), max_length=8, choices=MULTIWELL_POSITION, default='HG')
calibration_default_feed = models.PositiveIntegerField(_("Vitesse de calibration"), help_text=_("Vitesse de déplacement pour la calibration en mm/mn"), default=1000)
calibration_default_step = models.FloatField(_("Pas de calibration"), help_text=_("Pas de déplacement pour la calibration en mm"), default=1.0)
-
+ calibration_default_duration = models.FloatField(_("Duruée calibration"), help_text=_("Durée de pose entre chaque puits en s"), default=3.0)
active = models.BooleanField(_("Actif"), default=False)
class Meta:
@@ -80,15 +82,19 @@ class Well(models.Model):
def __str__(self):
return f'{self.name}'
+
class MultiWell(models.Model):
label = models.CharField(_("Label"), help_text=_("Label du multi-puit"), max_length=100, null=True, blank=True)
author = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="Auteur", null=True, blank=True)
position = models.CharField(_("Position"), help_text=_('Position du multi-puits sur la table'), unique=True, max_length=8, choices=MULTIWELL_POSITION, null=True, blank=False)
+ default = models.BooleanField(_("Par défaut"), help_text=_('Multi-puit par défaut'), default=False)
cols = models.PositiveSmallIntegerField(_("Colonnes"), help_text=_('Nombre de colonnes'), blank=False, default=6)
- rows = models.PositiveSmallIntegerField(_("Lignes"), help_text=_('Nombre de lignes'), blank=False, default=4)
+ rows = models.PositiveSmallIntegerField(_("Lignes"), help_text=_('Nombre de lignes'), blank=False, default=4)
+ diameter = models.FloatField(_("Diamètre"), help_text=_('Diamètre des tubes en mm'), blank=False, default=16.0)
+
row_def = models.CharField(_("Définition"), help_text=_('Définition des lignes'), max_length=16, null=True, blank=False, default="A,B,C,D")
- row_order = models.CharField(_("Ordre"), help_text=_('Ordre de lecture en serpentin'), max_length=16, null=True, blank=False, default="D,C,B,A")
+ row_order = models.CharField(_("Ordre ligne"), help_text=_('Ordre ligne de puit. Lecture en serpentin dans le sens des +- X'), max_length=16, null=True, blank=False, default="D,C,B,A")
order = models.PositiveSmallIntegerField(_("Ordre"), help_text=_('Ordre de lecture du multi-puit'), blank=False, default=0)
duration = models.PositiveIntegerField(_("Durée"), help_text=_('Durée du film en secondes'), blank=False, default=120)
@@ -98,10 +104,14 @@ class MultiWell(models.Model):
dx = models.FloatField(_("Pas X"), help_text=_('Pas ou interval sur X en mm'), blank=False, default=19.5)
dy = models.FloatField(_("Pas Y"), help_text=_('Pas ou interval sur Y en mm'), blank=False, default=19.5)
feed = models.PositiveIntegerField(_("Vitesse"), help_text=_('Vitesse déplacement en mm/mn '), blank=False, default=1000)
+
+ well_position = models.BooleanField(_("Positions"), help_text=_('Positions des puits générées ?'), default=False)
active = models.BooleanField(_("Active"), default=True)
+
def config(self):
return dict(
+ position=self.position,
cols=self.cols,
rows=self.rows,
row_def=self.row_def,
@@ -139,6 +149,56 @@ class MultiWell(models.Model):
def __str__(self):
return f'{self.position}: {self.label}'
+
+class WellPostion(models.Model):
+ author = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="Auteur", null=True, blank=True)
+ well = models.ForeignKey(Well, verbose_name=_("Puit"), on_delete=models.SET_NULL, null=True, blank=True)
+ multiwell = models.ForeignKey(MultiWell, verbose_name=_("Multi-puits"), on_delete=models.SET_NULL, null=True, blank=True)
+
+ order = models.PositiveSmallIntegerField(_("Ordre"), help_text=_('Ordre de lecture du puit'), blank=False, default=0)
+ x = models.FloatField(_("X"), help_text=_('Axe X en mm'), blank=False, default=10.0)
+ y = models.FloatField(_("Y"), help_text=_('Axe Y en mm'), blank=False, default=10.0)
+
+
+ class Meta:
+ ordering = ['order']
+ unique_together = ["multiwell", "well"]
+ verbose_name = _("Position d'un puit")
+ verbose_name_plural = _("Position des puits")
+
+ def __str__(self):
+ return f'{self.multiwell.position}: {self.well.name}'
+
+
+@receiver(post_save, sender=MultiWell)
+def create_well_position(sender, instance, created, **kwargs):
+ if not instance.well_position:
+ row_order = instance.row_order.split(',')
+ n = 0
+ for row in range(instance.rows):
+ if row % 2 == 0:
+ cols = range(instance.cols)
+ else:
+ cols = range(instance.cols - 1, -1, -1)
+ for col in cols:
+ x = instance.xbase + col * instance.dx
+ y = instance.ybase + row * instance.dy
+ try:
+ name = f'{row_order[row]}{col+1}'
+ well = Well.objects.get(name__exact=name)
+ WellPostion.objects.update_or_create(
+ multiwell=instance,
+ well=well,
+ author=instance.author,
+ defaults={'order': n, 'x': round(x, 4), 'y': round(y, 4)}
+ )
+ n += 1
+ except:
+ pass
+ instance.well_position=True
+ instance.save()
+
+
class Observation(models.Model):
title = models.CharField(_("Titre de l'observation"), max_length=100, null=True, blank=False)
comment = models.TextField(_("Commentaires"), help_text=_("Descriptions de l'observations"), null=True, blank=True)
diff --git a/test_tube_scanner/scanner/multiwell.py b/test_tube_scanner/scanner/multiwell.py
new file mode 100644
index 0000000..f67a87f
--- /dev/null
+++ b/test_tube_scanner/scanner/multiwell.py
@@ -0,0 +1,321 @@
+'''
+Created on 20 avr. 2026
+
+@author: denis
+'''
+import logging
+import time
+from django.utils.translation import gettext_lazy as _
+from threading import Thread, Event
+from django.utils import timezone
+from django.utils.html import mark_safe
+from modules import grbl
+from . import models
+
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class WellIterator:
+ """Itérateur personnalisé pour naviguer dans les Wells"""
+
+ def __init__(self, wells_queryset):
+ self.wells = list(wells_queryset) # Convertir en liste
+ self.current_index = -1
+ self.total_count = len(self.wells)
+
+ def __iter__(self):
+ """Permet d'utiliser l'itérateur dans une boucle for"""
+ return self
+
+ def __next__(self):
+ """Retourne l'élément suivant"""
+ self.current_index += 1
+ if self.current_index >= self.total_count:
+ raise StopIteration
+ return self.wells[self.current_index]
+
+ def next(self):
+ """Méthode next() pour avancer manuellement"""
+ if self.current_index + 1 < self.total_count:
+ self.current_index += 1
+ return self.wells[self.current_index]
+ raise StopIteration("Fin de la liste atteinte")
+
+ def previous(self):
+ """Méthode previous() pour revenir en arrière"""
+ if self.current_index > 0:
+ self.current_index -= 1
+ return self.wells[self.current_index]
+ raise StopIteration("Début de la liste atteint")
+
+ def seek(self, index):
+ """Méthode seek() pour sauter à un index spécifique"""
+ if 0 <= index < self.total_count:
+ self.current_index = index
+ return self.wells[index]
+ raise IndexError(f"Index {index} hors limites (0-{self.total_count - 1})")
+
+ def get_current(self):
+ """Retourne l'élément courant"""
+ if -1 < self.current_index < self.total_count:
+ return self.wells[self.current_index]
+ return None
+
+ def reset(self):
+ """Réinitialise l'itérateur au début"""
+ self.current_index = -1
+
+
+class MultiWellManager:
+
+ def __init__(self, process):
+ self.process = process
+ self.cnc_controller = process.grbl
+ self.stop_playing = Event()
+ self.well_iterator = None
+
+ self.scanner = None
+
+ self.multiwel = None
+ self.set_default_values()
+ self.set_multiwell()
+
+
+ def set_default_values(self, feed=None, step=None, duration=None):
+ self._feed = feed or self.process.conf.calibration_default_feed
+ self._step = step or self.process.conf.calibration_default_step
+ self._duration = duration or self.process.conf.calibration_default_duration
+
+
+ def set_multiwell(self, position=None):
+ if position is None:
+ self.multiwell = models.MultiWell.objects.filter(default=True).first()
+ else:
+ self.multiwell = models.MultiWell.by_position(position)
+
+ wells = models.WellPostion.objects.filter(multiwell_id=self.multiwell.id).order_by('order').all()
+ self.well_iterator = WellIterator(wells)
+
+ self.position = self.multiwell.position
+ self._xbase = self.multiwell.xbase
+ self._ybase = self.multiwell.ybase
+ self._dx = self.multiwell.dx
+ self._dy = self.multiwell.dy
+ return self.multiwell.config()
+
+
+ def multiwell_buttons(self):
+ multiwells = []
+ multiwells.append('''''')
+ for w in self.well_iterator:
+ multiwells.append(f"""""")
+ multiwells.append('''
''')
+ self.well_iterator.reset()
+ return mark_safe("\n".join(multiwells))
+
+
+ def _grid_scanning_capture(self, uuid, duration):
+ self.process.data.uuid = uuid
+ self.process.data.record = True
+
+ start = time.monotonic()
+ while not self.stop_playing.is_set():
+ if time.monotonic() - start > duration:
+ break
+ self.cnc_controller.wait_for(1.0)
+
+ logger.info(f"Arrêter l'enregistrement {uuid}")
+ self.process.data.record = False
+ self.process.data.uuid = None
+
+
+ def _grid_scanning(self, observation, xnext=0, ynext=0):
+ multiwell = observation.multiwell
+ wells = models.WellPostion.objects.filter(multiwell_id=multiwell.id).order_by('order').all()
+
+ self.stop_playing = Event()
+ for w in wells:
+ if self.stop_playing.is_set():
+ break
+ self.cnc_controller.move_to(w.x, w.y, feed=w.multiwell.feed)
+
+ uuid = f'{self.process.data.session}-{multiwell.position}-{w.well.name}'
+ self._grid_scanning_capture(uuid, multiwell.duration)
+
+ logger.info(f"Scan terminé — retour à l'origine (X={xnext:.1f} Y={ynext:.1f})")
+ self.cnc_controller.move_to(xnext, ynext, feed=multiwell.feed*2)
+
+
+ def _start_scanning(self, session, observations):
+ xynext = []
+ for obs in observations:
+ xynext.append((obs.multiwell.xbase, obs.multiwell.ybase))
+ xynext.append((0, 0))
+
+ pos = 1
+ self.process.data.session = session.id
+ started = timezone.now()
+ for obs in observations:
+ obs.started = timezone.now()
+ obs.save()
+
+ xnext, ynext = xynext[pos]
+ pos +=1
+ self._grid_scanning(obs, xnext=xnext, ynext=ynext)
+
+ obs.finished = timezone.now()
+ obs.save()
+ session.finished = timezone.now()
+ session.active = False
+ session.scanning_task.enabled = False
+ session.save()
+ logger.info(f"==== Session {session.name} terminée à {session.finished} après {session.finished - started} secondes.")
+
+
+ def halt_scanning(self):
+ self.process.data.record = False
+ return self.stop_playing.set()
+
+
+ def scanning(self, sid):
+ try:
+ session = models.Session.objects.get(pk=sid)
+ observations = models.SessionObservation.observation_by_session(sid)
+ Thread(target=self._start_scanning, args=(session, observations, ), daemon=True).start()
+ except Exception as e:
+ print("MultiWellManager::scan error", e)
+
+
+ def previous_well(self):
+ w = self.well_iterator.previous()
+ self.cnc_controller.move_to(w.x, w.y, feed=w.multiwell.feed)
+ return {"state": "previous", "msg": f">>> ({w.x}, {w.y})"}
+
+
+ def next_well(self):
+ w = self.well_iterator.next()
+ self.cnc_controller.move_to(w.x, w.y, feed=w.multiwell.feed)
+ return {"state": "next", "msg": f">>> ({w.x}, {w.y})"}
+
+
+ def goto_well(self, numwell):
+ w = self.well_iterator.seek(numwell)
+ self.cnc_controller.move_to(w.x, w.y, feed=w.multiwell.feed)
+ return {"state": "goto", "msg": f">>> ({w.x}, {w.y})"}
+
+
+ def set_well_position(self):
+ w = self.well_iterator.get_current()
+ w.x, w.y = self.cnc_controller.get_mpos()
+ w.save()
+ return {"state": "well_position", "msg": f">>> saved ({w.x}, {w.y})"}
+
+
+ def _scanning_test(self, xnext=0, ynext=0):
+ self.stop_playing = Event()
+ for w in self.well_iterator:
+ if self.stop_playing.is_set():
+ break
+ self.cnc_controller.move_to(w.x, w.y, feed=w.multiwell.feed)
+
+ start = time.monotonic()
+ while not self.stop_playing.is_set():
+ if time.monotonic() - start > self.duration:
+ break
+ self.cnc_controller.wait_for(1.0)
+
+ logger.info(f"Arrêter la simulation")
+
+ self.well_iterator.reset()
+
+ logger.info(f"Scan terminé — retour à l'origine (X={xnext:.1f} Y={ynext:.1f})")
+ self.cnc_controller.move_to(xnext, ynext, feed=self.multiwell.feed*2)
+
+
+ def scan_test(self):
+ Thread(target=self._scanning_test, daemon=True).start()
+
+
+ @property
+ def position(self):
+ return self._position
+
+ @position.setter
+ def position(self, value):
+ self._position = value
+
+ @property
+ def duration(self):
+ return self._duration
+
+ @duration.setter
+ def duration(self, value):
+ self._duration = value
+
+ @property
+ def step(self):
+ return self._step
+
+ @step.setter
+ def step(self, value):
+ self._step = value
+
+ @property
+ def feed(self):
+ return self._feed
+
+ @feed.setter
+ def feed(self, value):
+ self._feed = value
+
+ @property
+ def xbase(self):
+ return self._xbase
+
+ @xbase.setter
+ def xbase(self, value):
+ self._xbase = value
+
+ @property
+ def ybase(self):
+ return self._ybase
+
+ @ybase.setter
+ def ybase(self, value):
+ self._ybase = value
+
+ @property
+ def dx(self):
+ return self._dx
+
+ @dx.setter
+ def dx(self, value):
+ self._dx = value
+
+ @property
+ def dy(self):
+ return self._dy
+
+ @dy.setter
+ def dy(self, value):
+ self._dy = value
+
+
+ def set_xy_step(self):
+ models.MultiWell.objects.filter(position__exact=self.position).update(dx=self.dx, dy=self.dy)
+
+
+ def set_position(self):
+ x, y = self.cnc_controller.get_mpos()
+ self.cnc_controller.wait_for(2.0)
+ models.MultiWell.objects.filter(position__exact=self.position).update(xbase=x, ybase=y)
+ self._xbase, self._ybase = x, y
+
+
+ def calib_toggle_debug(self):
+ """ Active / désactive le mode debug sur le stream."""
+ aligner = self.process.cam._aligner
+ aligner.debug = not aligner.debug
+ return {"state": "debug", "msg": f"Debug: {aligner.debug}"}
diff --git a/test_tube_scanner/scanner/process.py b/test_tube_scanner/scanner/process.py
index 57e2cf7..3ecc933 100644
--- a/test_tube_scanner/scanner/process.py
+++ b/test_tube_scanner/scanner/process.py
@@ -4,6 +4,8 @@ import os
os.environ['OPENCV_LOG_LEVEL']="0"
os.environ['OPENCV_FFMPEG_LOGLEVEL']="0"
import cv2
+import numpy as np
+
from django.utils.translation import gettext_lazy as _
from datetime import datetime
import time, asyncio, bisect
@@ -25,14 +27,16 @@ from modules import reductstore, grbl, utils
## camera devices
from modules.circular_crop import CircularCrop, CropStrategy
+from .multiwell import MultiWellManager
from . import models
@dataclass
-class ProcTag:
+class ProcessData:
play: bool = True
record: bool = False
uuid: str = None
session: int = 0
+ frame: bytes = None
logger = get_task_logger(__name__)
redisDB = Redis(host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=0, decode_responses=True)
@@ -132,159 +136,12 @@ class CameraRecordManager():
asyncio.run(self.remove_uuid(uuid, start, stop, when=when))
-class MultiWellManager:
-
- def __init__(self, position, feed=None, step=None, process=None):
- self.set_multiwell(position)
- self._feed = feed
- self._step = step
- self.process = process
- self.tag = process.tag
- self.scanner = None
-
- def set_multiwell(self, position):
- self._position = position
- self.well = models.MultiWell.by_position(position)
- self._xbase = self.well.xbase
- self._ybase = self.well.ybase
- self._dx = self.well.dx
- self._dy = self.well.dy
-
- def _start_test(self):
- self.scanner.start()
-
- def _start(self, machine, session, observations):
- xynext = []
- for obs in observations:
- xynext.append((obs.multiwell.xbase, obs.multiwell.ybase))
- xynext.append((0, 0))
-
- pos = 1
- self.tag.session = session.id
- started = timezone.now()
- for obs in observations:
- conf = obs.multiwell.config()
- self.scanner = grbl.GridScanner(machine, process=self.process, **conf)
- obs.started = timezone.now()
- obs.save()
-
- xnext, ynext = xynext[pos]
- pos +=1
- self.scanner.start(xnext=xnext, ynext=ynext, position=obs.multiwell.position)
-
- obs.finished = timezone.now()
- obs.save()
- session.finished = timezone.now()
- session.active = False
- session.scanning_task.enabled = False
- session.save()
- logger.info(f"==== Session {session.name} terminée à {session.finished} après {session.finished - started} secondes.")
-
-
- def scan_test(self, machine, duration=5.0):
- conf = self.well.config()
- conf['duration'] = duration
- conf['feed'] = self.feed
- conf['xnext'] = self._xbase
- conf['ynext'] = self._ybase
-
- self.tag.session = 0
- self.scanner = grbl.GridScanner(machine, process=self.process, **conf)
- Thread(target=self._start_test, daemon=True).start()
-
- def scan(self, machine, sid):
- try:
- session = models.Session.objects.get(pk=sid)
- observations = models.SessionObservation.observation_by_session(sid)
- Thread(target=self._start, args=(machine, session, observations, ), daemon=True).start()
- except Exception as e:
- print("MultiWellManager::scan error", e)
-
- def halt(self):
- if self.scanner:
- self.scanner.halt()
-
- @property
- def position(self):
- return self._position
-
- @position.setter
- def position(self, value):
- self._position = value
-
- @property
- def step(self):
- return self._step
-
- @step.setter
- def step(self, value):
- self._step = value
-
- @property
- def feed(self):
- return self._feed
-
- @feed.setter
- def feed(self, value):
- self._feed = value
-
- @property
- def xbase(self):
- return self._xbase
-
- @xbase.setter
- def xbase(self, value):
- self._xbase = value
-
- @property
- def ybase(self):
- return self._ybase
-
- @ybase.setter
- def ybase(self, value):
- self._ybase = value
-
- @property
- def dx(self):
- return self._dx
-
- @dx.setter
- def dx(self, value):
- self._dx = value
-
- @property
- def dy(self):
- return self._dy
-
- @dy.setter
- def dy(self, value):
- self._dy = value
-
- def set_xy_step(self):
- models.MultiWell.objects.filter(position__exact=self.position).update(dx=self.dx, dy=self.dy)
-
- def set_position(self, machine):
- x, y = machine.get_mpos()
- machine.wait_for(2.0)
- models.MultiWell.objects.filter(position__exact=self.position).update(xbase=x, ybase=y)
- self._xbase, self._ybase = x, y
-
class ScannerProcess(Task):
- '''
- video_quality = settings.VIDEO_JPG_QUALITY
- image_quality = settings.IMAGE_JPG_QUALITY
- video_fps = settings.VIDEO_FPS
- video_width = settings.VIDEO_WIDTH
- video_height = settings.VIDEO_HEIGHT
- crop_radius = settings.CALIBRATION_CROP_RADIUS
- default_multiwell = settings.CALIBRATION_DEFAULT_MULTIWELL
- default_feed = settings.CALIBRATION_DEFAULT_FEED
- default_step = settings.CALIBRATION_DEFAULT_STEP'''
-
- def __init__(self):
+ def __init__(self, use_tracking=False):
super().__init__()
+ self.use_tracking = use_tracking
self.channel_layer = get_channel_layer()
self.group = f'scanner_proc'
self.stop_event = Event()
@@ -294,7 +151,7 @@ class ScannerProcess(Task):
self.multiwel = None
self.conf = None
self.record_queue = Queue()
- self.tag = ProcTag()
+ self.data = ProcessData()
self.manager = None
self.recordDB = CameraRecordManager(cameraDB)
@@ -313,21 +170,37 @@ class ScannerProcess(Task):
self.video_fps = self.conf.video_frame_rate
self.video_width = self.conf.video_width_capture
self.video_height = self.conf.video_height_capture
-
self.crop_radius = self.conf.calibration_crop_radius
- self.default_multiwell = self.conf.calibration_default_multiwell
- self.default_feed = self.conf.calibration_default_feed
- self.default_step = self.conf.calibration_default_step
-
+
self.video_jpg_quality = [int(cv2.IMWRITE_JPEG_QUALITY), self.video_quality]
self.image_jpg_quality = [int(cv2.IMWRITE_JPEG_QUALITY), self.image_quality]
self.grbl_xmax = self.conf.grbl_xmax
self.grbl_ymax = self.conf.grbl_ymax
- #self.crop = CircularCrop(radius=self.crop_radius, strategy=CropStrategy.CROP_JPEG, jpeg_quality=self.image_quality)
self.crop = self.set_crop_radius(self.crop_radius)
- '''
- if not self.conf.use_rpicam:
+ if settings.TEST_VIDEOFILE:
+ from modules.videofile_capture import VideoFileCapture
+ self.cam = VideoFileCapture(
+ video_file=settings.MEDIA_ROOT / 'simulation' / 'part4-5fps.mp4',
+ fps=self.video_fps,
+ width=self.video_width,
+ height=self.video_height,
+ jpeg_quality=self.video_quality,
+ use_tracking=self.use_tracking,
+ px_per_mm = self.conf.px_per_mm,
+ display=self._display,
+ video_lists=[],
+ )
+ '''
+ settings.MEDIA_ROOT / 'simulation' / 'part1-5fps.mp4',
+ settings.MEDIA_ROOT / 'simulation' / 'part2-5fps.mp4',
+ settings.MEDIA_ROOT / 'simulation' / 'part3-5fps.mp4',
+ settings.MEDIA_ROOT / 'simulation' / 'part4-5fps.mp4',
+ settings.MEDIA_ROOT / 'simulation' / 'part5-5fps.mp4',
+ ]
+ )
+ '''
+ elif not self.conf.use_rpicam:
from modules.webcam_capture import WebcamCapture
self.cam = WebcamCapture(
device_index=self.conf.webcam_device_index,
@@ -335,6 +208,9 @@ class ScannerProcess(Task):
width=self.video_width,
height=self.video_height,
jpeg_quality=self.video_quality,
+ use_tracking=self.use_tracking,
+ px_per_mm = self.conf.px_per_mm,
+ display=self._display,
)
else:
from modules.picamera2_capture import PiCamera2Capture
@@ -343,27 +219,15 @@ class ScannerProcess(Task):
width=self.video_width,
height=self.video_height,
jpeg_quality=self.video_quality,
+ use_tracking=self.use_tracking,
+ px_per_mm = self.conf.px_per_mm,
+ display=self._display,
)
- '''
- from modules.videofile_capture import VideoFileCapture
- self.cam = VideoFileCapture(
- video_file=settings.MEDIA_ROOT / 'simulation' / 'part2-5fps.mp4',
- fps=self.video_fps,
- width=self.video_width,
- height=self.video_height,
- jpeg_quality=self.video_quality,
- video_lists = [
- settings.MEDIA_ROOT / 'simulation' / 'part1-5fps.mp4',
- settings.MEDIA_ROOT / 'simulation' / 'part2-5fps.mp4',
- settings.MEDIA_ROOT / 'simulation' / 'part3-5fps.mp4',
- settings.MEDIA_ROOT / 'simulation' / 'part4-5fps.mp4',
- settings.MEDIA_ROOT / 'simulation' / 'part5-5fps.mp4',
- ]
- )
self.cam.set_frame_callback(self._on_frame)
- self.cam.set_median(False)
+ self.cam._active_median = False
self.cam.set_circular_crop(None)
+
self.stop_event.clear()
self.start_services()
except Exception as e:
@@ -399,10 +263,11 @@ class ScannerProcess(Task):
self._send(**msg)
def _on_frame(self, jpeg_bytes: bytes, ts: datetime, metrics: dict) -> None:
- if self.tag.record:
+ self.data.frame = jpeg_bytes
+ if self.data.record:
# record images
- self.record_queue.put((self.tag.uuid, ts, jpeg_bytes, metrics))
- if self.tag.play:
+ self.record_queue.put((self.data.uuid, ts, jpeg_bytes, metrics))
+ if self.data.play:
# play image
self._send(ts=ts.timestamp(), jpeg=base64.b64encode(jpeg_bytes).decode(), **metrics)
@@ -411,7 +276,7 @@ class ScannerProcess(Task):
while not self.stop_event.is_set():
try:
(uuid, ts, frame, metrics) = self.record_queue.get()
- labels = dict(fps=self.video_fps, session=self.tag.session, detected="1" if metrics.get("detected") else "0")
+ labels = dict(fps=self.video_fps, session=self.data.session, detected="1" if metrics.get("detected") else "0")
if metrics.get("detected"):
labels.update({
"cx" : str(metrics["cx"]),
@@ -441,13 +306,7 @@ class ScannerProcess(Task):
pubsub = redisDB.pubsub()
pubsub.subscribe(self.group)
self._init_grbl()
-
- self.manager = MultiWellManager(
- self.default_multiwell,
- feed=self.default_feed,
- step=self.default_step,
- process=self
- )
+ self.manager = MultiWellManager(process=self)
for message in pubsub.listen():
try:
@@ -466,7 +325,7 @@ class ScannerProcess(Task):
topic = cmd.get("topic")
if topic == 'init':
self.cam.set_circular_crop(self.crop)
- self.cam.set_median(is_median=False)
+ self.cam._active_median = False
self.grbl.go_origin(feed=self.manager.feed)
elif topic == 'scan':
@@ -474,65 +333,129 @@ class ScannerProcess(Task):
if sid == "0":
self._send(state='error', msg=str(_('La session est nulle!...')))
else:
- self.cam.set_median(is_median=False)
- self.manager.scan(self.grbl, sid)
+ self.cam._active_median = False
+ self.manager.scanning(sid)
elif cmd["type"]=="calibrate":
topic = cmd.get("topic")
value = cmd.get("value")
+ buttons = None
+ if topic == 'init':
+ self.manager.set_default_values(
+ feed=int(cmd.get("feed")),
+ step=float(cmd.get("step")),
+ duration=float(cmd.get("duration"))
+ )
+ position = cmd.get("position")
+ self.manager.set_multiwell(position)
+ self.cam.set_circular_crop(None)
+ self.cam._active_median = False
+ buttons = self.manager.multiwell_buttons()
- if topic == 'init':
- self.manager.feed = int(cmd.get("feed", self.default_feed))
- self.manager.step = float(cmd.get("step", self.default_step))
- position = cmd.get("position", self.default_multiwell)
- if self.manager.position != position:
- self.manager.set_multiwell(position)
- self.cam.set_circular_crop(None)
- self.cam.set_median(is_median=False)
elif topic == 'up':
self.grbl.move_relative(dy=self.manager.step, feed=self.manager.feed)
+
elif topic == 'down':
self.grbl.move_relative(dy=-self.manager.step, feed=self.manager.feed)
+
elif topic == 'right':
self.grbl.move_relative(dx=self.manager.step, feed=self.manager.feed)
+
elif topic == 'left':
self.grbl.move_relative(dx=-self.manager.step, feed=self.manager.feed)
+
elif topic == 'median':
- self.cam.set_median(is_median=value)
- elif topic == 'crop':
- self.cam.set_circular_crop(self.crop) if value else self.cam.set_circular_crop(None)
+ self.cam._active_median = not self.cam._active_median
continue
+
+ elif topic == 'crop':
+ self.cam._active_crop = not self.cam._active_crop
+ self.cam.set_circular_crop(self.crop) if self.cam._active_crop else self.cam.set_circular_crop(None)
+ continue
+
elif topic == 'crop_radius':
self.conf.calibration_crop_radius=int(value)
self.crop = self.set_crop_radius(self.conf.calibration_crop_radius)
self.conf.save()
self.cam.set_circular_crop(self.crop)
- continue
+ continue
+
elif topic == 'position':
self.manager.set_multiwell(value)
+ buttons = self.manager.multiwell_buttons()
+
elif topic == 'step':
self.manager.step = float(value)
+
elif topic == 'feed':
self.manager.feed = int(value)
+
+ elif topic == 'duration':
+ self.manager.duration = float(value)
+
elif topic == 'goto_0':
self.grbl.go_origin(feed=self.manager.feed)
+ self.manager.well_iterator.reset()
+
elif topic == 'goto_xy':
self.grbl.move_to(self.manager.xbase, self.manager.ybase, feed=self.manager.feed)
+ self.manager.well_iterator.reset()
+
elif topic == 'xy_base':
- self.manager.set_position(self.grbl)
+ self.manager.set_position()
+
elif topic == 'dx':
self.manager.dx = float(value)
+
elif topic == 'dy':
self.manager.dy = float(value)
+
elif topic == 'xy_step':
self.manager.set_xy_step()
+
elif topic == 'test':
- self.manager.scan_test(self.grbl)
+ pass
+ #self.manager.scan_test()
continue
+ elif topic == 'center':
+ #self.manager.scan_test()
+ dx_mm = self.cam._last_detection["offset_x_mm"]
+ dy_mm = self.cam._last_detection["offset_y_mm"]
+ self.grbl.move_to(self.grbl.x + dx_mm, self.grbl.y + dy_mm, feed=150)
+
+ msg = f"Correction CNC move_relative(dx={dx_mm:.3f}, dy={dy_mm:.3f})"
+ self._send(state='center', msg=msg)
+ continue
+
elif topic == 'halt':
- self.manager.halt()
+ self.manager.halt_scanning()
continue
-
+
+ elif topic == 'calib_debug':
+ msg = self.manager.calib_toggle_debug()
+ self._send(**msg)
+ continue
+
+ elif topic == 'previous':
+ msg = self.manager.previous_well()
+ self._send(**msg)
+ continue
+
+ elif topic == 'next':
+ msg = self.manager.next_well()
+ self._send(**msg)
+ continue
+
+ elif topic == 'goto':
+ msg = self.manager.goto_well(int(value))
+ self._send(**msg)
+ continue
+
+ elif topic == 'set_well':
+ msg = self.manager.set_well_position()
+ self._send(**msg)
+ continue
+
self._send(
xbase=self.manager.xbase,
ybase=self.manager.ybase,
@@ -541,7 +464,8 @@ class ScannerProcess(Task):
xy=True,
dxy=True,
dx=self.manager.dx,
- dy=self.manager.dy
+ dy=self.manager.dy,
+ buttons=buttons,
)
except Exception as e:
diff --git a/test_tube_scanner/scanner/static/scanner/css/calibration.css b/test_tube_scanner/scanner/static/scanner/css/calibration.css
index 5c73d51..1cdd922 100644
--- a/test_tube_scanner/scanner/static/scanner/css/calibration.css
+++ b/test_tube_scanner/scanner/static/scanner/css/calibration.css
@@ -32,3 +32,17 @@
align-self: start;
grid-area: move;
}
+
+.well {
+ padding: 0.2em;
+}
+
+.well-btn {
+ display: grid;
+ grid-template-columns: repeat(6, 1fr);
+ justify-items: center;
+ align-items: center;
+}
+
+
+
diff --git a/test_tube_scanner/scanner/static/scanner/js/calibration.js b/test_tube_scanner/scanner/static/scanner/js/calibration.js
index d561558..1729647 100644
--- a/test_tube_scanner/scanner/static/scanner/js/calibration.js
+++ b/test_tube_scanner/scanner/static/scanner/js/calibration.js
@@ -9,17 +9,14 @@ class ScannerManager {
this.debug_count = 0
}
- toggle_median() { this.axes = !this.axes; return this.axes; }
- toggle_crop() { this.croping = !this.croping; return this.croping; }
-
init_controls() {
- this.ts = sId("_ts");
- this.cx = sId("_cx");
- this.cy = sId("_cy");
- this.speed_px_s = sId("_speed_px_s");
+ this.ts = sId("_ts");
+ this.cx = sId("_cx");
+ this.cy = sId("_cy");
+ this.speed_px_s = sId("_speed_px_s");
this.axial_speed = sId("_axial_speed");
- this.axial_pos = sId("_axial_pos");
- this.area_px = sId("_area_px");
+ this.axial_pos = sId("_axial_pos");
+ this.area_px = sId("_area_px");
this.frame_count = sId("_count");
const goto_0 = sId("_goto-0");
@@ -30,6 +27,7 @@ class ScannerManager {
const down = sId("_down");
const left = sId("_left");
const right = sId("_right");
+ this.duration = sId("_duration");
this.feed = sId("_feed");
this.step = sId("_step");
this.well = sId("_well");
@@ -40,34 +38,50 @@ class ScannerManager {
this.xbase = sId("_xbase");
this.ybase = sId("_ybase");
this.debug = sId("_debug");
+ this.well_btn = sId("_well_btn");
+
const test = sId("_test");
const halt = sId("_halt");
+ const calib_debug = sId("_calib_debug");
+ const calib_center = sId("_calib_center");
+ const previous = sId("_previous");
+ const next = sId("_next");
+ const set_well = sId("_set_well");
+
const median = sId("_median");
const crop = sId("_crop");
const crop_radius = sId("_crop_radius");
+
up.addEventListener('mousedown', (e) => { this._send({ type: 'calibrate', topic: "up" }); });
down.addEventListener('mousedown', (e) => { this._send({ type: 'calibrate', topic: "down" }); });
left.addEventListener('mousedown', (e) => { this._send({ type: 'calibrate', topic: "left" }); });
right.addEventListener('mousedown', (e) => { this._send({ type: 'calibrate', topic: "right" }); });
- goto_0.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "goto_0" }); });
- goto_xy.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "goto_xy" }); });
+ goto_0.addEventListener('click', (e) => { this.clear_buttons(); this._send({ type: 'calibrate', topic: "goto_0" }); });
+ goto_xy.addEventListener('click', (e) => { this.clear_buttons(); this._send({ type: 'calibrate', topic: "goto_xy" }); });
xy_base.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "xy_base" }); });
xy_step.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "xy_step" }); });
-
- median.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "median", value: this.toggle_median() }); });
- crop.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "crop", value: this.toggle_crop() }); });
+
+ calib_debug.addEventListener('click',(e) => { this._send({ type: 'calibrate', topic: "calib_debug" }); });
+ previous.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "previous" }); });
+ next.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "next" }); });
+ set_well.addEventListener('click',(e) => { this._send({ type: 'calibrate', topic: "set_well" }); });
+
+ median.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "median" }); });
+ crop.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "crop" }); });
crop_radius.addEventListener('change',(e) => { this._send({ type: 'calibrate', topic: "crop_radius", value: crop_radius.value }); });
this.well.addEventListener("change", (e) => { this._send({ type: 'calibrate', topic: "position", value: e.target.value }); });
this.step.addEventListener("change", (e) => { this._send({ type: 'calibrate', topic: "step", value: e.target.value }); });
this.feed.addEventListener("change", (e) => { this._send({ type: 'calibrate', topic: "feed", value: e.target.value }); });
-
+ this.duration.addEventListener("change", (e) => { this._send({ type: 'calibrate', topic: "duration", value: e.target.value }); });
+
this.dx.addEventListener("change", (e) => { this._send({ type: 'calibrate', topic: "dx", value: e.target.value }); });
this.dy.addEventListener("change", (e) => { this._send({ type: 'calibrate', topic: "dy", value: e.target.value }); });
test.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "test" }); });
+ calib_center.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "center" }); });
halt.addEventListener('click', (e) => { this._send({ type: 'calibrate', topic: "halt" }); });
}
@@ -85,7 +99,7 @@ class ScannerManager {
if (payload.state) { this.debug.insertAdjacentHTML('afterbegin', `[ ${++this.debug_count} - ${payload.state} ]: ${payload.msg}`); }
if (payload.ts) { this.ts.textContent = timestampToLocalISOString(payload.ts); }
- if (payload.detected) {
+ if (payload.detected && use_tracking) {
this.cx.textContent = payload.cx; this.cy.textContent = payload.cy;
this.speed_px_s.textContent = payload.speed_px_s;
this.axial_speed.textContent = payload.axial_speed;
@@ -93,19 +107,24 @@ class ScannerManager {
this.area_px.textContent = payload.area_px;
this.frame_count.textContent = payload.count;
}
+ if (payload.buttons) { this.well_btn.innerHTML = payload.buttons; }
} catch(e) { console.log(e); }
}
-
+
+ clear_buttons() { document.querySelectorAll('button.w3-button.well').forEach(btn => {btn.classList.remove('w3-green'); }); }
+ goto_well(b) { this.clear_buttons(); b.classList.add('w3-green'); this._send({ type: 'calibrate', topic: "goto", value: b.value }); }
init() {
this.axes = 0;
this.cropping = 0;
+ this.clear_buttons();
this._send({
type: 'calibrate',
topic: "init",
feed: this.feed.value,
step: this.step.value,
- position: this.well.value
+ position: this.well.value,
+ duration: this.duration.value
});
}
start() { this._send({ type: 'scanner', topic: "start"}); }
diff --git a/test_tube_scanner/scanner/static/scanner/js/main.js b/test_tube_scanner/scanner/static/scanner/js/main.js
index b5e4fc2..4c39c19 100644
--- a/test_tube_scanner/scanner/static/scanner/js/main.js
+++ b/test_tube_scanner/scanner/static/scanner/js/main.js
@@ -48,7 +48,7 @@ class ScannerManager {
if (payload.xy) { this.x.textContent=payload.x.toFixed(2); this.y.textContent=payload.y.toFixed(2); }
if (payload.state) { this.debug.insertAdjacentHTML('afterbegin', `[ ${++this.debug_count} - ${payload.state} ]: ${payload.msg}`); }
if (payload.ts) { this.ts.textContent = timestampToLocalISOString(payload.ts); }
- if (payload.detected) {
+ if (payload.detected && use_tracking) {
this.cx.textContent = payload.cx; this.cy.textContent = payload.cy;
this.speed_px_s.textContent = payload.speed_px_s;
this.axial_speed.textContent = payload.axial_speed;
diff --git a/test_tube_scanner/scanner/tasks.py b/test_tube_scanner/scanner/tasks.py
index c1e3b09..bf85812 100644
--- a/test_tube_scanner/scanner/tasks.py
+++ b/test_tube_scanner/scanner/tasks.py
@@ -3,6 +3,7 @@ import asyncio
from celery import shared_task, group, chord, chain
from celery.utils.log import get_task_logger
from django.utils import timezone
+from django.conf import settings
from .process import ScannerProcess, ReplayProcess
from .export_tasks import shm_download_video, export_images_zip, export_video_mp4
@@ -18,7 +19,7 @@ class ScannerTaskManager:
def start_scanner(self):
if self.scanner is None:
- self.scanner = ScannerProcess()
+ self.scanner = ScannerProcess(use_tracking=settings.TRACKING)
self.scanner.start()
def stop_scanner(self):
diff --git a/test_tube_scanner/scanner/templates/scanner/calibration.html b/test_tube_scanner/scanner/templates/scanner/calibration.html
index 9910e73..46135e2 100644
--- a/test_tube_scanner/scanner/templates/scanner/calibration.html
+++ b/test_tube_scanner/scanner/templates/scanner/calibration.html
@@ -1,5 +1,5 @@
{% extends 'scanner/base.html' %}
-{% load i18n home_tags %}
+{% load i18n home_tags scanner_tags %}
{% block styles %}
{{ block.super }}
@@ -11,15 +11,15 @@