v0.4.0

Author: jet-logic

.github/workflows/build.yml

@@ -6,7 +6,7 @@ on:
   pull_request:
 
 jobs:
-  test:
+  Linux:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
@@ -22,8 +22,6 @@ jobs:
           pip install pytest pytest-cov toml
           pip install -e .
 
-      # - name: Run tests
-      #   run: pytest --cov=runce --cov-report=xml
       - name: Run tests
         run: sh tests/run_tests_with_coverage.sh
 
@@ -35,5 +33,40 @@ jobs:
             echo "Version mismatch!"
             exit 1
           fi
-    # - name: Upload coverage to Codecov
-    #   uses: codecov/codecov-action@v3
+
+  Windows:
+    runs-on: windows-latest
+
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.10"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install pytest pytest-cov toml
+          pip install -e .
+
+      # - name: Prep tests
+      #   env:
+      #     TARGET: runce
+      #   run: python tests/remove_emojis.py
+
+      # - name: Run tests/test_windows.py
+      #   run: python tests/test_windows.py
+      #   continue-on-error: true
+
+      # - name: Run tests/test_utils.py
+      #   run: python tests/test_utils.py
+      #   continue-on-error: true
+
+      # - name: Run tests/test_cli.py
+      #   run: python tests/test_cli.py
+      #   continue-on-error: true
+
+      - name: Run tests
+        run: python -m pytest

pyproject.toml

@@ -12,7 +12,6 @@ description = "Ensure only one instance of your process runs at a time."
 authors = [{name = "Jet-Logic"}]
 readme = "README.md"
 requires-python = ">=3.8"
