post processing (artic fox, peacock) edits for lynx

Author: Ryan Calvin Barron

TELF/applications/Lynx/frontend/pages/doc_view.py

@@ -181,6 +181,7 @@
     for ii in selected_nodes["checked"]:
         directory = st.session_state.data_map[ii]["path"]
         peacock_dir = os.path.join(directory, "peacock")
+        print( peacock_dir )
         files = find_files_by_extensions(peacock_dir, extensions=("html", "png"))
         with st.expander(st.session_state.data_map[ii]["label"]):
             open_explorer_button(peacock_dir, key=f"button_tab2_{ii}")

TELF/pipeline/__init__.py

@@ -73,3 +73,4 @@
 from .blocks.termite_vector_block import TermiteVectorBlock
 
 from .blocks.author_affiliation_tables import AffiliationsAndAuthorsBlock
+from .blocks.block_helpers.KernelServer import KernelTiedServer

TELF/pipeline/blocks/__init__.py

@@ -63,4 +63,6 @@
 from .collect_hnmfk_leaf_block import CollectHNMFkLeafBlock 
 from .termite_neo4j_block import TermiteNeo4jBlock
 from .termite_vector_block import TermiteVectorBlock
-from .author_affiliation_tables import AffiliationsAndAuthorsBlock
+from .author_affiliation_tables import AffiliationsAndAuthorsBlock
+
+from .block_helpers.KernelServer import KernelTiedServer

TELF/pipeline/blocks/artic_fox_block.py

@@ -1,14 +1,21 @@
 from pathlib import Path
-from typing import Dict, Sequence, Any, Tuple
+from typing import Dict, Any
 from .base_block import AnimalBlock
 from .data_bundle import DataBundle, SAVE_DIR_BUNDLE_KEY
 
 from ...post_processing import ArcticFox
 from ...factorization import HNMFk
 
+
 class ArticFoxBlock(AnimalBlock):
+    """
+    Block wrapper for the ArcticFox post-process/label/stats pipeline.
+    Use call_settings['steps'] to run any subset: ["post"], ["label"], ["stats"],
+    or combinations like ["post","label"], ["label","stats"], ["post","stats"], ["post","label","stats"].
+    If steps is None, legacy behavior uses the label_clusters/generate_stats booleans.
+    """
 
