diff --git a/test_tube_scanner/home/settings.py b/test_tube_scanner/home/settings.py index b89a3b5..8b24530 100644 --- a/test_tube_scanner/home/settings.py +++ b/test_tube_scanner/home/settings.py @@ -391,6 +391,9 @@ EXPORT_DESTINATIONS = ["local", "remote"] TEST_VIDEOFILE = False TRACKING = False -TRACKER_TUBE_AXIS = "horizontal" #"vertical" +TRACKER_TUBE_AXIS = "vertical" TRACKER_MIN_AREA = 200 +CALIBRATION_AUTO_DURATION = 45.0 +CALIBRATION_AUTO_TIMEOUT = 2.5 + diff --git a/test_tube_scanner/modules/capture_interface.py b/test_tube_scanner/modules/capture_interface.py index a976b71..27ba9df 100644 --- a/test_tube_scanner/modules/capture_interface.py +++ b/test_tube_scanner/modules/capture_interface.py @@ -20,7 +20,7 @@ import threading import logging from datetime import datetime, timezone from pathlib import Path -from typing import Optional, Callable, TYPE_CHECKING +from typing import Optional, Callable, TYPE_CHECKING, Any from django.conf import settings from modules.planarian_tracker import PlanarianTracker @@ -78,15 +78,15 @@ class VideoCaptureInterface(abc.ABC): dead_zone_px = 5, # en-dessous → rien à faire display = display, ) - self._last_detection = None # résultat du dernier alignement - - + self.align_detection = None # résultat du test + def on_well_change(self): """ Appelé par le CNC lors du changement de puits. Réinitialise le fond appris et l'état inter-frame du tracker. """ - self._tracker.reset() + self._tracker.reset() + # ------------------------------------------------------------------ # Méthodes abstraites — obligatoires dans les sous-classes @@ -236,8 +236,8 @@ class VideoCaptureInterface(abc.ABC): # Mode debug if self._aligner.debug: - self._last_detection = self._aligner.detect_tube(frame, self.parent.data.tube_diameter or 16.0) - annotated = self._last_detection.get('frame_annotated') + self.align_detection = self._aligner.detect_tube(frame) + annotated = self.align_detection.get('frame_annotated') frame = annotated if annotated is not None else frame # mode racking diff --git a/test_tube_scanner/modules/grbl.py b/test_tube_scanner/modules/grbl.py index a4513e1..acad912 100644 --- a/test_tube_scanner/modules/grbl.py +++ b/test_tube_scanner/modules/grbl.py @@ -168,7 +168,7 @@ class GRBLController: def get_mpos(self): return self._mpos(self.get_status()) - def wait_idle(self, timeout=10): + def wait_idle(self, timeout=20): start = time.time() while True: if time.time() - start > timeout: diff --git a/test_tube_scanner/modules/tube_aligner.py b/test_tube_scanner/modules/tube_aligner.py index eb00f51..c72694a 100644 --- a/test_tube_scanner/modules/tube_aligner.py +++ b/test_tube_scanner/modules/tube_aligner.py @@ -27,14 +27,21 @@ class TubeAligner: self.grbl_threshold_px = grbl_threshold_px self.dead_zone_px = dead_zone_px self.debug = debug - self.display = display + self.display = display + self.TUBE_DIAMETER_MM = 16.0 + + + def set_tube_diameter(self, tube_diameter: float = 16.0) -> None: + self.TUBE_DIAMETER_MM = tube_diameter # ------------------------------------------------------------------ # # Détection principale # ------------------------------------------------------------------ # - def detect_tube(self, frame: np.ndarray, tube_diameter: float = 16.0) -> dict: - TUBE_DIAMETER_MM = tube_diameter + def detect_tube(self, frame: np.ndarray, tube_diameter: float = None) -> dict: + if tube_diameter is not None: + self.set_tube_diameter(tube_diameter) + h, w = frame.shape[:2] cx_img = w // 2 cy_img = h // 2 @@ -44,12 +51,13 @@ class TubeAligner: "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, + "offset_x_px" : 0, + "offset_y_px" : 0, + "offset_x_mm" : 0.0, + "offset_y_mm" : 0.0, "action" : "none", "frame_annotated": None, + "msg" : None, } frame_out = frame.copy() @@ -81,7 +89,8 @@ class TubeAligner: all_r.append(int(best[2])) if not all_cx: - logger.warning("TubeAligner: aucun cercle détecté (%dx%d)", w, h) + msg = f"TubeAligner: aucun cercle détecté ({w}x{h})" + result["msg"] =msg if self.debug: frame_out = self._draw_debug_no_detection(frame_out, cx_img, cy_img) result["frame_annotated"] = frame_out @@ -92,7 +101,7 @@ class TubeAligner: ty = int(np.mean(all_cy)) tr = int(np.mean(all_r)) if tr > 0: - self.px_per_mm = (2 * tr) / TUBE_DIAMETER_MM + self.px_per_mm = (2 * tr) / self.TUBE_DIAMETER_MM offset_x_px = tx - cx_img offset_y_px = ty - cy_img @@ -118,7 +127,8 @@ class TubeAligner: dist_px, action, votes=len(all_cx), # ← affiche le nombre de configs ayant détecté ) - + + dx_mm , dy_mm = round(offset_x_mm, 3), round(offset_y_mm, 3) result.update({ "detected" : True, "tube_cx" : tx, @@ -126,40 +136,15 @@ class TubeAligner: "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), + "offset_x_mm" : dx_mm, + "offset_y_mm" : dy_mm, "action" : action, "frame_annotated": frame_out, + "msg" : f"Correction CNC relative (dx={dx_mm:}, dy={dy_mm}), action: {action}" }) 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) - - # ------------------------------------------------------------------ # + + # ------------------------------------------------------------------ # # Dessin debug # ------------------------------------------------------------------ # diff --git a/test_tube_scanner/scanner/multiwell.py b/test_tube_scanner/scanner/multiwell.py index 96e6733..c3c36c9 100644 --- a/test_tube_scanner/scanner/multiwell.py +++ b/test_tube_scanner/scanner/multiwell.py @@ -13,10 +13,9 @@ from threading import Thread, Event #from django.utils.translation import gettext_lazy as _ from django.utils import timezone from django.utils.html import mark_safe +from django.conf import settings from . import models -CALIBRATION_AUTO_DURATION = 45.0 -CALIBRATION_AUTO_TIMEOUT = 3.0 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -113,8 +112,8 @@ class MultiWellManager: def multiwell_buttons(self): multiwells = [] multiwells.append('''
''') - for w in self.well_iterator: - multiwells.append(f"""""") + for wl in self.well_iterator: + multiwells.append(f"""""") multiwells.append('''
''') self.well_iterator.reset() return mark_safe("\n".join(multiwells)) @@ -138,14 +137,16 @@ class MultiWellManager: 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() - + cam = self.process.cam + cam._aligner.set_tube_diameter(multiwell.diameter) + self.stop_playing = Event() - for w in wells: + for wl in wells: if self.stop_playing.is_set(): break - self.cnc_controller.move_to(w.x, w.y, feed=w.multiwell.feed) + self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) - uuid = f'{self.process.data.session}-{multiwell.position}-{w.well.name}' + uuid = f'{self.process.data.session}-{multiwell.position}-{wl.well.name}' self._grid_scanning_capture(uuid, multiwell.duration) logger.info(f"Scan terminé — retour à l'origine (X={xnext:.1f} Y={ynext:.1f})") @@ -176,102 +177,114 @@ class MultiWellManager: session.scanning_task.enabled = False session.save() logger.info(f"==== Session {session.name} terminée à {session.finished} après {session.finished - started} secondes.") + self.scan_thread = None def halt_scanning(self): self.process.data.record = False self.stop_playing.set() - self.scan_thread = None - self.test_thread = None self.well_iterator.reset() + self.process.cam._aligner.debugg = False def scanning(self, sid): try: - if self.scan_thread is None: - session = models.Session.objects.get(pk=sid) - observations = models.SessionObservation.observation_by_session(sid) - self.scan_thread = Thread(target=self._start_scanning, args=(session, observations, ), daemon=True).start() + if self.scan_thread: + return + session = models.Session.objects.get(pk=sid) + observations = models.SessionObservation.observation_by_session(sid) + self.scan_thread = 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})"} + wl = self.well_iterator.previous() + self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) + return {"state": "previous", "msg": f">>> ({wl.x}, {wl.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})"} + wl = self.well_iterator.next() + self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) + return {"state": "next", "msg": f">>> ({wl.x}, {wl.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})"} + wl = self.well_iterator.seek(numwell) + self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) + return {"state": "goto", "msg": f">>> ({wl.x}, {wl.y})"} def set_well_position(self): - w = self.well_iterator.get_current() - if w: - w.x, w.y = self.cnc_controller.get_mpos() - w.save() - if w.order == 0: - models.MultiWell.objects.filter(position__exact=w.multiwell.position).update(xbase=w.x, ybase=w.y) - return {"state": "well_position", "msg": f">>> saved ({w.x}, {w.y})"} + wl = self.well_iterator.get_current() + if wl: + wl.x, wl.y = self.cnc_controller.get_mpos() + wl.save() + if wl.order == 0: + models.MultiWell.objects.filter(position__exact=wl.multiwell.position).update(xbase=wl.x, ybase=wl.y) + return {"state": "well_position", "msg": f">>> saved ({wl.x}, {wl.y})"} return {"state": "well_position", "msg": f">>> pas de puit"} - + def _scanning_test(self, auto=False): self.stop_playing = Event() - self.process.data.tube_diameter = self.multiwell.diameter cam = self.process.cam - duration = self.duration if not auto else CALIBRATION_AUTO_DURATION - start_test = time.monotonic() - - for w in self.well_iterator: - if self.stop_playing.is_set(): - break - self.cnc_controller.wait_for(2.0) - self.cnc_controller.move_to(w.x, w.y, feed=w.multiwell.feed) - self.process._send(current=w.order) - - start = time.monotonic() - while not self.stop_playing.is_set(): - if time.monotonic() - start > duration: + cam._aligner.set_tube_diameter(self.multiwell.diameter) + duration = self.duration if not auto else settings.CALIBRATION_AUTO_DURATION + try: + start_test = time.monotonic() + for wl in self.well_iterator: + if self.stop_playing.is_set(): break - if auto and cam._last_detection: - if cam._last_detection.get('action')=="grbl": - self.cnc_controller.wait_for(CALIBRATION_AUTO_TIMEOUT) - dx_mm = cam._last_detection["offset_x_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) - msg = f"Correction CNC move_relative(dx={dx_mm:.3f}, dy={dy_mm:.3f})" - self.process._send(state='center', msg=msg) - 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})" - logger.info(msg) - self.process._send(state='save', msg=msg) - w.x, w.y = self.cnc_controller.x, self.cnc_controller.y - w.save() - if w.order == 0: - models.MultiWell.objects.filter(position__exact=self.position).update(xbase=w.x, ybase=w.y) + self.cnc_controller.wait_for(2.0) + self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) + self.process._send(current=wl.order) + + start = time.monotonic() + while not self.stop_playing.is_set(): + if time.monotonic() - start > duration: break - self.cnc_controller.wait_for(0.1) - logger.info("Fin du centrage") + + if auto: + msg = cam.align_detection["msg"] + if cam.align_detection.get('detected'): + if cam.align_detection.get('action')=="grbl": + self.cnc_controller.wait_for(settings.CALIBRATION_AUTO_TIMEOUT) + dx_mm, dy_mm = cam.align_detection["offset_x_mm"], cam.align_detection["offset_y_mm"] + + self.cnc_controller.move_to(self.cnc_controller.x + dx_mm, self.cnc_controller.y + dy_mm, feed=150) + self.process._send(state='center', msg=msg) + + elif cam.align_detection.get('action') in ['none',]: + logger.info(msg) + self.process._send(state='save', msg=msg) + wl.x, wl.y = self.cnc_controller.x, self.cnc_controller.y + wl.save() + if wl.order == 0: + models.MultiWell.objects.filter(position__exact=self.position).update(xbase=wl.x, ybase=wl.y) + break + else: + logger.info(msg) + self.process._send(state='center', msg=msg) + + self.cnc_controller.wait_for(0.1) + logger.info("Fin du centrage") + except Exception as e: + print(e) + self.well_iterator.reset() - - logger.info(f"Scan terminé — retour à l'origine (X=0, Y=0) en {time.monotonic()-start_test} s") + self.process.cam._aligner.debug = False + + logger.info(f"Scan terminé — retour à l'origine (X=0, Y=0) en {int(time.monotonic()-start_test)} s") self.cnc_controller.move_to(0, 0, feed=self.multiwell.feed*2) + self.test_thread = None def scan_test(self, auto=False): - if self.test_thread is None: - self.test_thread = Thread(target=self._scanning_test, args=(auto, ), daemon=True).start() + if self.test_thread: + return + self.test_thread = Thread(target=self._scanning_test, args=(auto, ), daemon=True).start() @property @@ -340,9 +353,9 @@ class MultiWellManager: def get_well_order(self): - w = self.well_iterator.get_current() - if w: - return w.order + wl = self.well_iterator.get_current() + if wl: + return wl.order return None @@ -351,9 +364,9 @@ class MultiWellManager: 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 - w = self.well_iterator.seek(0) # base puit 0 - w.x, w.y = x, y - w.save() + wl = self.well_iterator.seek(0) # base puit 0 + wl.x, wl.y = x, y + wl.save() def calib_toggle_debug(self): diff --git a/test_tube_scanner/scanner/process.py b/test_tube_scanner/scanner/process.py index 629e888..e1adb41 100644 --- a/test_tube_scanner/scanner/process.py +++ b/test_tube_scanner/scanner/process.py @@ -412,11 +412,10 @@ class ScannerProcess(Task): continue elif topic == 'center': - dx_mm = self.cam._last_detection["offset_x_mm"] - dy_mm = self.cam._last_detection["offset_y_mm"] + dx_mm = self.cam.align_detection["offset_x_mm"] + dy_mm = self.cam.align_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) + self._send(state='center', msg=self.cam.align_detection["msg"]) continue elif topic == 'halt': diff --git a/test_tube_scanner/scanner/templates/scanner/base.html b/test_tube_scanner/scanner/templates/scanner/base.html index 745ddb1..c577eb6 100644 --- a/test_tube_scanner/scanner/templates/scanner/base.html +++ b/test_tube_scanner/scanner/templates/scanner/base.html @@ -91,24 +91,25 @@ {% trans "Logs des services" %} - - - {% trans "Calibration" %} - + {% endif %} {% block sidebar_smenu %} - + {% if request.user.is_superuser and request.user.is_staff %} + + {% trans "Calibration" %} + + {% endif %} - {% trans "Balayage multi-puits" %} + {% trans "Balayage multi-puits" %} - {% trans "Gestionnaire d'images" %} + {% trans "Gestionnaire d'images" %} - {% trans "Redifusion vidéos" %} + {% trans "Redifusion vidéos" %} {% endblock %}