viewer.py

Go to the documentation of this file.

from pathlib import Path
import pandas as pd, json, threading, urllib.parse
from http.server import HTTPServer, BaseHTTPRequestHandler
from IPython.display import display, IFrame
import mimetypes
import json
import re
import time
import weakref
import socket
from tqdm import tqdm
import os
import platform
from .isiCellPy import Simu, CellVector
 
STATIC_DIR = Path(__file__).resolve().parent / 'static'
 
def find_free_port(start=8000, end=9000, host='localhost'):
    for port in range(start, end + 1):
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            # permet de réutiliser immédiatement un port TIME_WAIT
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            try:
                s.bind((host, port))
                return port
            except OSError:
                # port occupé, on essaie le suivant
                continue
    raise RuntimeError(f"No free port in range {start}-{end}")
 
 
 
def find_vscode_settings():
    """Return the path to settings.json: workspace first, then user."""
    # 1) Workspace settings
    cwd = Path.cwd()
    ws_settings = cwd / ".vscode" / "settings.json"
    if ws_settings.exists():
        return ws_settings
 
    # 2) User settings
    home = Path.home()
    system = platform.system()
    if system == 'Linux':
        user_settings = home / ".config/Code/User/settings.json"
    elif system == 'Darwin':
        user_settings = home / "Library/Application Support/Code/User/settings.json"
    elif system == 'Windows':
        appdata = os.environ.get('APPDATA')
        if not appdata:
            raise FileNotFoundError("APPDATA not set on Windows")
        user_settings = Path(appdata) / "Code/User/settings.json"
    else:
        raise OSError(f"Unsupported system {system!r}")
 
    if user_settings.exists():
        return user_settings
 
    raise FileNotFoundError(f"settings.json not found at: {user_settings}")
 
 
def get_default_theme_resources():
    """Return the default theme name and the VS Code resources directory."""
    system = platform.system()
    if system == 'Linux':
        resources = Path("/usr/share/code/resources/app")
    elif system == 'Darwin':
        resources = Path("/Applications/Visual Studio Code.app/Contents/Resources/app")
    elif system == 'Windows':
        pf = os.environ.get('ProgramFiles', 'C:/Program Files')
        resources = Path(pf) / "Microsoft VS Code/resources/app"
    else:
        raise OSError(f"Unsupported system {system!r}")
 
    prod_file = resources / "product.json"
    try:
        cfg = json.loads(prod_file.read_text(encoding='utf-8'))
        default_theme = cfg.get('defaultColorTheme', '')
        stripped = re.sub(r'(?i)^default\s*', '', default_theme).strip()
        # Fallback to Dark+ if none specified
        return (stripped or 'Dark+'), resources
    except Exception:
        # On any error, default back to Dark+
        return 'Dark+', resources
 
 
def normalize(text):
    """Normalize a string to lowercase alphanumeric only."""
    return re.sub(r'[^a-z0-9]', '', text.lower() if text else '')
 
 
def matches(norm_name, norm_target):
    """Return True if either normalized name matches or contains the target."""
    if not norm_name or not norm_target:
        return False
    return norm_name == norm_target or norm_name in norm_target or norm_target in norm_name
 
 
def scan_for_theme(base_dir, norm_target):
    """Scan base_dir/*/themes/*.json to find a matching theme."""
    if not base_dir.exists():
        return None
 
    for ext in sorted(base_dir.iterdir()):
        themes_dir = ext / "themes"
        if not themes_dir.exists():
            continue
 
        for f in sorted(themes_dir.glob("*.json")):
            try:
                data = json.loads(f.read_text(encoding='utf-8'))
            except Exception:
                continue
 
            name_field = data.get("name") or data.get("label") or ''
            if matches(normalize(name_field), norm_target) or matches(normalize(f.stem), norm_target):
                return f
 
    return None
 
 
