diff --git a/test_tube_scanner/home/management/commands/start_workers.py b/test_tube_scanner/home/management/commands/start_workers.py index 84f18ef..4448892 100644 --- a/test_tube_scanner/home/management/commands/start_workers.py +++ b/test_tube_scanner/home/management/commands/start_workers.py @@ -7,7 +7,6 @@ from django.core.management.base import BaseCommand from scanner import tasks as scanner_tasks - class Command(BaseCommand): help = "Démarre les tâches Celery." @@ -18,9 +17,9 @@ class Command(BaseCommand): def handle(self, *args, **options): # @UnusedVariable #task = options['task'] try: - scanner_tasks.scanner_start.delay() # @UndefinedVariable - scanner_tasks.replay_start.delay() # @UndefinedVariable + scanner_tasks.replay_start.delay() # @UndefinedVariable + except Exception as e: print(e) diff --git a/test_tube_scanner/scanner/consumers.py b/test_tube_scanner/scanner/consumers.py index 0fc6abf..3daf173 100644 --- a/test_tube_scanner/scanner/consumers.py +++ b/test_tube_scanner/scanner/consumers.py @@ -24,6 +24,7 @@ class ScannerConsumer(AsyncWebsocketConsumer): logger.info( f"==== Disconnect from {self.this_group}") async def scanner_message(self, event): + logger.info( f"===={event.get('type')}") await self.send(text_data=json.dumps(event["text"])) ## Receive message from WebSocket @@ -33,9 +34,6 @@ class ScannerConsumer(AsyncWebsocketConsumer): if msg_type in ["scanner", "calibrate"]: redisDB.publish(self.this_group, json.dumps(data)) - async def replay_message(self, event): - await self.send(text_data=json.dumps(event["text"])) - class ReplayConsumer(AsyncWebsocketConsumer): diff --git a/test_tube_scanner/scanner/models.py b/test_tube_scanner/scanner/models.py index da3a86e..45054af 100644 --- a/test_tube_scanner/scanner/models.py +++ b/test_tube_scanner/scanner/models.py @@ -38,6 +38,12 @@ CAPTURE_TYPE = [ ('file', _("mp4")), ] + +class TaskID(models.Model): + code = models.CharField(unique=True, max_length=64, null=True, blank=False) + task_id = models.CharField(max_length=250, null=True, blank=False, default="") + + class Configuration(models.Model): name = models.CharField(_("Nom de la Configuration"), help_text=_("Nom de la configuration"), max_length=100, null=True, blank=False, default=_("Configuration par défaut")) author = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name="Auteur", null=True, blank=True) @@ -279,8 +285,8 @@ class Session(models.Model): class Meta: ordering = ['-created', ] - verbose_name = _("Session d'expérience") - verbose_name_plural = _("Sessions d'expériences") + verbose_name = _("Session") + verbose_name_plural = _("Sessions") def __str__(self): state = _("Terminée") if not self.active else _("Active") diff --git a/test_tube_scanner/scanner/process.py b/test_tube_scanner/scanner/process.py index f46dee5..8d15098 100644 --- a/test_tube_scanner/scanner/process.py +++ b/test_tube_scanner/scanner/process.py @@ -139,11 +139,12 @@ class CameraRecordManager(): class ScannerProcess(Task): - + name='scanner.scanner_process' def __init__(self): super().__init__() + self.channel_layer = get_channel_layer() - self.group = f'scanner_proc' + self.group = 'scanner_proc' self.stop_event = Event() self.cam = None self.grbl = None @@ -162,6 +163,8 @@ class ScannerProcess(Task): def start(self, *args, **kwargs): try: + logger.warning("ScannerProcess==================", self.name) + self.conf = ScannerConstants().get() self.use_tracking = self.conf.tracking @@ -234,6 +237,8 @@ class ScannerProcess(Task): self.stop_event.clear() self.start_services() + + return self.name except Exception as e: logger.error(f"Scanner started error: {e}") raise Ignore() @@ -254,6 +259,7 @@ class ScannerProcess(Task): Thread(target=self._recording, daemon=True).start() Thread(target=self._init_grbl, daemon=True).start() self.cam.start() + logger.warning(f"Scanner services started ...") def _send(self, **payload): async_to_sync(self.channel_layer.group_send)( @@ -335,6 +341,8 @@ class ScannerProcess(Task): self.grbl.go_origin(feed=self.manager.feed) elif topic == 'scan': + + logger.info(f"==== Scan {cmd}") sid = cmd.get("session", '0') if sid == "0": self._send(state='error', msg=str(_('La session est nulle!...'))) diff --git a/test_tube_scanner/scanner/static/scanner/js/main.js b/test_tube_scanner/scanner/static/scanner/js/main.js index 4530138..e8c3b75 100644 --- a/test_tube_scanner/scanner/static/scanner/js/main.js +++ b/test_tube_scanner/scanner/static/scanner/js/main.js @@ -43,7 +43,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.scan_state) { this.scan_state.textContent=payload.scan_state;} + if (payload.scan_state) { this.scan_state.textContent=payload.scan_state;} if (payload.detected && use_tracking) { this.cx.textContent = payload.cx; this.cy.textContent = payload.cy; diff --git a/test_tube_scanner/scanner/tasks.py b/test_tube_scanner/scanner/tasks.py index b5c6373..a2aab83 100644 --- a/test_tube_scanner/scanner/tasks.py +++ b/test_tube_scanner/scanner/tasks.py @@ -1,62 +1,77 @@ # tasks.py import asyncio +import json +from asgiref.sync import async_to_sync #, sync_to_async +from channels.layers import get_channel_layer from celery import shared_task, group, chord, chain from celery.utils.log import get_task_logger from django.utils import timezone -from .process import ScannerProcess, ReplayProcess +from .process import ScannerProcess, ReplayProcess, redisDB from .export_tasks import shm_download_video, export_images_zip, export_video_mp4 from . import models logger = get_task_logger(__name__) + +SCANNER = 'scanner' +REPLAY = 'replay' + class ScannerTaskManager: def __init__(self): - self.scanner = None - self.replay = None + self.active_task = {} def start_scanner(self): - if self.scanner is None: - self.scanner = ScannerProcess() - self.scanner.start() - + if SCANNER not in self.active_task: + scanner = ScannerProcess() + scanner.start() + self.active_task[SCANNER] = scanner + logger.warning(f"scanner {self.active_task[SCANNER]} running!...") + + else: + logger.warning(f"scanner already running!...") + + logger.warning(f"scanner {self.active_task} ...") + def stop_scanner(self): - if self.scanner: - self.scanner.stop() + if SCANNER in self.active_task: + self.active_task[SCANNER].stop() + del self.active_task[SCANNER] def start_replay(self, latency=5.0): - if self.replay is None: - self.replay = ReplayProcess(latency=latency) - self.replay.start() + if REPLAY not in self.active_task: + replay = ReplayProcess(latency=latency) + replay.start() + self.active_task[REPLAY] = replay + else: + logger.warning(f"replay already running!...") + + logger.warning(f"replay {self.active_task} ...") def stop_replay(self): - if self.replay: - self.replay.stop() + if REPLAY in self.active_task: + self.active_task[REPLAY].stop() + del self.active_task[REPLAY] task_manager = ScannerTaskManager() -@shared_task(bind=True) -def scanner_start(self): - task_manager.start_scanner() + +@shared_task() +def scanner_start(): + scanner = ScannerProcess() + scanner.start() + #task_manager.start_scanner() return f"Scanner démarré." -#@shared_task(bind=True) -#def scanner_stop(self): -# task_manager.stop_scanner() -# return f"Scanner arrêté." - -@shared_task(bind=True) -def replay_start(self): - task_manager.start_replay() +@shared_task +def replay_start(): + replay = ReplayProcess(latency=5.0) + replay.start() + #task_manager.start_replay() return f"Replay démarré." -#@shared_task(bind=True) -#def replay_stop(self): -# task_manager.stop_replay() -# return f"Replay arrêté." - @shared_task def download_video(uuid, start_ts, end_ts, frame_rate=10, opencv_fourcc_format='mp4v', opencv_video_type='mp4'): try: @@ -200,11 +215,10 @@ def scanning(session_id: str): Scanning différé. """ try: - scanner = task_manager.scanner - scanner.cam.set_median(is_median=False) - scanner.cam.set_circular_crop(scanner.crop) - scanner.grbl.go_origin(feed=scanner.manager.feed) - scanner.manager.scan(scanner.grbl, session_id) + logger.warning("==== scanning session %s start", session_id) + group = 'scanner_proc' + data = { "type": "scanner", "topic": "scan", "session": session_id } + redisDB.publish(group, json.dumps(data)) except Exception as e: logger.error(f"scanning session: {session_id} error {e}") return {"status": "error", "message": str(e)} @@ -222,7 +236,7 @@ def run_scanning(self, session_id: str): session.scanning_status = models.Session.Status.RUNNING session.save(update_fields=["scanning_status"]) chain( - scanning.s(session_id), + scanning.s(session_id), on_scanning_done.s(session_id=session_id), ).apply_async() @@ -243,4 +257,5 @@ def on_scanning_done(result: dict, session_id: str): session.scanning_finished_at = timezone.now() session.save(update_fields=["scanning_status", "scanning_finished_at"]) + logger.info("==== Scanning done: session %s", session_id) \ No newline at end of file