-    CANONICAL_NEEDS = ("df", 'vocabulary', "model_path",)
+    CANONICAL_NEEDS = ("df", "vocabulary", "model_path")
 
     def __init__(
         self,
@@ -21,62 +28,64 @@ def __init__(
         call_settings: Dict[str, Any] = None,
         **kw,
     ) -> None:
-        
+
         self.col = col
         default_init = {
-            'clean_cols_name': self.col,
-            'embedding_model': "SCINCL", 
+            "clean_cols_name": self.col,
+            "embedding_model": "SCINCL",
         }
         default_call = {
-            'ollama_model': "llama3.2:3b-instruct-fp16",  # Language model used for semantic label generation
-            'label_clusters': True,             # Enable automatic labeling of clusters
-            'generate_stats': True,             # Generate cluster-level statistics
-            'process_parents': True,            # Propagate labels or stats upward through the hierarchy
-            'skip_completed': True,             # Skip processing of nodes already labeled/stored
-            'label_criteria': {                 # Rules to filter generated labels
-                "minimum words": 2,
-                "maximum words": 6
-            },
-            'label_info': {                     # Additional metadata to associate with generated labels
-                 "source": "Science"
-            },
-            'number_of_labels': 5               # Number of candidate labels to generate per node
+            "ollama_model": "llama3.2:3b-instruct-fp16",  # Language model used for semantic label generation
+            "label_clusters": True,            # Back-compat: used when steps is None
+            "generate_stats": True,            # Back-compat: used when steps is None
+            "process_parents": True,
+            "skip_completed": True,
+            "label_criteria": {"minimum words": 2, "maximum words": 6},
+            "label_info": {"source": "Science"},
+            "number_of_labels": 5,
+            # NEW: choose subset explicitly; None keeps legacy boolean behavior
+            # Examples: ["post"], ["label"], ["stats"], ["post","label"], ["label","stats"], ["post","stats"], ["post","label","stats"]
+            "steps": None,
         }
 
         super().__init__(
-            needs = needs,
-            provides = provides,
+            needs=needs,
+            provides=provides,
             init_settings=self._merge(default_init, init_settings),
             call_settings=self._merge(default_call, call_settings),
             tag=tag,
             **kw,
         )
 
-
     def run(self, bundle: DataBundle) -> None:
+        # Resolve inputs
         df = self.load_path(bundle[self.needs[0]])
         vocabulary = self.load_path(bundle[self.needs[1]])
-
         raw_model_path = str(bundle[self.needs[2]])
-        # Try to resolve to an absolute path for traceability; fall back to the raw string.
+
         try:
             resolved_model_path = str(Path(raw_model_path).expanduser().resolve())
         except Exception:
             resolved_model_path = raw_model_path
 
+        # Load HNMFk model
         model = HNMFk(experiment_name=raw_model_path)
         model.load_model()
 
+        # Run selected steps (order enforced inside ArcticFox)
         pipeline = ArcticFox(model=model, **self.init_settings)
-        pipeline.run_full_pipeline(data_df=df, vocab=vocabulary, **self.call_settings)
+        pipeline.run_full_pipeline(
+            data_df=df,
+            vocab=vocabulary,
+            **self.call_settings
+        )
 
+        # Write a lightweight status checkpoint
         status_value = "Done"
-
         if SAVE_DIR_BUNDLE_KEY in bundle:
             out_dir = Path(bundle[SAVE_DIR_BUNDLE_KEY]) / self.tag
             out_dir.mkdir(parents=True, exist_ok=True)
             status_file = out_dir / "status.txt"
-            # Include model path info in the checkpointed status file
             status_file.write_text(
                 f"status: {status_value}\n"
                 f"model_path: {raw_model_path}\n"
@@ -86,7 +95,3 @@ def run(self, bundle: DataBundle) -> None:
             self.register_checkpoint(self.provides[0], status_file)
 
         bundle[f"{self.tag}.{self.provides[0]}"] = status_value
-
-
-
-        

TELF/pipeline/blocks/block_helpers/KernelServer.py

@@ -0,0 +1,109 @@
+import os, signal, subprocess, pathlib, shlex, time
+
+class KernelTiedServer:
+    def __init__(self, cmd, *, cwd=None, env=None, log_dir=None):
+        """
+        cmd: list or string (string is shell-split)
+        cwd: working dir
+        env: dict of env vars
+        log_dir: if set, stdout/stderr are appended to files here
+        """
+        if isinstance(cmd, str):
+            cmd = shlex.split(cmd)
+        self.cmd = cmd
+        self.cwd = cwd
+        self.env = env
+        self.log_dir = pathlib.Path(log_dir).expanduser() if log_dir else None
+        self.proc = None
+        self.watchdog = None
+        self.stdout_f = None
+        self.stderr_f = None
+
+    def start(self):
+        if self.proc and self.running:
+            raise RuntimeError("Already running")
+        if self.log_dir:
+            self.log_dir.mkdir(parents=True, exist_ok=True)
+            self.stdout_f = open(self.log_dir / "server.out", "ab", buffering=0)
+            self.stderr_f = open(self.log_dir / "server.err", "ab", buffering=0)
+
+        # 1) start the server in its own process group
+        self.proc = subprocess.Popen(
+            self.cmd,
+            cwd=self.cwd,
+            env=self.env,
+            stdout=self.stdout_f or subprocess.DEVNULL,
+            stderr=self.stderr_f or subprocess.DEVNULL,
+            preexec_fn=os.setpgrp,  # new PGID == proc.pid (POSIX)
+        )
+
+        # 2) watchdog: when kernel PID disappears, kill the whole PGID
+        kernel_pid = os.getpid()
+        pgid = self.proc.pid
+        script = f"""
+        while kill -0 {kernel_pid} 2>/dev/null; do sleep 2; done
+        kill -TERM -{pgid} 2>/dev/null
+        sleep 5
+        kill -KILL -{pgid} 2>/dev/null
+        """
+        self.watchdog = subprocess.Popen(
+            ["bash", "-c", script],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
+        )
+        return self
+
+    @property
+    def running(self):
+        return self.proc is not None and self.proc.poll() is None
+
+    def status(self):
+        if not self.proc:
+            return "not started"
+        rc = self.proc.poll()
+        return f"running (pid {self.proc.pid}, pgid {self.proc.pid})" if rc is None else f"exited rc={rc}"
+
+    def stop(self, sig=signal.SIGTERM, hard_after=5):
+        if not self.proc:
+            return
+        if self.running:
+            try:
+                os.killpg(self.proc.pid, sig)
+            except ProcessLookupError:
+                pass
+            # optional hard kill after grace
+            t0 = time.time()
+            while self.running and time.time() - t0 < hard_after:
+                time.sleep(0.2)
+            if self.running:
+                try:
+                    os.killpg(self.proc.pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+        # stop watchdog
+        if self.watchdog and self.watchdog.poll() is None:
+            try:
+                self.watchdog.terminate()
+            except Exception:
+                pass
+        # close logs
+        for f in (self.stdout_f, self.stderr_f):
+            try:
+                f and f.close()
+            except Exception:
+                pass
+
+    def tail(self, n=50, which="out"):
+        if not self.log_dir:
+            print("(no logs: set log_dir=...)")
+            return
+        path = self.log_dir / ( "server.out" if which=="out" else "server.err" )
+        try:
+            with open(path, "rb") as f:
+                print(b"".join(f.readlines()[-n:]).decode(errors="replace"))
+        except FileNotFoundError:
+            print("(no log yet)")
+
+# --- Example ---
+# srv = KernelTiedServer(["zsh", "../../Lynx/start_lynx.sh"], log_dir="~/.logs/lynx").start()
+# print(srv.status()); srv.tail(100)         # view last 100 lines of stdout
+# srv.stop()                                 # stop manually

TELF/pipeline/blocks/block_helpers/__init__.py

@@ -0,0 +1 @@
+from .KernelServer import KernelTiedServer