def find_theme_file(theme_name=None):
    """Find the JSON file for the given VS Code theme."""
    # Determine theme name: workspace/user or default
    settings_path = find_vscode_settings()
    settings = json.loads(settings_path.read_text(encoding='utf-8'))
    theme = theme_name or settings.get("workbench.colorTheme")
 
    default, resources_app = get_default_theme_resources()
    name = theme or default
    norm_target = normalize(name)
 
    # 1) User extensions directory
    user_ext = Path.home() / ".vscode" / "extensions"
    file = scan_for_theme(user_ext, norm_target)
    if file:
        return file
 
    # 2) Built-in extensions directory
    builtin_ext = resources_app / "extensions"
    file = scan_for_theme(builtin_ext, norm_target)
    if file:
        return file
 
    raise FileNotFoundError(f"Theme '{name}' not found in {user_ext} or {builtin_ext}")
 
 
def load_theme_colors(theme_path):
    """Load and return the 'colors' dict from the theme JSON, following includes if needed."""
    data = json.loads(theme_path.read_text(encoding='utf-8'))
    colors = data.get("colors")
    if isinstance(colors, dict):
        return colors
 
    inc = data.get("include")
    if inc:
        inc_path = theme_path.parent / inc
        if not inc_path.exists() and inc.startswith("./"):
            inc_path = theme_path.parent / inc[2:]
        if inc_path.exists():
            return load_theme_colors(inc_path)
 
    return {}
 
 
 
def get_colors():
    try:
        theme_name = None
        theme_file = find_theme_file(theme_name)
        colors = load_theme_colors(theme_file)
        return colors
    except Exception as e:
        return {}
 
def render_template(file, context=None):
    """
    Lit le fichier à `path`, puis pour chaque {{ expr }} dans le texte
    évalue `expr` avec eval(expr, {}, context) et remplace par str(result).
    
    context : dict des variables disponibles dans vos expressions.
    """
    context = {} if context is None else context
    # Lecture brute
    
    text = (STATIC_DIR / 'templates' / file).read_text()
 
    # Regex pour capturer {{ ... }} non-greedy
    pattern = re.compile(r"\{\{\s*(.*?)\s*\}\}", re.DOTALL)
    
    def _repl(match):
        expr = match.group(1)
        try:
            # Évaluer dans un contexte silencieux
            result = eval(expr, {}, context)
        except Exception as e:
            # En cas d’erreur, on restitue l’original (ou lever)
            print(f"Erreur lors de l'évaluation de l'expression '{expr}': {e}")
            return match.group(0)
        return str(result)
    
    return pattern.sub(_repl, text).encode('utf-8')
 
