''' scanner/multiwell.py WellIterator: Itérateur personnalisé pour naviguer dans les Wells MultiWellManager: Manager des multi-puits Created on 20 avr. 2026 @author: denis ''' import logging import time 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 planarian.models import ExperimentConfig from . import models 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 logger.info(f"MultiWellManager initialized with CNC controller: {self.cnc_controller}") self.stop_playing = Event() self.well_iterator = None self.multiwel = None self.set_default_values() self.set_multiwell() self.scan_thread = None self.test_thread = None self.tracker_config = dict( tube_axis = settings.TRACKER_TUBE_AXIS, min_area_px = self.process.conf.min_area_px, max_area_ratio = self.process.conf.max_area_ratio, max_planarians = self.process.conf.max_planarians, merge_kernel_size = self.process.conf.merge_kernel_size, min_contour_dist_px = self.process.conf.min_contour_dist_px, ) def set_tracker_config(self): self.tracker_config = dict( tube_axis = settings.TRACKER_TUBE_AXIS, min_area_px = self.process.conf.min_area_px, max_area_ratio = self.process.conf.max_area_ratio, max_planarians = self.process.conf.max_planarians, merge_kernel_size = self.process.conf.merge_kernel_size, min_contour_dist_px = self.process.conf.min_contour_dist_px, ) logger.info(f"Tracker config: {self.tracker_config}") 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 self.px_per_mm = 50.0 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.WellPosition.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, btn_class="w3-button", onclick=''' onclick="goto_well(this)"'''): multiwells = [] multiwells.append('''
''') for wl 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): def _grid_scanning_capture(self, experiment, well_position, simulate=False): try: well = well_position.well multiwell = experiment.multiwell if self.process.use_tracking: cfg = ExperimentConfig.objects.filter(experiment_id=experiment.id, well_id=well.id).first() if not cfg: raise Exception(f"Configuration d'expérience introuvable pour {experiment} / {well}") # reset PlanarianTracker => on_well_change self.process.cam.on_well_change(cfg, draw_contours=False) ## create uuid for this capture uuid = f'{self.process.data.session}-{multiwell.position}-{well.name}' ## start recording self.process.data.uuid = uuid if not simulate: self.process.data.record = True self.process._send(current=well_position.order) start = time.monotonic() while not self.stop_playing.is_set(): ## stop after duration in experiemnt now if time.monotonic() - start > experiment.duration: break self.cnc_controller.wait_for(0.1) self.process.data.record = False self.process.data.uuid = None msg = f"{uuid}: capture done..." except Exception as e: msg = f"error during capture - {e}" logger.error(msg) logger.info(msg) self.process._send(scan_state=msg) def _capture_file_simulation(self, name): vf = settings.MEDIA_ROOT / 'simulation' / f'{name}.mp4' if vf.exists(): self.process.cam._video_file = str(vf) self.process.cam._error_occured = True logger.info(f"Simulating capture with file {vf}") def _is_well_valid(self, welposition): names = models.ExperimentWell.wellname_by_experiment(welposition.experiment.id) if welposition.well.name not in names: return False return True def _grid_scanning(self, experiment, xnext=0, ynext=0, simulate=False): try: multiwell = experiment.multiwell wellpositions = models.WellPosition.objects.filter(multiwell_id=multiwell.id).order_by('order').all() cam = self.process.cam cam._aligner.set_tube_diameter(multiwell.diameter) for wl in wellpositions: if self.stop_playing.is_set(): break if not self._is_well_valid(wl): continue self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) if self.process.conf.capture_type == 'file': self._capture_file_simulation(wl.well.name) self._grid_scanning_capture(experiment, wl, simulate=simulate) msg =f"Scan terminé — retour à l'origine (X={xnext:.1f} Y={ynext:.1f})" logger.info(msg) self.process._send(state='scan_finished', msg=msg) except Exception as e: msg = f"Error during grid scanning - {e}" logger.error(msg) self.process._send(state='error', msg=msg) finally: self.cnc_controller.move_to(xnext, ynext, feed=multiwell.feed*2) def _start_scanning(self, session, experiments, simulate=False): try: self.process.cam._aligner.debug = False self.stop_playing.clear() xynext = [] for obs in experiments: 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 experiments: if self.stop_playing.is_set(): break obs.started = timezone.now() obs.save() xnext, ynext = xynext[pos] pos +=1 self._grid_scanning(obs, xnext=xnext, ynext=ynext, simulate=simulate) obs.finished = timezone.now() obs.save() session.finished = timezone.now() if self.stop_playing.is_set(): msg = f"Session {session.name} abandonnée à {session.finished} après {session.finished - started} secondes." else: if not simulate: session.active = False if session.scanning_task: session.scanning_task.enabled = False session.save() msg = f"Session {session.name} terminée à {session.finished} après {session.finished - started} secondes." logger.info(msg) self.process._send(scan_state=msg) except Exception as e: logger.error("Error during scanning process", e) finally: self.scan_thread = None def halt_scanning(self): self.process.data.record = False self.stop_playing.set() self.well_iterator.reset() self.process.cam._aligner.debug = False self.scan_thread = None self.test_thread = None def scan_process(self, sid, simulate=False): if self.scan_thread: return session = models.Session.objects.get(pk=sid) experiments = models.SessionExperiment.experiment_by_session(sid) self.scan_thread = Thread(target=self._start_scanning, args=(session, experiments, simulate, ), daemon=True) self.scan_thread.start() def previous_well(self): wl = self.well_iterator.previous() if self.process.conf.capture_type == 'file': self._capture_file_simulation(wl.well.name) self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) return {"state": "previous", "msg": f">>> {wl.well.name}: ({wl.x}, {wl.y})"} def next_well(self): wl = self.well_iterator.next() if self.process.conf.capture_type == 'file': self._capture_file_simulation(wl.well.name) self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) return {"state": "next", "msg": f">>> {wl.well.name}: ({wl.x}, {wl.y})"} def goto_well(self, numwell): wl = self.well_iterator.seek(numwell) if self.process.conf.capture_type == 'file': self._capture_file_simulation(wl.well.name) self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) return {"state": "goto", "msg": f">>> {wl.well.name}: ({wl.x}, {wl.y})"} def goto_xy(self): wl = self.well_iterator.seek(0) if self.process.conf.capture_type == 'file': self._capture_file_simulation(wl.well.name) self.cnc_controller.move_to(self.xbase, self.ybase, feed=self.feed) def goto_0(self): self.well_iterator.reset() if self.process.conf.capture_type == 'file': self._capture_file_simulation('zero') self.cnc_controller.move_to(0, 0, feed=self.feed*2) def set_well_position(self): 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.clear() cam = self.process.cam 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 self.cnc_controller.wait_for(2.0) self.cnc_controller.move_to(wl.x, wl.y, feed=wl.multiwell.feed) ## change file if self.process.conf.capture_type == 'file': self._capture_file_simulation(wl.well.name) self.process._send(current=wl.order) start = time.monotonic() while not self.stop_playing.is_set(): if time.monotonic() - start > duration: break 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']: msg = f"Ok centre trouvé. {msg}" logger.info(msg) self.process._send(state='save', msg=msg) wl.x, wl.y = self.cnc_controller.x, self.cnc_controller.y wl.px_per_mm = cam.align_detection.get('px_per_mm') 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 test") except Exception as e: logger.error("Error during scanning test", e) finally: self.test_thread = None self.well_iterator.reset() 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.goto_0() self.process._send(current=self.get_well_order()) def scan_test(self, auto=False): if self.test_thread: return self.test_thread = Thread(target=self._scanning_test, args=(auto, ), daemon=True) self.test_thread.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 get_well_order(self): wl = self.well_iterator.get_current() if wl: return wl.order return None 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 wl = self.well_iterator.seek(0) # base puit 0 wl.x, wl.y = x, y wl.px_per_mm = self.px_per_mm wl.save() 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", "value": aligner.debug, "msg": f"Debug: {aligner.debug}"} def set_calib_debug(self, value=True): """ Active / désactive le mode debug sur le stream.""" aligner = self.process.cam._aligner aligner.debug = value return {"state": "debug", "value": aligner.debug, "msg": f"Debug: {aligner.debug}"}