-license = {text = "MIT"}
 classifiers = [
     "Operating System :: POSIX :: Linux",
     "Development Status :: 4 - Beta",

runce/__init__.py

@@ -3,8 +3,6 @@
 
 A lightweight process manager that guarantees single execution.
 Ensures each command runs exactly once per unique ID.
-
-License: GPL-3.0
 """
 
-__version__ = "0.3.0"
+__version__ = "0.4.0"

runce/cli.py

@@ -1,14 +1,12 @@
-import signal
 from argparse import ArgumentParser
 from shlex import join
 from sys import stderr, stdout
-from os import getpgid, killpg
 from shutil import copyfileobj
 from subprocess import Popen, PIPE, run
+import sys
 from typing import Dict, Any
-from .utils import check_pid
+from .utils import check_pid, filesizepu, kill_pid, tail_bytes, tail_file
 from .main import Main, flag, arg
-
 from .procdb import ProcessDB as Manager
 
 # from .spawn import Spawn as Manager
@@ -17,7 +15,7 @@
 class FormatDict(dict):
     def __missing__(self, key: str) -> str:
         if key == "pid?":
-            return f'{self["pid"]}{"" if check_pid(self["pid"]) else "?👻"}'
+            return f'{self["pid"]}{"" if check_pid(self["pid"]) else "?"}'
         elif key == "elapsed":
             import time
 
@@ -27,7 +25,7 @@ def __missing__(self, key: str) -> str:
                 return self["cmd"]
             return join(self["cmd"])
         elif key == "pid_status":
-            return "✅ Live" if check_pid(self["pid"]) else "❌ Gone"
+            return "Running" if check_pid(self["pid"]) else "Stopped"
         raise KeyError(f"No {key!r}")
 
 
@@ -40,11 +38,19 @@ def fn(x: Dict[str, Any]) -> str:
 
 
 def no_record(name):
-    print(f"🤷‍ No record of {name!r}")
+    try:
+        print("🤷‍ ", end="")
+    except UnicodeEncodeError:
+        pass
+    print(f"No record of {name!r}")
 
 
 def ambiguous(name):
-    print(f"⁉️ {name!r} is ambiguous")
+    try:
+        print("⁉️ ", end="")
+    except UnicodeEncodeError:
+        pass
+    print(f"{name!r} is ambiguous")
 
 
 class Clean(Main):
@@ -61,7 +67,11 @@ def start(self) -> None:
         for d in sp.find_names(self.ids, ambiguous, no_record):
             if check_pid(d["pid"]):
                 continue
-            print(f"🧹 Cleaning {d['pid']} {d['name']}")
+            try:
+                print("🧹 ", end="")
+            except UnicodeEncodeError:
+                pass
+            print(f"Cleaning {d['pid']} {d['name']}")
             sp.drop(d)
 
 
@@ -81,6 +91,7 @@ def init_argparse(self, argp: ArgumentParser) -> None:
 
     def start(self) -> None:
         f = format_prep(self.format)
+        e = ["✅", "❌"]
         for d in Manager().find_names(self.ids, ambiguous, no_record):
             print(f(d))
 
@@ -91,70 +102,92 @@ class Kill(Main):
     ids: list[str] = arg("ID", "run ids", nargs="+")
     dry_run: bool = flag("dry-run", "dry run (don't actually kill)", default=False)
     remove: bool = flag("remove", "remove entry after killing", default=False)
+    group: bool = flag("group", "kill process group", default=False)
+    signal: str = flag("signal", "send signal", default=None)
 
     def init_argparse(self, argp: ArgumentParser) -> None:
         argp.description = "Kill the process of a run id"
         return super().init_argparse(argp)
 
     def start(self) -> None:
+        _errdef = ["❌", "Error"]
+        _noproc = ["👻", "No process"]
+        _killed = ["💀", "Killed"]
+        signal = int(self.signal) if self.signal else None
         sp = Manager()
         if self.ids:
             for x in sp.find_names(self.ids, ambiguous, no_record):
-                pref = "❌ Error"
+                s = _errdef
+                if self.dry_run:
+                    s = _killed
+                else:
+                    if check_pid(x["pid"]):
+                        if kill_pid(x["pid"], process_group=self.group):
+                            s = _killed
+                    else:
+                        s = _noproc
                 try:
-                    pgid = getpgid(x["pid"])
-                    if not self.dry_run:
-                        killpg(pgid, signal.SIGTERM)
-                    pref = "💀 Killed"
-                except ProcessLookupError:
-                    pref = "👻 No process"
-                finally:
-                    print(f'{pref} PID={x["pid"]} {x["name"]!r}')
-                    if not self.dry_run and self.remove:
-                        sp.drop(x)
+                    print(f"{s[0]} ", end="")
+                except UnicodeEncodeError:
+                    pass
+                print(f'{s[1]} PID={x["pid"]} {x["name"]!r}')
+                if not self.dry_run and self.remove:
+                    sp.drop(x)
+
+
+def _tail(n: float, u="", out="", tab=None):
+    if u:
+        stdout.buffer.write(tail_bytes(out, int(n)))
+    elif n > 0:
+        if sys.platform.startswith("win"):
+            cmd = [
+                "powershell",
+                "-c",
+                f"Get-Content -Tail {int(n)} '{out}'",
+            ]
+        else:
+            cmd = ["tail", "-n", str(int(n)), out]
+        if tab:
+            with Popen(cmd, stdout=PIPE).stdout as o:
+                for line in o:
+                    stdout.buffer.write(b"\t" + line)
+        else:
+            run(cmd)
 
 
 class Tail(Main):
     """Tail process output."""
 
     ids: list[str] = arg("ID", "run ids", nargs="*")
     format: str = flag("header", "header format")
-    lines: int = flag("n", "lines", "how many lines")
+    lines: str = flag("n", "lines", "how many lines or bytes")
     existing: bool = flag(
         "x", "only-existing", "only show existing processes", default=False
     )
     tab: bool = flag("t", "tab", "prefix tab space", default=False)
     err: bool = flag("e", "err", "output stderr", default=False)
-    p_open: str = "📜 "
-    p_close: str = ""
+    p_open: str = "=== "
+    p_close: str = " ==="
 
     def start(self) -> None:
+        import sys
+
         if self.format == "no":
             hf = None
         else:
             hf = format_prep(self.format or r"{pid?}: {name}")
-        lines = self.lines or 10
+        n, u = filesizepu(self.lines or "10")
         j = 0
         out = "err" if self.err else "out"
 
         for x in Manager().find_names(self.ids, ambiguous, no_record):
             if self.existing and not check_pid(x["pid"]):
                 continue
 
-            j > 1 and lines > 0 and print()
+            j > 1 and n > 0 and print()
             if hf:
                 print(f"{self.p_open}{hf(x)}{self.p_close}", flush=True)
-
-            if lines > 0:
-                # TODO: pythonify
-                cmd = ["tail", "-n", str(lines), x[out]]
-                if self.tab:
-                    with Popen(cmd, stdout=PIPE).stdout as o:
-                        for line in o:
-                            stdout.buffer.write(b"\t" + line)
-                else:
-                    run(cmd)
-                stdout.flush()
+            _tail(n, u, x[out], self.tab)
             j += 1
 
 
@@ -165,11 +198,11 @@ class Run(Main):
     tail: int = flag("t", "tail", "tail the output with n lines", default=0)
     run_id: str = flag("id", "Unique run identifier", default="")
     cwd: str = flag("Working directory for the command")
-    tail: int = flag(
+    tail: str = flag(
         "t",
         "tail",
         "Tail the output (n lines). Use `-t -1` to print the entire output",
-        default=0,
+        # default=0,
     )
     overwrite: bool = flag("overwrite", "Overwrite existing entry", default=False)
     cmd_after: str = flag("run-after", "Run command after", metavar="command")
@@ -183,24 +216,25 @@ def start(self) -> None:
         # Check for existing process first
         e = sp.find_name(name) if name else None
         if e:
-            hf = format_prep(r"🚨 Found: PID={pid} ({pid_status}) {name}")
-            print(hf(e), file=stderr)
+            s = ["🚨", r"Found: PID={pid} ({pid_status}) {name}"]
         else:
             # Start new process
             e = sp.spawn(
                 args, name, overwrite=self.overwrite, cwd=self.cwd, split=self.split
             )
-            hf = format_prep("🚀 Started: PID={pid} ({pid_status}) {name}")
-            print(hf(e), file=stderr)
+            s = ["🚀", r"Started: PID={pid} ({pid_status}) {name}"]
         assert e
+        try:
+            print(f"{s[0]} ", end="", file=stderr)
+        except UnicodeEncodeError:
+            pass
+        hf = format_prep(s[1])
+        print(hf(e), file=stderr, flush=True)
 
         # Handle tail output
         if self.tail:
-            if self.tail < 0:
-                with open(e["out"], "rb") as f:
-                    copyfileobj(f, stdout.buffer)
-            elif self.tail > 0:
-                run(["tail", "-n", str(self.tail), e["out"]])
+            n, u = filesizepu(self.tail)
+            _tail(n, u, e["out"])
 
         # Run post-command if specified
         if self.cmd_after:
@@ -228,7 +262,7 @@ def start(self) -> None:
         else:
             f = "{pid_status} {elapsed} {pid}\t{name}, {command}"
             print("Status  Elapsed  PID\tName, Command")
-            print("─────── ──────── ────── ────────────")
+            print("------- -------- ------ ------------")
         fp = format_prep(f)
         for d in Manager().all():
             print(fp(d))