class StepHandler(BaseHTTPRequestHandler):
 
    def log_message(self, format, *args):
        pass
 
    def do_POST(self):
        url = urllib.parse.urlparse(self.path)
        if url.path == '/update_colors':
            content_length = int(self.headers['Content-Length'])
            body = self.rfile.read(content_length)
            colors = json.loads(body.decode('utf-8'))
            with open(STATIC_DIR / 'config' / 'colors.json', 'w') as f:
                json.dump(colors, f, indent=2)
            self.send_response(200)
            self.end_headers()
            self.wfile.write(b'Colors updated successfully.')
        else:
            self.send_error(404, "Not Found")
 
    def do_GET(self):
        url = urllib.parse.urlparse(self.path)
 
        if url.path == '/data':
            params = urllib.parse.parse_qs(url.query)
            step    = int(params.get('step', ['0'])[0])
            feature = params.get('feature', [''])[0]
            if hasattr(self.server, 'df'):
                df = self.server.df
                df0 = df[df['step'] == step]
            else:
                df0 = self.server.df_get(step)
            if df0 is None:
                self.generic_response('data.json', body=json.dumps({"error": "No data for this step"}).encode())
                return
            query = params.get('filter', [''])[0]
            if query:
                try:
                    df0 = df0.query(query)
                except Exception as e:
                    self.generic_response('data.json', body=json.dumps({"error": str(e)}).encode())
                    return
            if self.server.is3D:
                payload = {
                    'x':      df0['positionx'].tolist(),
                    'y':      df0['positiony'].tolist(),
                    'z':      df0['positionz'].tolist(),
                    'radius': df0['radius'].tolist(),
                    'info':    df0[feature].cat.codes.tolist() if isinstance(df0[feature].dtype, pd.CategoricalDtype) else df0[feature].tolist()
                }
            else:
                payload = {
                    'x':      df0['positionx'].tolist(),
                    'y':      df0['positiony'].tolist(),
                    'info':    df0[feature].cat.codes.tolist() if isinstance(df0[feature].dtype, pd.CategoricalDtype) else df0[feature].tolist()
                }
            body = json.dumps(payload).encode()
            self.generic_response('data.json', body=body)
 
        elif url.path == '/':
 
            if hasattr(self.server, 'df'):
                steps = sorted(self.server.df['step'].unique())
                df = self.server.df
            else:
                steps = self.server.steps
                df = self.server.df_get(0)
 
            feats = [c for c in df.columns if c not in ('step','id','positionx','positiony','positionz','radius')]
            colors = json.loads((STATIC_DIR / 'config' / 'colors.json').read_text())
            html = render_template('index.html',
                context={
                    'json': json,
                    'steps': steps,
                    'dt': self.server.dt,
                    'colors': colors,
                    'is3D': self.server.is3D,
                    'isDebug': self.server.isDebug,
                    'themeColors': get_colors(),
                    'features': {f: ({'isEnum':True,'values':list(df[f].dtypes.categories)} if isinstance(df[f].dtype, pd.CategoricalDtype) else {'isEnum':False,'min':float(df[f].min()),'max':float(df[f].max())}) for f in feats}
            })
            self.generic_response('index.html', body=html)
        elif '..' not in url.path:
            self.generic_response(url.path)
 
    def generic_response(self, path, body=None):
            self.send_response(200)
            self.send_header('Content-Type', (mimetypes.guess_type(path)[0] or 'application/octet-stream'))
            self.send_header('Access-Control-Allow-Origin', '*')
            self.end_headers()
            try:
                if body:
                    self.wfile.write(body)
                else:
                    self.wfile.write((STATIC_DIR / path[1:]).read_bytes())
            except FileNotFoundError as e:
                print(e)
                self.send_error(404)
 
def _cleanup(server, thread):
    """Stop the HTTPServer and join its thread."""
    try:
        #print("Arrêt du serveur HTTP...")
        server.shutdown()
        server.server_close()
    except Exception:
        pass
    try:
        if thread.is_alive():
            thread.join()
    except Exception:
        pass
 
 
