"""Global bounded training renderer; independent of Event ledger and transport."""
from collections import Counter
import fcntl
import os
import subprocess
import sys
import threading
import time

from app.training_store import TrainingStore, root_path


class TrainingService:
    def __init__(self, root, event_service=None):
        self.store = TrainingStore(root)
        self.events = event_service
        self.stop = threading.Event()
        self.thread = None
        self.metrics = Counter()
        self.worker_lock = None
        self.last_failure = None

    def start(self):
        self.worker_lock = os.open(self.store.root/'.worker-lock',os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW,0o600)
        fcntl.flock(self.worker_lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
        with self.store.db() as db:
            abandoned = db.execute("SELECT id FROM jobs WHERE state IN ('RUNNING','CAPTURING')").fetchall()
        for (job,) in abandoned:
            self.store.finish(job,'SERVICE_RESTART')
        self.thread = threading.Thread(target=self.run,name='training-worker',daemon=True)
        self.thread.start()

    def run(self):
        while not self.stop.wait(.5):
            job = None
            slot = False
            try:
                with self.store.db() as db:
                    expired = db.execute("SELECT id FROM jobs WHERE state IN ('CAPTURING','QUEUED') AND created<?",
                                         (time.time()-120,)).fetchall()
                for (expired_id,) in expired:
                    self.store.finish(expired_id,'QUEUE_EXPIRED')
                    self.metrics['expired'] += 1
                # Event rendering is higher priority; never add training to a busy renderer pool.
                if self.events is not None and self.events.status().get('rendering',0):
                    continue
                if self.events is not None:
                    slot = self.events.render_slots.acquire(blocking=False)
                    if not slot:
                        continue
                with self.store.lock(), self.store.db() as db:
                    row = db.execute("SELECT id,path FROM jobs WHERE state='QUEUED' ORDER BY created LIMIT 1").fetchone()
                    if not row: continue
                    job,relative = row
                    db.execute("UPDATE jobs SET state='RUNNING' WHERE id=?", (job,))
                stage = self.store.safe(relative)
                begin = time.monotonic()
                child = subprocess.Popen([sys.executable,'-m','app.training_media',str(stage)],
                    stdout=subprocess.DEVNULL,stderr=subprocess.PIPE,env=dict(os.environ,OMP_NUM_THREADS='1',OPENBLAS_NUM_THREADS='1',
                    CONTEXTRA_TRAINING_PARENT_PID=str(os.getpid())))
                from app.artifact_diagnostics import StderrTail
                tail = StderrTail(child.stderr)
                try:
                    child.wait(timeout=25)
                except subprocess.TimeoutExpired:
                    child.kill(); child.wait(timeout=3)
                stderr = tail.finish()
                if child.returncode:
                    self.last_failure = dict(jobId=job, returncode=child.returncode, stderr_tail=stderr)
                self.metrics['render_ms_total'] += round((time.monotonic()-begin)*1000)
                self.metrics['last_exit_code'] = child.returncode
                self.store.finish(job, 'RENDER_FAILED' if child.returncode else None)
                self.metrics['failed' if child.returncode else 'saved'] += 1
            except Exception as error:
                self.metrics['error_'+type(error).__name__] += 1
                if job:
                    try: self.store.finish(job,type(error).__name__)
                    except Exception: pass
            finally:
                if slot:
                    self.events.render_slots.release()

    def status(self):
        return dict(self.metrics, **self.store.summary(), last_failure=self.last_failure,
                    worker_alive=bool(self.thread and self.thread.is_alive()))

    def close(self):
        self.stop.set()
        if self.thread: self.thread.join(30)
        if self.thread and self.thread.is_alive(): return False
        if self.worker_lock is not None:
            os.close(self.worker_lock); self.worker_lock = None
        return True