class Viewer:
    """
    A live or replay 2D/3D simulation viewer embedded in Jupyter.
 
    This class starts an HTTPServer in a background thread.
 
    You can use it in **replay** mode—feeding it a precomputed DataFrame—or
    in **live** mode—running a Simu instance step by step and streaming
    results as they arrive.
 
    The server is automatically cleaned up when the Viewer object is
    garbage-collected or when you explicitly call `.stop()`.
    """
    def __init__(self,is3D,port=None,debug=False):
        """
        Instantiate and start the HTTP server.
 
        Args:
            is3D (bool):  If True, viewer expects 3D coordinates (x,y,z),
                          otherwise only (x,y).
            port (int, optional):  TCP port to bind the server on. If None,
                          tries to find a free port in [8765, 8865].
            debug (bool): If True, enables javascript debug mode with additional logging.
 
        Raises:
            RuntimeError: If the server thread fails to start within 5 seconds.
        """
        self.port = port or find_free_port(start=8765, end=8765+100)
        self.server = HTTPServer(('localhost', self.port), StepHandler)
        self.server.is3D = is3D
        self.server.ready = False
        self.server.isLive = False
        self.server.isReplay = False
        self.server.isDebug = debug
        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
        self.thread.start()
        timeout = time.time() + 5.0   # 5 secondes max
        while not self.thread.is_alive():
            if time.time() > timeout:
                self.server.shutdown()
                self.server.server_close()
                raise RuntimeError("Le serveur n'a pas pu démarrer à temps.")
            time.sleep(0.01)
        
        self._finalizer = weakref.finalize(self, _cleanup, self.server, self.thread)
 
    def replay(self, df: pd.DataFrame, dt: float) -> "Viewer":
        """
        Load a precomputed DataFrame for step-by-step replay.
 
        Args:
            df (pd.DataFrame):  Must contain columns ['step','id',…,'positionx',…].
            dt (float):         Time interval between steps.
 
        Returns:
            self, for method chaining.
 
        Raises:
            RuntimeError: If already in live mode.
            TypeError:    If df is not a pandas DataFrame.
        """
        if self.server.isLive:
            raise RuntimeError("Le viewer est déjà en mode live.")
        if not isinstance(df, pd.DataFrame):
            raise TypeError("df doit être un DataFrame Pandas.")
        self.server.df = df.sort_values(by=['step','id']).reset_index(drop=True)
        self.server.dt = dt
        self.server.isReplay = True
        self.server.steps = sorted(df.step.unique())
        self.server.ready = True
        return self
 
    def live(self, params: dict, infos: list[str]|str = ['type', 'state'], seed: int = None, benchmark: bool = False, withTqdm: bool = False) -> "Viewer":
        """
        Run a Simu instance live, collect data each step, and stream it.
 
        Args:
            params (dict):      Simulation parameters (must include Scenario.maxStep & dt).
            infos (list[str]|str):  List of cell attributes to retrieve, or 'all'
            withTqdm (bool):    If True, show a tqdm progress bar.
 
        Returns:
            self, for method chaining.
 
        Raises:
            RuntimeError: If already in replay mode.
        """
        if self.server.isReplay:
            raise RuntimeError("Le viewer est déjà en mode replay.")
        maxStep = params['input']['Scenario']['maxStep']
        dt = params['input']['Scenario']['dt']
        datas = []
        if infos == 'all':
            infos = CellVector.attrs
        if self.server.is3D:
            infos = list(set(infos) | set(['id','positionx','positiony','positionz','radius']))
        else:
            infos = list(set(infos) | set(['id','positionx','positiony']))
 
        def dataCatcher(simu):
            return simu.cells.getPandas(infos).sort_values('id').reset_index(drop=True)
 
        def df_get(step):
            return datas[step] if step < len(datas) else None
 
        self.server.df_get = df_get
        self.server.dt = dt
        self.server.steps = list(range(maxStep))
        if seed is None:
            simu = Simu(params,benchmark=benchmark)
        else:
            simu = Simu(params, seed=seed, benchmark=benchmark)
        datas.append(dataCatcher(simu))
        self.server.ready = True
        self.show()
        if withTqdm:
            stepIter = tqdm(simu.iterSteps(self.server.steps[1:]), total=maxStep-1, desc="Processing steps")
        else:
            stepIter = simu.iterSteps(self.server.steps[1:])
        for s in stepIter:
            datas.append(dataCatcher(simu))
 
        self.datas = datas
        if benchmark:
            simu.showBenchmarksSummaries()
        self.simu = simu
        return self
    
    def getData(self) -> pd.DataFrame:
        """
        Returns a single concatenated DataFrame of all steps collected in live mode.
        If replay mode was used, returns the original df.
 
        Returns:
            pd.DataFrame:  Index “step”, with all columns from dataCatcher.
 
        """
        if not hasattr(self.server, 'df'):
            self.server.df = pd.concat(self.datas, keys=self.server.steps, names=['step'], copy=False).droplevel(1).reset_index()
            del self.server.df_get
        return self.server.df
 
    def show(self, width: int = None, height: int = None):
        """
        Display the viewer inline in a notebook via IFrame.
 
        Args:
            width (int, optional):  Width of the plotting area in px; if None uses 100% width.
            height (int, optional): Height in px; if None uses a default of 800px.
        """
        width  = f"{width+250}px" if width else "100%"
        height = f"{height}px" if height else "800px"
        while (not self.server.ready):
            time.sleep(0.01)
        display(IFrame(src=f"http://localhost:{self.port}", width=width, height=height))
 
    def _repr_html_(self) -> str:
        """
        Jupyter HTML repr: Embed the viewer automatically when returning this object.
        """
        while (not self.server.ready):
            time.sleep(0.01)
        return f'<iframe src="http://localhost:{self.port}" width="100%" height="800px" frameborder="0"></iframe>'
 
    def stop(self):
        """
        Stop the HTTP server and join its thread immediately.
        Also prevents the automatic finalizer from running again.
        """
        self._finalizer()