chore(ci-image): fix tld in ci-image workflow + additional workflow improvements #137
@@ -2,7 +2,7 @@ name: Rebuild CI Image
|
||||
|
||||
on:
|
||||
schedule:
|
||||
- cron: '0 6 * * 1'
|
||||
- cron: "0 6 * * 1"
|
||||
workflow_dispatch:
|
||||
|
||||
jobs:
|
||||
@@ -11,7 +11,7 @@ jobs:
|
||||
container:
|
||||
image: node:26-alpine
|
||||
env:
|
||||
IMAGE: git.aramjonghu.nl/aramjonghu/zshell-ci:latest
|
||||
IMAGE: git.aramjonghu.dev/aramjonghu/zshell-ci:latest
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
@@ -21,7 +21,7 @@ jobs:
|
||||
run: apk add --no-cache docker-cli
|
||||
|
||||
- name: Login to registry
|
||||
run: echo "${{ secrets.REGISTRY_TOKEN }}" | docker login git.aramjonghu.nl --username aramjonghu --password-stdin
|
||||
run: echo "${{ secrets.REGISTRY_TOKEN }}" | docker login git.aramjonghu.dev --username aramjonghu --password-stdin
|
||||
|
||||
- name: Build image
|
||||
run: docker build -t "$IMAGE" -f ci/Dockerfile .
|
||||
|
||||
@@ -4,10 +4,43 @@ on:
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
fmt:
|
||||
runs-on: alpine
|
||||
container:
|
||||
image: git.aramjonghu.dev/aramjonghu/zshell-ci:latest
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Format check
|
||||
run: |
|
||||
find . \( -name '*.cpp' -o -name '*.h' -o -name '*.hpp' \) \
|
||||
-not -path './build/*' \
|
||||
-exec clang-format -i --style=file {} +
|
||||
git diff --exit-code && echo "clang-format: passed"
|
||||
|
||||
build:
|
||||
runs-on: alpine
|
||||
container:
|
||||
image: git.aramjonghu.nl/aramjonghu/zshell-ci:latest
|
||||
image: git.aramjonghu.dev/aramjonghu/zshell-ci:latest
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
with:
|
||||
fetch-depth: 0
|
||||
|
||||
- name: Configure
|
||||
run: cmake -B build -G Ninja -DENABLE_MODULES=plugin -DCMAKE_BUILD_TYPE=Release
|
||||
|
||||
- name: Build
|
||||
run: ninja -C build
|
||||
|
||||
clang-tidy:
|
||||
runs-on: alpine
|
||||
container:
|
||||
image: git.aramjonghu.dev/aramjonghu/zshell-ci:latest
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
|
||||
@@ -1,10 +1,10 @@
|
||||
name: Lint & Format (JS/TS)
|
||||
name: JS/TS
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
lint-format:
|
||||
fmt:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
|
||||
@@ -18,7 +18,6 @@ jobs:
|
||||
git
|
||||
|
||||
- name: Prettier
|
||||
continue-on-error: true
|
||||
run: |
|
||||
if [ -n "$(find . \( -iname "*.js" -o -iname "*.jsx" -o -iname "*.ts" -o -iname "*.tsx" -o -iname "*.mjs" -o -iname "*.cjs" \) -print -quit)" ]; then
|
||||
npx --yes prettier --check "**/*.{js,jsx,ts,tsx,mjs,cjs}" --ignore-path .prettierignore
|
||||
@@ -26,6 +25,19 @@ jobs:
|
||||
echo "No JS/TS files found"
|
||||
fi
|
||||
|
||||
lint:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Install tools
|
||||
run: |
|
||||
apk add --no-cache \
|
||||
git
|
||||
|
||||
- name: ESLint
|
||||
run: |
|
||||
if [ -n "$(find . \( -iname "*.js" -o -iname "*.jsx" -o -iname "*.ts" -o -iname "*.tsx" -o -iname "*.mjs" -o -iname "*.cjs" \) -print -quit)" ]; then
|
||||
@@ -1,85 +0,0 @@
|
||||
name: Lint & Format (Rust)
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
lint-format:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
env:
|
||||
CARGO_HOME: ${{ github.workspace }}/.cargo
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Cache cargo packages
|
||||
uses: actions/cache@v4
|
||||
env:
|
||||
cache-name: cache-cargo-packages
|
||||
with:
|
||||
path: |
|
||||
.cargo/registry
|
||||
.cargo/git
|
||||
target
|
||||
key: rust-${{ runner.os }}-build-${{ env.cache-name }}-${{ hashFiles('**/Cargo.lock') }}
|
||||
restore-keys: |
|
||||
rust-${{ runner.os }}-build-${{ env.cache-name }}-
|
||||
rust-${{ runner.os }}-build-
|
||||
rust-
|
||||
|
||||
- name: Install tools
|
||||
run: |
|
||||
apk add --no-cache \
|
||||
git \
|
||||
cargo \
|
||||
rust \
|
||||
rustfmt \
|
||||
rust-clippy
|
||||
|
||||
- id: format-check
|
||||
name: Format check
|
||||
continue-on-error: true
|
||||
run: |
|
||||
if [ -n "$(find . -name "Cargo.toml" -print -quit)" ]; then
|
||||
status=0
|
||||
for manifest in $(find . -name "Cargo.toml"); do
|
||||
cargo fmt --manifest-path "$manifest" --check && \
|
||||
echo "$manifest: formatting OK" || \
|
||||
{ echo "$manifest: needs formatting"; status=1; }
|
||||
done
|
||||
exit $status
|
||||
elif [ -n "$(find . -name "*.rs" -print -quit)" ]; then
|
||||
echo "Rust files found but no Cargo.toml"
|
||||
exit 1
|
||||
else
|
||||
echo "No Rust project found"
|
||||
fi
|
||||
|
||||
- id: clippy
|
||||
name: Clippy
|
||||
run: |
|
||||
if [ -n "$(find . -name "Cargo.toml" -print -quit)" ]; then
|
||||
status=0
|
||||
for manifest in $(find . -name "Cargo.toml"); do
|
||||
cargo clippy --manifest-path "$manifest" --all-targets --all-features -- -D warnings && \
|
||||
echo "$manifest: Clippy passed" || \
|
||||
{ echo "$manifest: Clippy failed"; status=1; }
|
||||
done
|
||||
exit $status
|
||||
elif [ -n "$(find . -name "*.rs" -print -quit)" ]; then
|
||||
echo "Rust files found but no Cargo.toml"
|
||||
exit 1
|
||||
else
|
||||
echo "No Rust project found"
|
||||
fi
|
||||
|
||||
- name: Check results
|
||||
if: always()
|
||||
run: |
|
||||
if [ "${{ steps.format-check.outcome }}" = "failure" ] || [ "${{ steps.clippy.outcome }}" = "failure" ]; then
|
||||
echo "One or more checks failed"
|
||||
exit 1
|
||||
fi
|
||||
echo "All checks passed"
|
||||
@@ -4,7 +4,7 @@ on:
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
lint-format:
|
||||
fmt:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
|
||||
@@ -23,11 +23,28 @@ jobs:
|
||||
pip install --no-cache-dir ruff
|
||||
|
||||
- name: Format check
|
||||
continue-on-error: true
|
||||
run: |
|
||||
. .venv/bin/activate
|
||||
ruff format --check .
|
||||
|
||||
lint:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Install tools
|
||||
run: |
|
||||
apk add --no-cache \
|
||||
git \
|
||||
python3 \
|
||||
py3-pip
|
||||
python3 -m venv .venv
|
||||
. .venv/bin/activate
|
||||
pip install --no-cache-dir ruff
|
||||
|
||||
- name: Lint
|
||||
run: |
|
||||
. .venv/bin/activate
|
||||
@@ -63,3 +80,30 @@ jobs:
|
||||
. .venv/bin/activate
|
||||
cd cli
|
||||
python -m pytest tests/ -v
|
||||
|
||||
buildcheck:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Install tools
|
||||
run: |
|
||||
apk add --no-cache \
|
||||
git \
|
||||
python3 \
|
||||
py3-pip \
|
||||
build-base \
|
||||
python3-dev \
|
||||
gcc \
|
||||
g++
|
||||
python3 -m venv .venv
|
||||
. .venv/bin/activate
|
||||
pip install --no-cache-dir nuitka
|
||||
|
||||
- name: Nuitka module check
|
||||
run: |
|
||||
. .venv/bin/activate
|
||||
nuitka --module --include-package=zshell cli/src/zshell/
|
||||
@@ -0,0 +1,152 @@
|
||||
name: Rust
|
||||
|
||||
on:
|
||||
pull_request:
|
||||
|
||||
jobs:
|
||||
build:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
env:
|
||||
CARGO_HOME: ${{ github.workspace }}/.cargo
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Cache cargo packages
|
||||
uses: actions/cache@v4
|
||||
env:
|
||||
cache-name: cache-cargo-packages
|
||||
with:
|
||||
path: |
|
||||
.cargo/registry
|
||||
.cargo/git
|
||||
target
|
||||
key: rust-build-${{ runner.os }}-${{ env.cache-name }}-${{ hashFiles('**/Cargo.lock') }}
|
||||
restore-keys: |
|
||||
rust-${{ runner.os }}-build-${{ env.cache-name }}-
|
||||
rust-${{ runner.os }}-build-
|
||||
rust-
|
||||
|
||||
- name: Install tools
|
||||
run: |
|
||||
apk add --no-cache \
|
||||
git \
|
||||
cargo \
|
||||
rust
|
||||
|
||||
- name: Cargo check
|
||||
run: |
|
||||
if [ -n "$(find . -name "Cargo.toml" -print -quit)" ]; then
|
||||
for manifest in $(find . -name "Cargo.toml"); do
|
||||
cargo check --manifest-path "$manifest" && \
|
||||
echo "$manifest: check passed" || \
|
||||
{ echo "$manifest: check failed"; exit 1; }
|
||||
done
|
||||
elif [ -n "$(find . -name "*.rs" -print -quit)" ]; then
|
||||
echo "Rust files found but no Cargo.toml"
|
||||
exit 1
|
||||
else
|
||||
echo "No Rust project found"
|
||||
fi
|
||||
|
||||
fmt:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
env:
|
||||
CARGO_HOME: ${{ github.workspace }}/.cargo
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Cache cargo packages
|
||||
uses: actions/cache@v4
|
||||
env:
|
||||
cache-name: cache-cargo-packages
|
||||
with:
|
||||
path: |
|
||||
.cargo/registry
|
||||
.cargo/git
|
||||
target
|
||||
key: rust-fmt-${{ runner.os }}-${{ env.cache-name }}-${{ hashFiles('**/Cargo.lock') }}
|
||||
restore-keys: |
|
||||
rust-${{ runner.os }}-build-${{ env.cache-name }}-
|
||||
rust-${{ runner.os }}-build-
|
||||
rust-
|
||||
|
||||
- name: Install tools
|
||||
run: |
|
||||
apk add --no-cache \
|
||||
git \
|
||||
cargo \
|
||||
rust \
|
||||
rustfmt
|
||||
|
||||
- name: Format check
|
||||
run: |
|
||||
if [ -n "$(find . -name "Cargo.toml" -print -quit)" ]; then
|
||||
status=0
|
||||
for manifest in $(find . -name "Cargo.toml"); do
|
||||
cargo fmt --manifest-path "$manifest" --check && \
|
||||
echo "$manifest: formatting OK" || \
|
||||
{ echo "$manifest: needs formatting"; status=1; }
|
||||
done
|
||||
exit $status
|
||||
elif [ -n "$(find . -name "*.rs" -print -quit)" ]; then
|
||||
echo "Rust files found but no Cargo.toml"
|
||||
exit 1
|
||||
else
|
||||
echo "No Rust project found"
|
||||
fi
|
||||
|
||||
clippy:
|
||||
runs-on: alpine
|
||||
container: node:26-alpine
|
||||
env:
|
||||
CARGO_HOME: ${{ github.workspace }}/.cargo
|
||||
|
||||
steps:
|
||||
- name: Checkout
|
||||
uses: actions/checkout@v4
|
||||
|
||||
- name: Cache cargo packages
|
||||
uses: actions/cache@v4
|
||||
env:
|
||||
cache-name: cache-cargo-packages
|
||||
with:
|
||||
path: |
|
||||
.cargo/registry
|
||||
.cargo/git
|
||||
target
|
||||
key: rust-clippy-${{ runner.os }}-${{ env.cache-name }}-${{ hashFiles('**/Cargo.lock') }}
|
||||
restore-keys: |
|
||||
rust-${{ runner.os }}-build-${{ env.cache-name }}-
|
||||
rust-${{ runner.os }}-build-
|
||||
rust-
|
||||
|
||||
- name: Install tools
|
||||
run: |
|
||||
apk add --no-cache \
|
||||
git \
|
||||
cargo \
|
||||
rust \
|
||||
rust-clippy
|
||||
|
||||
- name: Clippy
|
||||
run: |
|
||||
if [ -n "$(find . -name "Cargo.toml" -print -quit)" ]; then
|
||||
status=0
|
||||
for manifest in $(find . -name "Cargo.toml"); do
|
||||
cargo clippy --manifest-path "$manifest" --all-targets --all-features -- -D warnings && \
|
||||
echo "$manifest: Clippy passed" || \
|
||||
{ echo "$manifest: Clippy failed"; status=1; }
|
||||
done
|
||||
exit $status
|
||||
elif [ -n "$(find . -name "*.rs" -print -quit)" ]; then
|
||||
echo "Rust files found but no Cargo.toml"
|
||||
exit 1
|
||||
else
|
||||
echo "No Rust project found"
|
||||
fi
|
||||
+51
-33
@@ -1,12 +1,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import typer
|
||||
from typer._completion_shared import install, _get_shell_name
|
||||
from typer._completion_classes import completion_init
|
||||
from zshell.subcommands import shell, scheme, screenshot, wallpaper, record
|
||||
from typer._completion_shared import _get_shell_name, install
|
||||
|
||||
from zshell.subcommands import record, scheme, screenshot, shell, wallpaper
|
||||
|
||||
app = typer.Typer(name="zshell-cli", add_completion=False)
|
||||
|
||||
@@ -18,40 +20,56 @@ app.add_typer(record.app, name="record")
|
||||
|
||||
|
||||
def _completion_installed() -> bool:
|
||||
shell = _get_shell_name()
|
||||
match shell:
|
||||
case "zsh":
|
||||
return (Path.home() / ".zfunc" / "_zshell-cli").exists()
|
||||
case "bash":
|
||||
return (Path.home() / ".bash_completions" / "zshell-cli.sh").exists()
|
||||
case "fish":
|
||||
return (Path.home() / ".config" / "fish" / "completions" / "zshell-cli.fish").exists()
|
||||
return False
|
||||
shell = _get_shell_name()
|
||||
match shell:
|
||||
case "zsh":
|
||||
return (Path.home() / ".zfunc" / "_zshell-cli").exists()
|
||||
case "bash":
|
||||
return (
|
||||
Path.home() / ".bash_completions" / "zshell-cli.sh"
|
||||
).exists()
|
||||
case "fish":
|
||||
return (
|
||||
Path.home()
|
||||
/ ".config"
|
||||
/ "fish"
|
||||
/ "completions"
|
||||
/ "zshell-cli.fish"
|
||||
).exists()
|
||||
return False
|
||||
|
||||
|
||||
def _install_completion() -> None:
|
||||
if _completion_installed():
|
||||
print("zshell-cli: Shell completion already installed.")
|
||||
sys.exit(0)
|
||||
shell = _get_shell_name()
|
||||
if shell is None:
|
||||
print("zshell-cli: Unable to detect shell type.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
try:
|
||||
_, path = install(prog_name="zshell-cli")
|
||||
print(f"zshell-cli: Shell completion installed ({shell}: {path})")
|
||||
print("zshell-cli: Restart your shell or source the file to enable tab-completion.")
|
||||
except Exception as e:
|
||||
print(f"zshell-cli: Failed to install shell completion: {e}", file=sys.stderr)
|
||||
raise typer.Exit(code=1)
|
||||
if _completion_installed():
|
||||
print("zshell-cli: Shell completion already installed.")
|
||||
sys.exit(0)
|
||||
shell = _get_shell_name()
|
||||
if shell is None:
|
||||
print("zshell-cli: Unable to detect shell type.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
try:
|
||||
_, path = install(prog_name="zshell-cli")
|
||||
print(f"zshell-cli: Shell completion installed ({shell}: {path})")
|
||||
print(
|
||||
"zshell-cli: Restart your shell or source the file to enable tab-completion."
|
||||
)
|
||||
except Exception as e:
|
||||
print(
|
||||
f"zshell-cli: Failed to install shell completion: {e}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
raise typer.Exit(code=1) from None
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if "--install-autocomplete" in sys.argv:
|
||||
_install_completion()
|
||||
return
|
||||
if "_ZSHELL_CLI_COMPLETE" in os.environ:
|
||||
completion_init()
|
||||
if sys.stdout.isatty() and not _completion_installed():
|
||||
print("zshell-cli: Tip: run with --install-autocomplete for tab completion.", file=sys.stderr)
|
||||
app()
|
||||
if "--install-autocomplete" in sys.argv:
|
||||
_install_completion()
|
||||
return
|
||||
if "_ZSHELL_CLI_COMPLETE" in os.environ:
|
||||
completion_init()
|
||||
if sys.stdout.isatty() and not _completion_installed():
|
||||
print(
|
||||
"zshell-cli: Tip: run with --install-autocomplete for tab completion.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
app()
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from zshell import main
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
main()
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import os
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import typer
|
||||
|
||||
@@ -18,193 +18,253 @@ TEMP_RECORDING = STATE_DIR / "recording.mp4"
|
||||
REPLAY_RECORDING = STATE_DIR / "replay.mp4"
|
||||
NOTIF_ID_FILE = STATE_DIR / "notifid.txt"
|
||||
|
||||
RECORDINGS_DIR = os.getenv("ZSHELL_RECORDINGS_DIR", str(Path(HOME) / "Videos/Recordings"))
|
||||
RECORDINGS_DIR = os.getenv(
|
||||
"ZSHELL_RECORDINGS_DIR", str(Path(HOME) / "Videos/Recordings")
|
||||
)
|
||||
|
||||
|
||||
def _read_extra_args() -> list[str]:
|
||||
try:
|
||||
if CONFIG.is_file():
|
||||
data = json.loads(CONFIG.read_text())
|
||||
return data.get("record", {}).get("extraArgs", [])
|
||||
except Exception:
|
||||
pass
|
||||
return []
|
||||
try:
|
||||
if CONFIG.is_file():
|
||||
data = json.loads(CONFIG.read_text())
|
||||
return data.get("record", {}).get("extraArgs", [])
|
||||
except Exception:
|
||||
pass
|
||||
return []
|
||||
|
||||
|
||||
def _is_recording() -> bool:
|
||||
return subprocess.run(["pidof", RECORDER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0
|
||||
return (
|
||||
subprocess.run(
|
||||
["pidof", RECORDER],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
).returncode
|
||||
== 0
|
||||
)
|
||||
|
||||
|
||||
def _notify(summary: str, body: str = "", actions: list | None = None, timeout: int = 5000) -> Optional[int]:
|
||||
args = ["notify-send", summary, body, "-t", str(timeout), "-p"]
|
||||
if actions:
|
||||
for action in actions:
|
||||
args.extend(["-A", action])
|
||||
try:
|
||||
proc = subprocess.run(args, capture_output=True, text=True)
|
||||
return int(proc.stdout.strip()) if proc.stdout.strip().isdigit() else None
|
||||
except Exception:
|
||||
return None
|
||||
def _notify(
|
||||
summary: str,
|
||||
body: str = "",
|
||||
actions: list | None = None,
|
||||
timeout: int = 5000,
|
||||
) -> int | None:
|
||||
args = ["notify-send", summary, body, "-t", str(timeout), "-p"]
|
||||
if actions:
|
||||
for action in actions:
|
||||
args.extend(["-A", action])
|
||||
try:
|
||||
proc = subprocess.run(args, capture_output=True, text=True)
|
||||
return (
|
||||
int(proc.stdout.strip()) if proc.stdout.strip().isdigit() else None
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _close_notification(notif_id: int):
|
||||
subprocess.run(["notify-send", "--close", str(notif_id)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
subprocess.run(
|
||||
["notify-send", "--close", str(notif_id)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
|
||||
def _get_monitors() -> list[dict]:
|
||||
try:
|
||||
res = subprocess.run(["hyprctl", "monitors", "-j"], capture_output=True, text=True)
|
||||
return json.loads(res.stdout)
|
||||
except Exception:
|
||||
return []
|
||||
try:
|
||||
res = subprocess.run(
|
||||
["hyprctl", "monitors", "-j"], capture_output=True, text=True
|
||||
)
|
||||
return json.loads(res.stdout)
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def _focused_monitor_name() -> Optional[str]:
|
||||
for m in _get_monitors():
|
||||
if m.get("focused"):
|
||||
return m["name"]
|
||||
return None
|
||||
def _focused_monitor_name() -> str | None:
|
||||
for m in _get_monitors():
|
||||
if m.get("focused"):
|
||||
return m["name"]
|
||||
return None
|
||||
|
||||
|
||||
def _monitors_intersecting_region(x: int, y: int, w: int, h: int) -> list[dict]:
|
||||
region = (x, y, x + w, y + h)
|
||||
intersecting = []
|
||||
for m in _get_monitors():
|
||||
mx, my, mw, mh = m["x"], m["y"], m["width"], m["height"]
|
||||
if not (region[2] <= mx or region[0] >= mx + mw or region[3] <= my or region[1] >= my + mh):
|
||||
intersecting.append(m)
|
||||
return intersecting
|
||||
region = (x, y, x + w, y + h)
|
||||
intersecting = []
|
||||
for m in _get_monitors():
|
||||
mx, my, mw, mh = m["x"], m["y"], m["width"], m["height"]
|
||||
if not (
|
||||
region[2] <= mx
|
||||
or region[0] >= mx + mw
|
||||
or region[3] <= my
|
||||
or region[1] >= my + mh
|
||||
):
|
||||
intersecting.append(m)
|
||||
return intersecting
|
||||
|
||||
|
||||
def _highest_refresh(monitors: list[dict]) -> float:
|
||||
return max((m["refreshRate"] for m in monitors), default=60.0)
|
||||
return max((m["refreshRate"] for m in monitors), default=60.0)
|
||||
|
||||
|
||||
def _slurp_region() -> Optional[str]:
|
||||
try:
|
||||
return subprocess.check_output(["slurp", "-f", "%wx%h+%x+%y"], text=True).strip()
|
||||
except subprocess.CalledProcessError:
|
||||
return None
|
||||
def _slurp_region() -> str | None:
|
||||
try:
|
||||
return subprocess.check_output(
|
||||
["slurp", "-f", "%wx%h+%x+%y"], text=True
|
||||
).strip()
|
||||
except subprocess.CalledProcessError:
|
||||
return None
|
||||
|
||||
|
||||
def _parse_geometry(geometry: str) -> Optional[tuple[int, int, int, int]]:
|
||||
import re
|
||||
def _parse_geometry(geometry: str) -> tuple[int, int, int, int] | None:
|
||||
import re
|
||||
|
||||
match = re.match(r"(\d+)x(\d+)\+(\d+)\+(\d+)", geometry)
|
||||
if match:
|
||||
return int(match.group(3)), int(match.group(4)), int(match.group(1)), int(match.group(2))
|
||||
return None
|
||||
match = re.match(r"(\d+)x(\d+)\+(\d+)\+(\d+)", geometry)
|
||||
if match:
|
||||
return (
|
||||
int(match.group(3)),
|
||||
int(match.group(4)),
|
||||
int(match.group(1)),
|
||||
int(match.group(2)),
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def start_recording(region: Optional[str], sound: bool):
|
||||
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
cmd = [RECORDER]
|
||||
extra_args = _read_extra_args()
|
||||
def start_recording(region: str | None, sound: bool):
|
||||
STATE_DIR.mkdir(parents=True, exist_ok=True)
|
||||
cmd = [RECORDER]
|
||||
extra_args = _read_extra_args()
|
||||
|
||||
if region:
|
||||
if region.lower() == "slurp" or not region:
|
||||
geometry = _slurp_region()
|
||||
if not geometry:
|
||||
typer.echo("Region selection cancelled.")
|
||||
raise typer.Abort()
|
||||
else:
|
||||
geometry = region
|
||||
if region:
|
||||
if region.lower() == "slurp" or not region:
|
||||
geometry = _slurp_region()
|
||||
if not geometry:
|
||||
typer.echo("Region selection cancelled.")
|
||||
raise typer.Abort()
|
||||
else:
|
||||
geometry = region
|
||||
|
||||
parsed = _parse_geometry(geometry)
|
||||
if not parsed:
|
||||
typer.echo("Invalid geometry format.")
|
||||
raise typer.Abort()
|
||||
x, y, w, h = parsed
|
||||
parsed = _parse_geometry(geometry)
|
||||
if not parsed:
|
||||
typer.echo("Invalid geometry format.")
|
||||
raise typer.Abort()
|
||||
x, y, w, h = parsed
|
||||
|
||||
monitors = _monitors_intersecting_region(x, y, w, h)
|
||||
framerate = _highest_refresh(monitors)
|
||||
cmd.extend(["-w", "region", "-region", geometry, "-f", str(int(framerate))])
|
||||
monitors = _monitors_intersecting_region(x, y, w, h)
|
||||
framerate = _highest_refresh(monitors)
|
||||
cmd.extend(
|
||||
["-w", "region", "-region", geometry, "-f", str(int(framerate))]
|
||||
)
|
||||
|
||||
else:
|
||||
monitor_name = _focused_monitor_name()
|
||||
if not monitor_name:
|
||||
typer.echo("No focused monitor found.")
|
||||
raise typer.Abort()
|
||||
else:
|
||||
monitor_name = _focused_monitor_name()
|
||||
if not monitor_name:
|
||||
typer.echo("No focused monitor found.")
|
||||
raise typer.Abort()
|
||||
|
||||
monitors = _get_monitors()
|
||||
mon = next((m for m in monitors if m["name"] == monitor_name), None)
|
||||
rate = int(mon["refreshRate"]) if mon else 60
|
||||
cmd.extend(["-w", monitor_name, "-f", str(rate)])
|
||||
monitors = _get_monitors()
|
||||
mon = next((m for m in monitors if m["name"] == monitor_name), None)
|
||||
rate = int(mon["refreshRate"]) if mon else 60
|
||||
cmd.extend(["-w", monitor_name, "-f", str(rate)])
|
||||
|
||||
if sound:
|
||||
cmd.extend(["-a", "default_output"])
|
||||
if sound:
|
||||
cmd.extend(["-a", "default_output"])
|
||||
|
||||
cmd.extend(extra_args)
|
||||
cmd.extend(["-o", str(TEMP_RECORDING)])
|
||||
cmd.extend(extra_args)
|
||||
cmd.extend(["-o", str(TEMP_RECORDING)])
|
||||
|
||||
subprocess.Popen(cmd, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
subprocess.Popen(
|
||||
cmd,
|
||||
start_new_session=True,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
notif_id = _notify("Recording started", f"Saving to {TEMP_RECORDING}")
|
||||
if notif_id is not None:
|
||||
NOTIF_ID_FILE.write_text(str(notif_id))
|
||||
notif_id = _notify("Recording started", f"Saving to {TEMP_RECORDING}")
|
||||
if notif_id is not None:
|
||||
NOTIF_ID_FILE.write_text(str(notif_id))
|
||||
|
||||
time.sleep(1)
|
||||
if not _is_recording():
|
||||
_notify("Recording failed", "Check gpu-screen-recorder output.", timeout=5000)
|
||||
raise typer.Exit(code=1)
|
||||
time.sleep(1)
|
||||
if not _is_recording():
|
||||
_notify(
|
||||
"Recording failed",
|
||||
"Check gpu-screen-recorder output.",
|
||||
timeout=5000,
|
||||
)
|
||||
raise typer.Exit(code=1)
|
||||
|
||||
|
||||
def stop_recording(clipboard: bool):
|
||||
subprocess.run(["pkill", "-f", RECORDER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
subprocess.run(
|
||||
["pkill", "-f", RECORDER],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
for _ in range(50):
|
||||
if not _is_recording():
|
||||
break
|
||||
time.sleep(0.1)
|
||||
for _ in range(50):
|
||||
if not _is_recording():
|
||||
break
|
||||
time.sleep(0.1)
|
||||
|
||||
dest_dir = Path(RECORDINGS_DIR)
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
timestamp = time.strftime("%Y-%m-%d_%H-%M-%S")
|
||||
final_path = dest_dir / f"recording_{timestamp}.mp4"
|
||||
dest_dir = Path(RECORDINGS_DIR)
|
||||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||||
timestamp = time.strftime("%Y-%m-%d_%H-%M-%S")
|
||||
final_path = dest_dir / f"recording_{timestamp}.mp4"
|
||||
|
||||
if TEMP_RECORDING.exists():
|
||||
TEMP_RECORDING.rename(final_path)
|
||||
if TEMP_RECORDING.exists():
|
||||
TEMP_RECORDING.rename(final_path)
|
||||
|
||||
if NOTIF_ID_FILE.is_file():
|
||||
try:
|
||||
_close_notification(int(NOTIF_ID_FILE.read_text().strip()))
|
||||
except Exception:
|
||||
pass
|
||||
NOTIF_ID_FILE.unlink()
|
||||
if NOTIF_ID_FILE.is_file():
|
||||
with contextlib.suppress(Exception):
|
||||
_close_notification(int(NOTIF_ID_FILE.read_text().strip()))
|
||||
NOTIF_ID_FILE.unlink()
|
||||
|
||||
if clipboard:
|
||||
subprocess.run(
|
||||
["wl-copy", "--type", "text/uri-list", f"file://{final_path}"],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
if clipboard:
|
||||
subprocess.run(
|
||||
["wl-copy", "--type", "text/uri-list", f"file://{final_path}"],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
|
||||
_notify("Recording stopped", f"Saved to {final_path}", timeout=5000)
|
||||
_notify("Recording stopped", f"Saved to {final_path}", timeout=5000)
|
||||
|
||||
|
||||
def toggle_pause():
|
||||
subprocess.run(["pkill", "-USR2", "-f", RECORDER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
||||
typer.echo("Toggled pause.")
|
||||
subprocess.run(
|
||||
["pkill", "-USR2", "-f", RECORDER],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
typer.echo("Toggled pause.")
|
||||
|
||||
|
||||
@app.command()
|
||||
def record(
|
||||
region: Optional[str] = typer.Option(
|
||||
None,
|
||||
"--region",
|
||||
"-r",
|
||||
help="Record a region. Use 'slurp' (or omit value) to select interactively, or give 'WxH+X+Y'.",
|
||||
),
|
||||
sound: bool = typer.Option(False, "--sound", "-s", help="Record audio from default output."),
|
||||
pause: bool = typer.Option(False, "--pause", "-p", help="Toggle pause/resume."),
|
||||
clipboard: bool = typer.Option(False, "--clipboard", "-c", help="Copy the final recording path to clipboard."),
|
||||
region: str | None = typer.Option(
|
||||
None,
|
||||
"--region",
|
||||
"-r",
|
||||
help="Record a region. Use 'slurp' (or omit value) to select interactively, or give 'WxH+X+Y'.",
|
||||
),
|
||||
sound: bool = typer.Option(
|
||||
False, "--sound", "-s", help="Record audio from default output."
|
||||
),
|
||||
pause: bool = typer.Option(
|
||||
False, "--pause", "-p", help="Toggle pause/resume."
|
||||
),
|
||||
clipboard: bool = typer.Option(
|
||||
False,
|
||||
"--clipboard",
|
||||
"-c",
|
||||
help="Copy the final recording path to clipboard.",
|
||||
),
|
||||
):
|
||||
"""Start or stop a screen recording with gpu-screen-recorder."""
|
||||
if pause:
|
||||
toggle_pause()
|
||||
raise typer.Exit()
|
||||
"""Start or stop a screen recording with gpu-screen-recorder."""
|
||||
if pause:
|
||||
toggle_pause()
|
||||
raise typer.Exit()
|
||||
|
||||
if _is_recording():
|
||||
stop_recording(clipboard)
|
||||
else:
|
||||
start_recording(region, sound)
|
||||
if _is_recording():
|
||||
stop_recording(clipboard)
|
||||
else:
|
||||
start_recording(region, sound)
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,4 +1,5 @@
|
||||
import subprocess
|
||||
|
||||
import typer
|
||||
|
||||
args = ["qs", "-c", "zshell"]
|
||||
@@ -8,9 +9,9 @@ app = typer.Typer()
|
||||
|
||||
@app.command()
|
||||
def start():
|
||||
subprocess.run(args + ["ipc"] + ["call"] + ["picker"] + ["open"], check=True)
|
||||
subprocess.run([*args, "ipc", "call", "picker", "open"], check=True)
|
||||
|
||||
|
||||
@app.command()
|
||||
def start_freeze():
|
||||
subprocess.run(args + ["ipc"] + ["call"] + ["picker"] + ["openFreeze"], check=True)
|
||||
subprocess.run([*args, "ipc", "call", "picker", "openFreeze"], check=True)
|
||||
|
||||
@@ -11,76 +11,84 @@ app = typer.Typer()
|
||||
|
||||
@app.command()
|
||||
def kill():
|
||||
result = subprocess.run(args + ["kill"], capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write("No running instance to kill.\n")
|
||||
sys.exit(1)
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
result = subprocess.run([*args, "kill"], capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write("No running instance to kill.\n")
|
||||
sys.exit(1)
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
|
||||
|
||||
def start_instance(no_daemon: bool = False) -> None:
|
||||
result = subprocess.run(args + ["-n"] + ([] if no_daemon else ["-d"]), capture_output=True)
|
||||
stdout = result.stdout.decode().strip()
|
||||
if stdout:
|
||||
if "already running" in stdout.lower():
|
||||
sys.stderr.write(stdout + "\n")
|
||||
sys.exit(1)
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.decode().strip()
|
||||
sys.stderr.write(stderr + "\n")
|
||||
sys.exit(1)
|
||||
result = subprocess.run(
|
||||
args + ["-n"] + ([] if no_daemon else ["-d"]), capture_output=True
|
||||
)
|
||||
stdout = result.stdout.decode().strip()
|
||||
if stdout and "already running" in stdout.lower():
|
||||
sys.stderr.write(stdout + "\n")
|
||||
sys.exit(1)
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.decode().strip()
|
||||
sys.stderr.write(stderr + "\n")
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
@app.command()
|
||||
def start(no_daemon: bool = False):
|
||||
start_instance(no_daemon)
|
||||
start_instance(no_daemon)
|
||||
|
||||
|
||||
@app.command()
|
||||
def restart(no_daemon: bool = False):
|
||||
subprocess.run(args + ["kill"], capture_output=True)
|
||||
deadline = time.monotonic() + 2.5
|
||||
while time.monotonic() < deadline:
|
||||
result = subprocess.run(args + ["kill"], capture_output=True)
|
||||
if result.returncode == 255:
|
||||
break
|
||||
time.sleep(0.25)
|
||||
start_instance(no_daemon=no_daemon)
|
||||
subprocess.run([*args, "kill"], capture_output=True)
|
||||
deadline = time.monotonic() + 2.5
|
||||
while time.monotonic() < deadline:
|
||||
result = subprocess.run([*args, "kill"], capture_output=True)
|
||||
if result.returncode == 255:
|
||||
break
|
||||
time.sleep(0.25)
|
||||
start_instance(no_daemon=no_daemon)
|
||||
|
||||
|
||||
@app.command()
|
||||
def show():
|
||||
result = subprocess.run(args + ["ipc"] + ["show"], capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stdout.write(result.stdout.decode())
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
result = subprocess.run([*args, "ipc", "show"], capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stdout.write(result.stdout.decode())
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
|
||||
|
||||
@app.command()
|
||||
def log():
|
||||
result = subprocess.run(args + ["log"], capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stdout.write(result.stdout.decode())
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
result = subprocess.run([*args, "log"], capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stdout.write(result.stdout.decode())
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
|
||||
|
||||
@app.command()
|
||||
def lock():
|
||||
result = subprocess.run(args + ["ipc"] + ["call"] + ["lock"] + ["lock"], capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
result = subprocess.run(
|
||||
[*args, "ipc", "call", "lock", "lock"], capture_output=True
|
||||
)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
|
||||
|
||||
@app.command()
|
||||
def call(target: str, method: str, method_args: list[str] = typer.Argument(None)):
|
||||
result = subprocess.run(args + ["ipc"] + ["call"] + [target] + [method] + (method_args or []), capture_output=True)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
def call(
|
||||
target: str, method: str, method_args: list[str] = typer.Argument(None)
|
||||
):
|
||||
result = subprocess.run(
|
||||
args + ["ipc"] + ["call"] + [target] + [method] + (method_args or []),
|
||||
capture_output=True,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
sys.exit(1)
|
||||
sys.stderr.write(result.stderr.decode())
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import subprocess
|
||||
import typer
|
||||
|
||||
from typing import Annotated
|
||||
from PIL import Image, ImageFilter
|
||||
from pathlib import Path
|
||||
from typing import Annotated
|
||||
|
||||
import typer
|
||||
from PIL import Image, ImageFilter
|
||||
|
||||
args = ["qs", "-c", "zshell"]
|
||||
|
||||
@@ -12,32 +12,35 @@ app = typer.Typer()
|
||||
|
||||
@app.command()
|
||||
def set(wallpaper: Path):
|
||||
subprocess.run(args + ["ipc"] + ["call"] + ["wallpaper"] + ["set"] + [wallpaper], check=True)
|
||||
subprocess.run(
|
||||
[*args, "ipc", "call", "wallpaper", "set", wallpaper],
|
||||
check=True,
|
||||
)
|
||||
|
||||
|
||||
@app.command()
|
||||
def lockscreen(
|
||||
input_image: Annotated[
|
||||
Path,
|
||||
typer.Option(),
|
||||
],
|
||||
output_path: Annotated[
|
||||
Path,
|
||||
typer.Option(),
|
||||
],
|
||||
blur_amount: int = 20,
|
||||
input_image: Annotated[
|
||||
Path,
|
||||
typer.Option(),
|
||||
],
|
||||
output_path: Annotated[
|
||||
Path,
|
||||
typer.Option(),
|
||||
],
|
||||
blur_amount: int = 20,
|
||||
):
|
||||
img = Image.open(input_image)
|
||||
size = img.size
|
||||
if blur_amount == 0:
|
||||
img.save(output_path, "PNG")
|
||||
return
|
||||
img = Image.open(input_image)
|
||||
size = img.size
|
||||
if blur_amount == 0:
|
||||
img.save(output_path, "PNG")
|
||||
return
|
||||
|
||||
if size[0] < 3840 or size[1] < 2160:
|
||||
img = img.resize((size[0] // 2, size[1] // 2), Image.Resampling.NEAREST)
|
||||
else:
|
||||
img = img.resize((size[0] // 4, size[1] // 4), Image.Resampling.NEAREST)
|
||||
if size[0] < 3840 or size[1] < 2160:
|
||||
img = img.resize((size[0] // 2, size[1] // 2), Image.Resampling.NEAREST)
|
||||
else:
|
||||
img = img.resize((size[0] // 4, size[1] // 4), Image.Resampling.NEAREST)
|
||||
|
||||
img = img.filter(ImageFilter.GaussianBlur(blur_amount))
|
||||
img = img.filter(ImageFilter.GaussianBlur(blur_amount))
|
||||
|
||||
img.save(output_path, "PNG")
|
||||
img.save(output_path, "PNG")
|
||||
|
||||
@@ -10,137 +10,140 @@ ASSETS: Traversable = files("zshell") / "assets" / "schemes"
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SchemeVariant:
|
||||
id: str
|
||||
name: str
|
||||
modes: frozenset[str]
|
||||
accents: tuple[str, ...] = ()
|
||||
id: str
|
||||
name: str
|
||||
modes: frozenset[str]
|
||||
accents: tuple[str, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SchemeMeta:
|
||||
id: str
|
||||
name: str
|
||||
variants: tuple[SchemeVariant, ...]
|
||||
id: str
|
||||
name: str
|
||||
variants: tuple[SchemeVariant, ...]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Palette:
|
||||
colors: dict[str, str]
|
||||
mode: str
|
||||
scheme: str
|
||||
variant: str
|
||||
accent: str | None = None
|
||||
colors: dict[str, str]
|
||||
mode: str
|
||||
scheme: str
|
||||
variant: str
|
||||
accent: str | None = None
|
||||
|
||||
|
||||
def _parse_txt(path: Traversable) -> dict[str, str]:
|
||||
colors: dict[str, str] = {}
|
||||
for line in path.read_text().splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
parts = line.split(None, 1)
|
||||
if len(parts) == 2:
|
||||
key, val = parts
|
||||
colors[key] = f"#{val}" if not val.startswith("#") else val
|
||||
return colors
|
||||
colors: dict[str, str] = {}
|
||||
for line in path.read_text().splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
parts = line.split(None, 1)
|
||||
if len(parts) == 2:
|
||||
key, val = parts
|
||||
colors[key] = f"#{val}" if not val.startswith("#") else val
|
||||
return colors
|
||||
|
||||
|
||||
def _discover_schemes() -> dict[str, SchemeMeta]:
|
||||
schemes: dict[str, SchemeMeta] = {}
|
||||
schemes: dict[str, SchemeMeta] = {}
|
||||
|
||||
for scheme_dir in sorted(ASSETS.iterdir(), key=lambda p: p.name):
|
||||
if not scheme_dir.is_dir() or scheme_dir.name.startswith("."):
|
||||
continue
|
||||
for scheme_dir in sorted(ASSETS.iterdir(), key=lambda p: p.name):
|
||||
if not scheme_dir.is_dir() or scheme_dir.name.startswith("."):
|
||||
continue
|
||||
|
||||
sid = scheme_dir.name
|
||||
display_name = sid.capitalize()
|
||||
sid = scheme_dir.name
|
||||
display_name = sid.capitalize()
|
||||
|
||||
variants: list[SchemeVariant] = []
|
||||
for var_dir in sorted(scheme_dir.iterdir(), key=lambda p: p.name):
|
||||
if not var_dir.is_dir() or var_dir.name.startswith("."):
|
||||
continue
|
||||
variants: list[SchemeVariant] = []
|
||||
for var_dir in sorted(scheme_dir.iterdir(), key=lambda p: p.name):
|
||||
if not var_dir.is_dir() or var_dir.name.startswith("."):
|
||||
continue
|
||||
|
||||
modes: set[str] = set()
|
||||
accents: set[str] = set()
|
||||
modes: set[str] = set()
|
||||
accents: set[str] = set()
|
||||
|
||||
for f in var_dir.iterdir():
|
||||
name = PurePosixPath(f.name)
|
||||
if name.suffix != ".txt":
|
||||
continue
|
||||
stem = name.stem
|
||||
if "-" in stem:
|
||||
maybe_accent, maybe_mode = stem.rsplit("-", 1)
|
||||
if maybe_mode in ("dark", "light"):
|
||||
modes.add(maybe_mode)
|
||||
accents.add(maybe_accent)
|
||||
else:
|
||||
modes.add(stem)
|
||||
else:
|
||||
if stem in ("dark", "light"):
|
||||
modes.add(stem)
|
||||
for f in var_dir.iterdir():
|
||||
name = PurePosixPath(f.name)
|
||||
if name.suffix != ".txt":
|
||||
continue
|
||||
stem = name.stem
|
||||
if "-" in stem:
|
||||
maybe_accent, maybe_mode = stem.rsplit("-", 1)
|
||||
if maybe_mode in ("dark", "light"):
|
||||
modes.add(maybe_mode)
|
||||
accents.add(maybe_accent)
|
||||
else:
|
||||
modes.add(stem)
|
||||
else:
|
||||
if stem in ("dark", "light"):
|
||||
modes.add(stem)
|
||||
|
||||
if modes:
|
||||
vname = var_dir.name.capitalize()
|
||||
variants.append(
|
||||
SchemeVariant(
|
||||
id=var_dir.name,
|
||||
name=vname,
|
||||
modes=frozenset(modes),
|
||||
accents=tuple(sorted(accents)),
|
||||
)
|
||||
)
|
||||
if modes:
|
||||
vname = var_dir.name.capitalize()
|
||||
variants.append(
|
||||
SchemeVariant(
|
||||
id=var_dir.name,
|
||||
name=vname,
|
||||
modes=frozenset(modes),
|
||||
accents=tuple(sorted(accents)),
|
||||
)
|
||||
)
|
||||
|
||||
schemes[sid] = SchemeMeta(
|
||||
id=sid,
|
||||
name=display_name,
|
||||
variants=tuple(variants),
|
||||
)
|
||||
schemes[sid] = SchemeMeta(
|
||||
id=sid,
|
||||
name=display_name,
|
||||
variants=tuple(variants),
|
||||
)
|
||||
|
||||
return schemes
|
||||
return schemes
|
||||
|
||||
|
||||
SCHEMES: dict[str, SchemeMeta] = _discover_schemes()
|
||||
|
||||
|
||||
def get_palette(scheme: str, variant: str, mode: str, accent: str | None = None) -> Palette:
|
||||
if scheme not in SCHEMES:
|
||||
raise KeyError(
|
||||
f"Unknown scheme '{scheme}'. Available: {', '.join(SCHEMES)}")
|
||||
def get_palette(
|
||||
scheme: str, variant: str, mode: str, accent: str | None = None
|
||||
) -> Palette:
|
||||
if scheme not in SCHEMES:
|
||||
raise KeyError(
|
||||
f"Unknown scheme '{scheme}'. Available: {', '.join(SCHEMES)}"
|
||||
)
|
||||
|
||||
meta = SCHEMES[scheme]
|
||||
var_ids = {v.id for v in meta.variants}
|
||||
if variant not in var_ids:
|
||||
raise KeyError(
|
||||
f"Unknown variant '{variant}' for scheme '{scheme}'. Available: {', '.join(sorted(var_ids))}")
|
||||
meta = SCHEMES[scheme]
|
||||
var_ids = {v.id for v in meta.variants}
|
||||
if variant not in var_ids:
|
||||
raise KeyError(
|
||||
f"Unknown variant '{variant}' for scheme '{scheme}'. Available: {', '.join(sorted(var_ids))}"
|
||||
)
|
||||
|
||||
if accent:
|
||||
filename = f"{accent}-{mode}.txt"
|
||||
else:
|
||||
filename = f"{mode}.txt"
|
||||
filename = f"{accent}-{mode}.txt" if accent else f"{mode}.txt"
|
||||
|
||||
txt_path = ASSETS / scheme / variant / filename
|
||||
if not txt_path.is_file():
|
||||
txt_path = ASSETS / scheme / variant / f"{mode}.txt"
|
||||
txt_path = ASSETS / scheme / variant / filename
|
||||
if not txt_path.is_file():
|
||||
txt_path = ASSETS / scheme / variant / f"{mode}.txt"
|
||||
|
||||
if not txt_path.is_file():
|
||||
var_info = next(v for v in meta.variants if v.id == variant)
|
||||
raise FileNotFoundError(
|
||||
f"No {mode} palette for '{scheme}:{variant}'. Available modes: {sorted(var_info.modes)}"
|
||||
)
|
||||
if not txt_path.is_file():
|
||||
var_info = next(v for v in meta.variants if v.id == variant)
|
||||
raise FileNotFoundError(
|
||||
f"No {mode} palette for '{scheme}:{variant}'. Available modes: {sorted(var_info.modes)}"
|
||||
)
|
||||
|
||||
colors = _parse_txt(txt_path)
|
||||
colors = _parse_txt(txt_path)
|
||||
|
||||
return Palette(colors=colors, mode=mode, scheme=scheme, variant=variant, accent=accent)
|
||||
return Palette(
|
||||
colors=colors, mode=mode, scheme=scheme, variant=variant, accent=accent
|
||||
)
|
||||
|
||||
|
||||
def list_schemes() -> dict[str, SchemeMeta]:
|
||||
return dict(SCHEMES)
|
||||
return dict(SCHEMES)
|
||||
|
||||
|
||||
def resolve_preset(spec: str) -> tuple[str, str]:
|
||||
parts = spec.split(":")
|
||||
if len(parts) == 2:
|
||||
return parts[0], parts[1]
|
||||
if len(parts) == 1:
|
||||
return parts[0], "default"
|
||||
raise ValueError(f"Invalid preset spec '{spec}'. Use <scheme>:<variant>")
|
||||
parts = spec.split(":")
|
||||
if len(parts) == 2:
|
||||
return parts[0], parts[1]
|
||||
if len(parts) == 1:
|
||||
return parts[0], "default"
|
||||
raise ValueError(f"Invalid preset spec '{spec}'. Use <scheme>:<variant>")
|
||||
|
||||
+149
-121
@@ -1,167 +1,195 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
from zshell.utils import schemepalettes as sp
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_schemes(tmp_path: Path) -> Path:
|
||||
schemes = tmp_path / "schemes"
|
||||
schemes.mkdir()
|
||||
schemes = tmp_path / "schemes"
|
||||
schemes.mkdir()
|
||||
|
||||
gmedium = schemes / "gruvbox" / "medium"
|
||||
gmedium.mkdir(parents=True)
|
||||
(gmedium / "dark.txt").write_text("background 101415\nonBackground e0e3e4\nprimary 81d3e0\nsurface 1c2021\n")
|
||||
(gmedium / "light.txt").write_text("background fbf1c7\nonBackground 3c3836\nprimary 6b5f10\nsurface fbf1c7\n")
|
||||
gmedium = schemes / "gruvbox" / "medium"
|
||||
gmedium.mkdir(parents=True)
|
||||
(gmedium / "dark.txt").write_text(
|
||||
"background 101415\nonBackground e0e3e4\nprimary 81d3e0\nsurface 1c2021\n"
|
||||
)
|
||||
(gmedium / "light.txt").write_text(
|
||||
"background fbf1c7\nonBackground 3c3836\nprimary 6b5f10\nsurface fbf1c7\n"
|
||||
)
|
||||
|
||||
ghard = schemes / "gruvbox" / "hard"
|
||||
ghard.mkdir(parents=True)
|
||||
(ghard / "dark.txt").write_text("background 0b0d0e\nprimary 81d3e0\n")
|
||||
ghard = schemes / "gruvbox" / "hard"
|
||||
ghard.mkdir(parents=True)
|
||||
(ghard / "dark.txt").write_text("background 0b0d0e\nprimary 81d3e0\n")
|
||||
|
||||
cmocha = schemes / "catppuccin" / "mocha"
|
||||
cmocha.mkdir(parents=True)
|
||||
(cmocha / "dark.txt").write_text("background 1e1e2e\nprimary cba6f7\nsecondary 756294\nsurface 313244\n")
|
||||
(cmocha / "mauve-dark.txt").write_text("background 1e1e2e\nprimary cba6f7\nsecondary 756294\nsurface 313244\n")
|
||||
(cmocha / "green-dark.txt").write_text("background 1e1e2e\nprimary a6e3a1\nsecondary 5b8964\nsurface 313244\n")
|
||||
cmocha = schemes / "catppuccin" / "mocha"
|
||||
cmocha.mkdir(parents=True)
|
||||
(cmocha / "dark.txt").write_text(
|
||||
"background 1e1e2e\nprimary cba6f7\nsecondary 756294\nsurface 313244\n"
|
||||
)
|
||||
(cmocha / "mauve-dark.txt").write_text(
|
||||
"background 1e1e2e\nprimary cba6f7\nsecondary 756294\nsurface 313244\n"
|
||||
)
|
||||
(cmocha / "green-dark.txt").write_text(
|
||||
"background 1e1e2e\nprimary a6e3a1\nsecondary 5b8964\nsurface 313244\n"
|
||||
)
|
||||
|
||||
clatte = schemes / "catppuccin" / "latte"
|
||||
clatte.mkdir(parents=True)
|
||||
(clatte / "light.txt").write_text("background eff1f5\nprimary 8839ef\nsecondary c2b8d0\nsurface ccd0da\n")
|
||||
(clatte / "mauve-light.txt").write_text("background eff1f5\nprimary 8839ef\nsecondary c2b8d0\nsurface ccd0da\n")
|
||||
clatte = schemes / "catppuccin" / "latte"
|
||||
clatte.mkdir(parents=True)
|
||||
(clatte / "light.txt").write_text(
|
||||
"background eff1f5\nprimary 8839ef\nsecondary c2b8d0\nsurface ccd0da\n"
|
||||
)
|
||||
(clatte / "mauve-light.txt").write_text(
|
||||
"background eff1f5\nprimary 8839ef\nsecondary c2b8d0\nsurface ccd0da\n"
|
||||
)
|
||||
|
||||
cextra = schemes / "extra" / "default"
|
||||
cextra.mkdir(parents=True)
|
||||
(cextra / "dark.txt").write_text(
|
||||
"# this is a comment\n\nbackground 000000\nprimary ffffff\n\n # indented comment \n secondary cccccc\n"
|
||||
)
|
||||
cextra = schemes / "extra" / "default"
|
||||
cextra.mkdir(parents=True)
|
||||
(cextra / "dark.txt").write_text(
|
||||
"# this is a comment\n\nbackground 000000\nprimary ffffff\n\n # indented comment \n secondary cccccc\n"
|
||||
)
|
||||
|
||||
return schemes
|
||||
return schemes
|
||||
|
||||
|
||||
class TestParseTxt:
|
||||
def test_basic(self, tmp_schemes):
|
||||
path = tmp_schemes / "gruvbox" / "medium" / "dark.txt"
|
||||
colors = sp._parse_txt(path)
|
||||
assert colors["background"] == "#101415"
|
||||
assert colors["primary"] == "#81d3e0"
|
||||
assert colors["surface"] == "#1c2021"
|
||||
def test_basic(self, tmp_schemes):
|
||||
path = tmp_schemes / "gruvbox" / "medium" / "dark.txt"
|
||||
colors = sp._parse_txt(path)
|
||||
assert colors["background"] == "#101415"
|
||||
assert colors["primary"] == "#81d3e0"
|
||||
assert colors["surface"] == "#1c2021"
|
||||
|
||||
def test_adds_hash_prefix(self, tmp_schemes):
|
||||
path = tmp_schemes / "gruvbox" / "medium" / "dark.txt"
|
||||
colors = sp._parse_txt(path)
|
||||
for v in colors.values():
|
||||
assert v.startswith("#"), f"value {v!r} missing # prefix"
|
||||
def test_adds_hash_prefix(self, tmp_schemes):
|
||||
path = tmp_schemes / "gruvbox" / "medium" / "dark.txt"
|
||||
colors = sp._parse_txt(path)
|
||||
for v in colors.values():
|
||||
assert v.startswith("#"), f"value {v!r} missing # prefix"
|
||||
|
||||
def test_skips_comments_and_empty_lines(self, tmp_schemes):
|
||||
path = tmp_schemes / "extra" / "default" / "dark.txt"
|
||||
colors = sp._parse_txt(path)
|
||||
assert colors["background"] == "#000000"
|
||||
assert colors["primary"] == "#ffffff"
|
||||
assert colors["secondary"] == "#cccccc"
|
||||
assert len(colors) == 3
|
||||
def test_skips_comments_and_empty_lines(self, tmp_schemes):
|
||||
path = tmp_schemes / "extra" / "default" / "dark.txt"
|
||||
colors = sp._parse_txt(path)
|
||||
assert colors["background"] == "#000000"
|
||||
assert colors["primary"] == "#ffffff"
|
||||
assert colors["secondary"] == "#cccccc"
|
||||
assert len(colors) == 3
|
||||
|
||||
|
||||
class TestDiscoverSchemes:
|
||||
def test_discovers_all_schemes(self):
|
||||
schemes = sp._discover_schemes()
|
||||
assert "gruvbox" in schemes
|
||||
assert "catppuccin" in schemes
|
||||
assert "everforest" in schemes
|
||||
assert "nord" in schemes
|
||||
assert len(schemes) >= 10
|
||||
def test_discovers_all_schemes(self):
|
||||
schemes = sp._discover_schemes()
|
||||
assert "gruvbox" in schemes
|
||||
assert "catppuccin" in schemes
|
||||
assert "everforest" in schemes
|
||||
assert "nord" in schemes
|
||||
assert len(schemes) >= 10
|
||||
|
||||
def test_scheme_has_variants(self):
|
||||
schemes = sp._discover_schemes()
|
||||
gruvbox = schemes["gruvbox"]
|
||||
var_ids = {v.id for v in gruvbox.variants}
|
||||
assert "medium" in var_ids
|
||||
assert "hard" in var_ids
|
||||
assert "soft" in var_ids
|
||||
def test_scheme_has_variants(self):
|
||||
schemes = sp._discover_schemes()
|
||||
gruvbox = schemes["gruvbox"]
|
||||
var_ids = {v.id for v in gruvbox.variants}
|
||||
assert "medium" in var_ids
|
||||
assert "hard" in var_ids
|
||||
assert "soft" in var_ids
|
||||
|
||||
def test_variant_has_modes(self):
|
||||
schemes = sp._discover_schemes()
|
||||
gmedium = next(v for v in schemes["gruvbox"].variants if v.id == "medium")
|
||||
assert "dark" in gmedium.modes
|
||||
assert "light" in gmedium.modes
|
||||
def test_variant_has_modes(self):
|
||||
schemes = sp._discover_schemes()
|
||||
gmedium = next(
|
||||
v for v in schemes["gruvbox"].variants if v.id == "medium"
|
||||
)
|
||||
assert "dark" in gmedium.modes
|
||||
assert "light" in gmedium.modes
|
||||
|
||||
def test_catppuccin_has_accents(self):
|
||||
schemes = sp._discover_schemes()
|
||||
mocha = next(v for v in schemes["catppuccin"].variants if v.id == "mocha")
|
||||
assert "mauve" in mocha.accents
|
||||
assert "green" in mocha.accents
|
||||
assert "rosewater" in mocha.accents
|
||||
assert len(mocha.accents) >= 14
|
||||
def test_catppuccin_has_accents(self):
|
||||
schemes = sp._discover_schemes()
|
||||
mocha = next(
|
||||
v for v in schemes["catppuccin"].variants if v.id == "mocha"
|
||||
)
|
||||
assert "mauve" in mocha.accents
|
||||
assert "green" in mocha.accents
|
||||
assert "rosewater" in mocha.accents
|
||||
assert len(mocha.accents) >= 14
|
||||
|
||||
def test_non_accent_scheme_has_no_accents(self):
|
||||
schemes = sp._discover_schemes()
|
||||
gmedium = next(v for v in schemes["gruvbox"].variants if v.id == "medium")
|
||||
assert gmedium.accents == ()
|
||||
def test_non_accent_scheme_has_no_accents(self):
|
||||
schemes = sp._discover_schemes()
|
||||
gmedium = next(
|
||||
v for v in schemes["gruvbox"].variants if v.id == "medium"
|
||||
)
|
||||
assert gmedium.accents == ()
|
||||
|
||||
|
||||
class TestGetPalette:
|
||||
def test_loads_basic_palette(self):
|
||||
pal = sp.get_palette("gruvbox", "medium", "dark")
|
||||
assert pal.scheme == "gruvbox"
|
||||
assert pal.variant == "medium"
|
||||
assert pal.mode == "dark"
|
||||
assert pal.colors["background"].startswith("#")
|
||||
assert pal.colors["primary"].startswith("#")
|
||||
def test_loads_basic_palette(self):
|
||||
pal = sp.get_palette("gruvbox", "medium", "dark")
|
||||
assert pal.scheme == "gruvbox"
|
||||
assert pal.variant == "medium"
|
||||
assert pal.mode == "dark"
|
||||
assert pal.colors["background"].startswith("#")
|
||||
assert pal.colors["primary"].startswith("#")
|
||||
|
||||
def test_loads_accent_palette(self):
|
||||
pal = sp.get_palette("catppuccin", "mocha", "dark", accent="mauve")
|
||||
assert pal.accent == "mauve"
|
||||
assert pal.colors["primary"] == "#cba6f7"
|
||||
def test_loads_accent_palette(self):
|
||||
pal = sp.get_palette("catppuccin", "mocha", "dark", accent="mauve")
|
||||
assert pal.accent == "mauve"
|
||||
assert pal.colors["primary"] == "#cba6f7"
|
||||
|
||||
def test_different_accent_changes_colors(self):
|
||||
mauve = sp.get_palette("catppuccin", "mocha", "dark", accent="mauve")
|
||||
green = sp.get_palette("catppuccin", "mocha", "dark", accent="green")
|
||||
assert mauve.colors["primary"] != green.colors["primary"]
|
||||
assert mauve.colors["secondary"] != green.colors["secondary"]
|
||||
def test_different_accent_changes_colors(self):
|
||||
mauve = sp.get_palette("catppuccin", "mocha", "dark", accent="mauve")
|
||||
green = sp.get_palette("catppuccin", "mocha", "dark", accent="green")
|
||||
assert mauve.colors["primary"] != green.colors["primary"]
|
||||
assert mauve.colors["secondary"] != green.colors["secondary"]
|
||||
|
||||
def test_unknown_scheme_raises(self):
|
||||
with pytest.raises(KeyError, match="Unknown scheme 'nope'"):
|
||||
sp.get_palette("nope", "medium", "dark")
|
||||
def test_unknown_scheme_raises(self):
|
||||
with pytest.raises(KeyError, match="Unknown scheme 'nope'"):
|
||||
sp.get_palette("nope", "medium", "dark")
|
||||
|
||||
def test_unknown_variant_raises(self):
|
||||
with pytest.raises(KeyError, match="Unknown variant 'bogus' for scheme 'gruvbox'"):
|
||||
sp.get_palette("gruvbox", "bogus", "dark")
|
||||
def test_unknown_variant_raises(self):
|
||||
with pytest.raises(
|
||||
KeyError, match="Unknown variant 'bogus' for scheme 'gruvbox'"
|
||||
):
|
||||
sp.get_palette("gruvbox", "bogus", "dark")
|
||||
|
||||
def test_unknown_accent_falls_back(self):
|
||||
pal = sp.get_palette("catppuccin", "mocha", "dark", accent="nonexistent")
|
||||
assert pal.accent == "nonexistent"
|
||||
assert pal.colors["primary"] is not None
|
||||
def test_unknown_accent_falls_back(self):
|
||||
pal = sp.get_palette(
|
||||
"catppuccin", "mocha", "dark", accent="nonexistent"
|
||||
)
|
||||
assert pal.accent == "nonexistent"
|
||||
assert pal.colors["primary"] is not None
|
||||
|
||||
def test_accent_on_non_accent_scheme(self):
|
||||
pal = sp.get_palette("gruvbox", "medium", "dark", accent="mauve")
|
||||
assert pal.colors is not None
|
||||
def test_accent_on_non_accent_scheme(self):
|
||||
pal = sp.get_palette("gruvbox", "medium", "dark", accent="mauve")
|
||||
assert pal.colors is not None
|
||||
|
||||
def test_non_existent_mode_raises(self):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
sp.get_palette("catppuccin", "mocha", "light")
|
||||
def test_non_existent_mode_raises(self):
|
||||
with pytest.raises(FileNotFoundError):
|
||||
sp.get_palette("catppuccin", "mocha", "light")
|
||||
|
||||
|
||||
class TestListSchemes:
|
||||
def test_returns_dict(self):
|
||||
schemes = sp.list_schemes()
|
||||
assert isinstance(schemes, dict)
|
||||
def test_returns_dict(self):
|
||||
schemes = sp.list_schemes()
|
||||
assert isinstance(schemes, dict)
|
||||
|
||||
def test_includes_known_schemes(self):
|
||||
schemes = sp.list_schemes()
|
||||
assert "catppuccin" in schemes
|
||||
assert "gruvbox" in schemes
|
||||
def test_includes_known_schemes(self):
|
||||
schemes = sp.list_schemes()
|
||||
assert "catppuccin" in schemes
|
||||
assert "gruvbox" in schemes
|
||||
|
||||
|
||||
class TestResolvePreset:
|
||||
def test_two_parts(self):
|
||||
assert sp.resolve_preset("gruvbox:medium") == ("gruvbox", "medium")
|
||||
def test_two_parts(self):
|
||||
assert sp.resolve_preset("gruvbox:medium") == ("gruvbox", "medium")
|
||||
|
||||
def test_three_parts(self):
|
||||
with pytest.raises(ValueError, match="Invalid preset spec"):
|
||||
sp.resolve_preset("catppuccin:mocha:mauve")
|
||||
def test_three_parts(self):
|
||||
with pytest.raises(ValueError, match="Invalid preset spec"):
|
||||
sp.resolve_preset("catppuccin:mocha:mauve")
|
||||
|
||||
def test_one_part(self):
|
||||
assert sp.resolve_preset("default") == ("default", "default")
|
||||
def test_one_part(self):
|
||||
assert sp.resolve_preset("default") == ("default", "default")
|
||||
|
||||
def test_edge_spaces(self):
|
||||
assert sp.resolve_preset(" catppuccin : mocha ") == (" catppuccin ", " mocha ")
|
||||
def test_edge_spaces(self):
|
||||
assert sp.resolve_preset(" catppuccin : mocha ") == (
|
||||
" catppuccin ",
|
||||
" mocha ",
|
||||
)
|
||||
|
||||
+134
-93
@@ -1,7 +1,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from subprocess import CompletedProcess
|
||||
from unittest.mock import patch, call
|
||||
from unittest.mock import call, patch
|
||||
|
||||
from typer.testing import CliRunner
|
||||
from zshell.subcommands.shell import app
|
||||
@@ -10,122 +10,163 @@ runner = CliRunner()
|
||||
|
||||
|
||||
def invoke(*args: str):
|
||||
result = runner.invoke(app, args)
|
||||
if result.exit_code != 0:
|
||||
raise RuntimeError(result.output)
|
||||
return result
|
||||
result = runner.invoke(app, args)
|
||||
if result.exit_code != 0:
|
||||
raise RuntimeError(result.output)
|
||||
return result
|
||||
|
||||
|
||||
class TestKill:
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_kill_runs_qs_kill_success(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"Killed abc\n")
|
||||
invoke("kill")
|
||||
mock_run.assert_called_once_with(["qs", "-c", "zshell", "kill"], capture_output=True)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_kill_runs_qs_kill_success(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"Killed abc\n")
|
||||
invoke("kill")
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "kill"], capture_output=True
|
||||
)
|
||||
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_kill_no_instance_errors(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 255, b"", b"No running instances\n")
|
||||
result = runner.invoke(app, ["kill"])
|
||||
assert result.exit_code != 0
|
||||
assert "No running instance to kill" in result.output
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_kill_no_instance_errors(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess(
|
||||
[], 255, b"", b"No running instances\n"
|
||||
)
|
||||
result = runner.invoke(app, ["kill"])
|
||||
assert result.exit_code != 0
|
||||
assert "No running instance to kill" in result.output
|
||||
|
||||
|
||||
class TestStart:
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_default_daemon(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"Launching config\n")
|
||||
invoke("start")
|
||||
mock_run.assert_called_once_with(["qs", "-c", "zshell", "-n", "-d"], capture_output=True)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_default_daemon(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess(
|
||||
[], 0, b"", b"Launching config\n"
|
||||
)
|
||||
invoke("start")
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "-n", "-d"], capture_output=True
|
||||
)
|
||||
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_no_daemon(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"Launching config\n")
|
||||
invoke("start", "--no-daemon")
|
||||
mock_run.assert_called_once_with(["qs", "-c", "zshell", "-n"], capture_output=True)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_no_daemon(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess(
|
||||
[], 0, b"", b"Launching config\n"
|
||||
)
|
||||
invoke("start", "--no-daemon")
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "-n"], capture_output=True
|
||||
)
|
||||
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_already_running_errors(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"An instance of this configuration is already running.\n", b"")
|
||||
result = runner.invoke(app, ["start"])
|
||||
assert result.exit_code != 0
|
||||
assert "already running" in result.output
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_already_running_errors(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess(
|
||||
[],
|
||||
0,
|
||||
b"An instance of this configuration is already running.\n",
|
||||
b"",
|
||||
)
|
||||
result = runner.invoke(app, ["start"])
|
||||
assert result.exit_code != 0
|
||||
assert "already running" in result.output
|
||||
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_other_failure_errors(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 1, b"", b"Config error\n")
|
||||
result = runner.invoke(app, ["start"])
|
||||
assert result.exit_code != 0
|
||||
assert "Config error" in result.output
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_start_other_failure_errors(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 1, b"", b"Config error\n")
|
||||
result = runner.invoke(app, ["start"])
|
||||
assert result.exit_code != 0
|
||||
assert "Config error" in result.output
|
||||
|
||||
|
||||
class TestShow:
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_show_runs_ipc_show(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"target visibilities\n", b"")
|
||||
result = invoke("show")
|
||||
assert "target visibilities" in result.output
|
||||
mock_run.assert_called_once_with(["qs", "-c", "zshell", "ipc", "show"], capture_output=True)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_show_runs_ipc_show(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess(
|
||||
[], 0, b"target visibilities\n", b""
|
||||
)
|
||||
result = invoke("show")
|
||||
assert "target visibilities" in result.output
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "ipc", "show"], capture_output=True
|
||||
)
|
||||
|
||||
|
||||
class TestLog:
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_log_runs_qs_log(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"log output\n", b"")
|
||||
invoke("log")
|
||||
mock_run.assert_called_once_with(["qs", "-c", "zshell", "log"], capture_output=True)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_log_runs_qs_log(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"log output\n", b"")
|
||||
invoke("log")
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "log"], capture_output=True
|
||||
)
|
||||
|
||||
|
||||
class TestLock:
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_lock_runs_ipc_call_lock(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"")
|
||||
invoke("lock")
|
||||
mock_run.assert_called_once_with(["qs", "-c", "zshell", "ipc", "call", "lock", "lock"], capture_output=True)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_lock_runs_ipc_call_lock(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"")
|
||||
invoke("lock")
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "ipc", "call", "lock", "lock"],
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
|
||||
class TestCall:
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_call_no_args(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"")
|
||||
invoke("call", "target", "method")
|
||||
mock_run.assert_called_once_with(["qs", "-c", "zshell", "ipc", "call", "target", "method"], capture_output=True)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_call_no_args(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"")
|
||||
invoke("call", "target", "method")
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "ipc", "call", "target", "method"],
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_call_with_args(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"")
|
||||
invoke("call", "target", "method", "arg1", "arg2")
|
||||
mock_run.assert_called_once_with(
|
||||
["qs", "-c", "zshell", "ipc", "call", "target", "method", "arg1", "arg2"],
|
||||
capture_output=True,
|
||||
)
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_call_with_args(self, mock_run):
|
||||
mock_run.return_value = CompletedProcess([], 0, b"", b"")
|
||||
invoke("call", "target", "method", "arg1", "arg2")
|
||||
mock_run.assert_called_once_with(
|
||||
[
|
||||
"qs",
|
||||
"-c",
|
||||
"zshell",
|
||||
"ipc",
|
||||
"call",
|
||||
"target",
|
||||
"method",
|
||||
"arg1",
|
||||
"arg2",
|
||||
],
|
||||
capture_output=True,
|
||||
)
|
||||
|
||||
|
||||
class TestRestart:
|
||||
@patch("zshell.subcommands.shell.start_instance")
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_restart_kills_then_starts(self, mock_run, mock_start):
|
||||
mock_run.side_effect = [
|
||||
CompletedProcess([], 0, b"", b"Killed abc\n"), # first kill (captured)
|
||||
CompletedProcess([], 255, b"", b""), # poll → no instance
|
||||
]
|
||||
invoke("restart")
|
||||
assert mock_run.call_args_list == [
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
]
|
||||
mock_start.assert_called_once_with(no_daemon=False)
|
||||
@patch("zshell.subcommands.shell.start_instance")
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_restart_kills_then_starts(self, mock_run, mock_start):
|
||||
mock_run.side_effect = [
|
||||
CompletedProcess(
|
||||
[], 0, b"", b"Killed abc\n"
|
||||
), # first kill (captured)
|
||||
CompletedProcess([], 255, b"", b""), # poll → no instance
|
||||
]
|
||||
invoke("restart")
|
||||
assert mock_run.call_args_list == [
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
]
|
||||
mock_start.assert_called_once_with(no_daemon=False)
|
||||
|
||||
@patch("zshell.subcommands.shell.start_instance")
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_restart_no_daemon(self, mock_run, mock_start):
|
||||
mock_run.side_effect = [
|
||||
CompletedProcess([], 0, b"", b"Killed abc\n"),
|
||||
CompletedProcess([], 255, b"", b""),
|
||||
]
|
||||
invoke("restart", "--no-daemon")
|
||||
assert mock_run.call_args_list == [
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
]
|
||||
mock_start.assert_called_once_with(no_daemon=True)
|
||||
@patch("zshell.subcommands.shell.start_instance")
|
||||
@patch("zshell.subcommands.shell.subprocess.run")
|
||||
def test_restart_no_daemon(self, mock_run, mock_start):
|
||||
mock_run.side_effect = [
|
||||
CompletedProcess([], 0, b"", b"Killed abc\n"),
|
||||
CompletedProcess([], 255, b"", b""),
|
||||
]
|
||||
invoke("restart", "--no-daemon")
|
||||
assert mock_run.call_args_list == [
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
call(["qs", "-c", "zshell", "kill"], capture_output=True),
|
||||
]
|
||||
mock_start.assert_called_once_with(no_daemon=True)
|
||||
|
||||
@@ -21,17 +21,39 @@ source = "vcs"
|
||||
|
||||
[tool.hatch.build]
|
||||
include = [
|
||||
"src/zshell/assets/**",
|
||||
"cli/src/zshell/assets/**",
|
||||
]
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["cli/src/zshell"]
|
||||
|
||||
[tool.hatch.build.targets.sdist]
|
||||
only-include = [
|
||||
"src",
|
||||
"cli/src",
|
||||
]
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 120
|
||||
line-length = 80
|
||||
|
||||
[tool.ruff.format]
|
||||
quote-style = "double"
|
||||
indent-style = "tab"
|
||||
line-ending = "lf"
|
||||
docstring-code-format = true
|
||||
docstring-code-line-length = "dynamic"
|
||||
|
||||
[tool.ruff.lint]
|
||||
ignore = ["E501", "B008"]
|
||||
select = [
|
||||
"E",
|
||||
"F",
|
||||
"I",
|
||||
"UP",
|
||||
"B",
|
||||
"SIM",
|
||||
"RUF",
|
||||
]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
pythonpath = ["src"]
|
||||
testpaths = ["cli/tests"]
|
||||
pythonpath = ["cli/src"]
|
||||
+352
-307
@@ -4,22 +4,24 @@ import json
|
||||
import re
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from functools import lru_cache
|
||||
from functools import cache
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@lru_cache(maxsize=None)
|
||||
@cache
|
||||
def read_lines(path: Path) -> tuple[str, ...]:
|
||||
return tuple(path.read_text().splitlines())
|
||||
return tuple(path.read_text().splitlines())
|
||||
|
||||
|
||||
ROW_RE = re.compile(
|
||||
r'^\s*(ToggleRow|SliderRow|SelectRow|SpinRow|NavRow|InfoRow|PopupRow|OverlayRow|DefaultRow)\s*\{')
|
||||
r"^\s*(ToggleRow|SliderRow|SelectRow|SpinRow|NavRow|InfoRow|PopupRow|OverlayRow|DefaultRow)\s*\{"
|
||||
)
|
||||
LABEL_RE = re.compile(r'^\s*(?:label|text):\s*qsTr\("([^"]+)"\)')
|
||||
ANCHOR_RE = re.compile(r'^\s*settingAnchor:\s*"([^"]+)"')
|
||||
CHECKED_RE = re.compile(r'^\s*checked:\s*(?:Config)\.([\w.]+)\s*$')
|
||||
CHECKED_RE = re.compile(r"^\s*checked:\s*(?:Config)\.([\w.]+)\s*$")
|
||||
ONTOGGLED_RE = re.compile(
|
||||
r'^\s*onToggled:\s*(?:Config)\.([\w.]+)\s*=\s*checked\s*$')
|
||||
r"^\s*onToggled:\s*(?:Config)\.([\w.]+)\s*=\s*checked\s*$"
|
||||
)
|
||||
ICON_RE = re.compile(r'^\s*icon:\s*"([^"]+)"')
|
||||
SKIP_LABELS = {"Muted", "None"}
|
||||
FIELD_WEIGHT = {"title": 1.0, "keywords": 0.4}
|
||||
@@ -27,14 +29,14 @@ STOPWORDS = {"the", "a", "an", "of", "and", "or", "to", "on", "in", "for"}
|
||||
|
||||
|
||||
def find_pages_dir(settings: Path) -> Path:
|
||||
return settings / "Pages"
|
||||
return settings / "Pages"
|
||||
|
||||
|
||||
def discover_files(settings: Path) -> dict[str, Path]:
|
||||
files: dict[str, Path] = {}
|
||||
for p in find_pages_dir(settings).rglob("*.qml"):
|
||||
files[p.stem] = p
|
||||
return files
|
||||
files: dict[str, Path] = {}
|
||||
for p in find_pages_dir(settings).rglob("*.qml"):
|
||||
files[p.stem] = p
|
||||
return files
|
||||
|
||||
|
||||
PAGE_NAME_RE = re.compile(r'^\s*name:\s*qsTr\("([^"]+)"\)')
|
||||
@@ -42,366 +44,409 @@ PAGE_ICON_RE = re.compile(r'^\s*icon:\s*"([^"]+)"')
|
||||
|
||||
|
||||
def parse_page_registry(settings: Path) -> list[tuple[str, str]]:
|
||||
text = (settings / "PageRegistry.qml").read_text().splitlines()
|
||||
text = (settings / "PageRegistry.qml").read_text().splitlines()
|
||||
|
||||
start = next(
|
||||
i for i, line in enumerate(text)
|
||||
if re.search(r'\bpages\s*:\s*\[', line)
|
||||
)
|
||||
start = next(
|
||||
i for i, line in enumerate(text) if re.search(r"\bpages\s*:\s*\[", line)
|
||||
)
|
||||
|
||||
out: list[tuple[str, str]] = []
|
||||
i = start + 1
|
||||
out: list[tuple[str, str]] = []
|
||||
i = start + 1
|
||||
|
||||
while i < len(text):
|
||||
line = text[i].strip()
|
||||
while i < len(text):
|
||||
line = text[i].strip()
|
||||
|
||||
if line.startswith("]"):
|
||||
break
|
||||
if line.startswith("]"):
|
||||
break
|
||||
|
||||
if line.startswith("//") or not line:
|
||||
i += 1
|
||||
continue
|
||||
if line.startswith("//") or not line:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
if line.startswith("{"):
|
||||
name = None
|
||||
icon = None
|
||||
i += 1
|
||||
if line.startswith("{"):
|
||||
name = None
|
||||
icon = None
|
||||
i += 1
|
||||
|
||||
while i < len(text):
|
||||
s = text[i].strip()
|
||||
while i < len(text):
|
||||
s = text[i].strip()
|
||||
|
||||
if s.startswith("}"):
|
||||
if name is not None:
|
||||
out.append((icon or "tune", name))
|
||||
break
|
||||
if s.startswith("}"):
|
||||
if name is not None:
|
||||
out.append((icon or "tune", name))
|
||||
break
|
||||
|
||||
if name is None:
|
||||
m = PAGE_NAME_RE.match(text[i])
|
||||
if m:
|
||||
name = m.group(1)
|
||||
if name is None:
|
||||
m = PAGE_NAME_RE.match(text[i])
|
||||
if m:
|
||||
name = m.group(1)
|
||||
|
||||
if icon is None:
|
||||
mi = PAGE_ICON_RE.match(text[i])
|
||||
if mi:
|
||||
icon = mi.group(1)
|
||||
if icon is None:
|
||||
mi = PAGE_ICON_RE.match(text[i])
|
||||
if mi:
|
||||
icon = mi.group(1)
|
||||
|
||||
i += 1
|
||||
i += 1
|
||||
|
||||
i += 1
|
||||
i += 1
|
||||
|
||||
return out
|
||||
return out
|
||||
|
||||
|
||||
BLOCK_RE = re.compile(r'^\s*([A-Za-z_][A-Za-z0-9_]*)\s*\{\s*$')
|
||||
BLOCK_RE = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*\{\s*$")
|
||||
|
||||
|
||||
def _strip_comment(line: str) -> str:
|
||||
return line.split("//", 1)[0].rstrip()
|
||||
return line.split("//", 1)[0].rstrip()
|
||||
|
||||
|
||||
def parse_block(lines: list[str], i: int) -> tuple[str, list[tuple[str, list]], int]:
|
||||
line = _strip_comment(lines[i]).strip()
|
||||
m = BLOCK_RE.match(line)
|
||||
if not m:
|
||||
raise ValueError(f"Expected block start at line {i + 1}: {lines[i]!r}")
|
||||
def parse_block(
|
||||
lines: list[str], i: int
|
||||
) -> tuple[str, list[tuple[str, list]], int]:
|
||||
line = _strip_comment(lines[i]).strip()
|
||||
m = BLOCK_RE.match(line)
|
||||
if not m:
|
||||
raise ValueError(f"Expected block start at line {i + 1}: {lines[i]!r}")
|
||||
|
||||
name = m.group(1)
|
||||
i += 1
|
||||
children: list[tuple[str, list]] = []
|
||||
name = m.group(1)
|
||||
i += 1
|
||||
children: list[tuple[str, list]] = []
|
||||
|
||||
while i < len(lines):
|
||||
s = _strip_comment(lines[i]).strip()
|
||||
if not s:
|
||||
i += 1
|
||||
continue
|
||||
while i < len(lines):
|
||||
s = _strip_comment(lines[i]).strip()
|
||||
if not s:
|
||||
i += 1
|
||||
continue
|
||||
|
||||
if s.startswith("}"):
|
||||
return name, children, i + 1
|
||||
if s.startswith("}"):
|
||||
return name, children, i + 1
|
||||
|
||||
if BLOCK_RE.match(s):
|
||||
child_name, child_children, i = parse_block(lines, i)
|
||||
children.append((child_name, child_children))
|
||||
continue
|
||||
if BLOCK_RE.match(s):
|
||||
child_name, child_children, i = parse_block(lines, i)
|
||||
children.append((child_name, child_children))
|
||||
continue
|
||||
|
||||
i += 1
|
||||
i += 1
|
||||
|
||||
raise ValueError(f"Unterminated block: {name}")
|
||||
raise ValueError(f"Unterminated block: {name}")
|
||||
|
||||
|
||||
def collect_page_names(block: tuple[str, list[tuple[str, list]]]) -> list[str]:
|
||||
name, children = block
|
||||
name, children = block
|
||||
|
||||
if name != "Component":
|
||||
return [name]
|
||||
if name != "Component":
|
||||
return [name]
|
||||
|
||||
for child_name, child_children in children:
|
||||
if child_name == "StackPage":
|
||||
out: list[str] = []
|
||||
for grand_name, grand_children in child_children:
|
||||
if grand_name == "Component":
|
||||
out.extend(collect_page_names((grand_name, grand_children)))
|
||||
return out
|
||||
for child_name, child_children in children:
|
||||
if child_name == "StackPage":
|
||||
out: list[str] = []
|
||||
for grand_name, grand_children in child_children:
|
||||
if grand_name == "Component":
|
||||
out.extend(collect_page_names((grand_name, grand_children)))
|
||||
return out
|
||||
|
||||
if child_name != "Component":
|
||||
return [child_name]
|
||||
if child_name != "Component":
|
||||
return [child_name]
|
||||
|
||||
return []
|
||||
return []
|
||||
|
||||
|
||||
def parse_page_comps(settings: Path) -> list[list[str]]:
|
||||
text = (settings / "PageCompRegistry.qml").read_text().splitlines()
|
||||
text = (settings / "PageCompRegistry.qml").read_text().splitlines()
|
||||
|
||||
start = next(
|
||||
i for i, line in enumerate(text)
|
||||
if re.search(r'\bpageComps\s*:\s*\[', _strip_comment(line))
|
||||
)
|
||||
start = next(
|
||||
i
|
||||
for i, line in enumerate(text)
|
||||
if re.search(r"\bpageComps\s*:\s*\[", _strip_comment(line))
|
||||
)
|
||||
|
||||
comps: list[list[str]] = []
|
||||
i = start + 1
|
||||
comps: list[list[str]] = []
|
||||
i = start + 1
|
||||
|
||||
while i < len(text):
|
||||
s = _strip_comment(text[i]).strip()
|
||||
if not s:
|
||||
i += 1
|
||||
continue
|
||||
if s.startswith("]"):
|
||||
break
|
||||
while i < len(text):
|
||||
s = _strip_comment(text[i]).strip()
|
||||
if not s:
|
||||
i += 1
|
||||
continue
|
||||
if s.startswith("]"):
|
||||
break
|
||||
|
||||
if BLOCK_RE.match(s) and BLOCK_RE.match(s).group(1) == "Component":
|
||||
block = parse_block(text, i)
|
||||
names = collect_page_names((block[0], block[1]))
|
||||
if names:
|
||||
comps.append(names)
|
||||
i = block[2]
|
||||
continue
|
||||
if BLOCK_RE.match(s) and BLOCK_RE.match(s).group(1) == "Component":
|
||||
block = parse_block(text, i)
|
||||
names = collect_page_names((block[0], block[1]))
|
||||
if names:
|
||||
comps.append(names)
|
||||
i = block[2]
|
||||
continue
|
||||
|
||||
i += 1
|
||||
i += 1
|
||||
|
||||
return comps
|
||||
return comps
|
||||
|
||||
|
||||
def dedup_crumbs(labels: list[str], icons: list[str]) -> tuple[list[str], list[str]]:
|
||||
out_labels: list[str] = []
|
||||
out_icons: list[str] = []
|
||||
for lbl, ico in zip(labels, icons):
|
||||
if out_labels and out_labels[-1] == lbl:
|
||||
continue
|
||||
out_labels.append(lbl)
|
||||
out_icons.append(ico)
|
||||
return out_labels, out_icons
|
||||
def dedup_crumbs(
|
||||
labels: list[str], icons: list[str]
|
||||
) -> tuple[list[str], list[str]]:
|
||||
out_labels: list[str] = []
|
||||
out_icons: list[str] = []
|
||||
for lbl, ico in zip(labels, icons, strict=False):
|
||||
if out_labels and out_labels[-1] == lbl:
|
||||
continue
|
||||
out_labels.append(lbl)
|
||||
out_icons.append(ico)
|
||||
return out_labels, out_icons
|
||||
|
||||
|
||||
def build_nav_map(settings: Path, files: dict[str, Path]) -> dict[str, dict]:
|
||||
comps = parse_page_comps(settings)
|
||||
registry = parse_page_registry(settings)
|
||||
comps = parse_page_comps(settings)
|
||||
registry = parse_page_registry(settings)
|
||||
|
||||
top_meta: dict[int, tuple[str, str]] = {}
|
||||
for i, (icon, label) in enumerate(registry):
|
||||
top_meta[i] = (icon, label)
|
||||
top_meta: dict[int, tuple[str, str]] = {}
|
||||
for i, (icon, label) in enumerate(registry):
|
||||
top_meta[i] = (icon, label)
|
||||
|
||||
nav_children: dict[str, dict[int, tuple[str, str, str]]] = {}
|
||||
for names in comps:
|
||||
for name in names:
|
||||
pf = files.get(name)
|
||||
if not pf:
|
||||
continue
|
||||
pending_icon = pending_label = None
|
||||
section = ""
|
||||
expect_section = False
|
||||
for ln in read_lines(pf):
|
||||
if SECTION_RE.match(ln):
|
||||
expect_section = True
|
||||
continue
|
||||
ml = LABEL_RE.match(ln)
|
||||
if ml:
|
||||
if expect_section:
|
||||
section = ml.group(1)
|
||||
expect_section = False
|
||||
else:
|
||||
pending_label = ml.group(1)
|
||||
continue
|
||||
mi = ICON_RE.match(ln)
|
||||
if mi:
|
||||
pending_icon = mi.group(1)
|
||||
mo = re.search(r"openSubPage\((\d+)\)", ln)
|
||||
if mo:
|
||||
pos = int(mo.group(1))
|
||||
nav_children.setdefault(name, {})[pos] = (
|
||||
pending_icon or "tune", pending_label or "", section)
|
||||
pending_icon = pending_label = None
|
||||
nav_children: dict[str, dict[int, tuple[str, str, str]]] = {}
|
||||
for names in comps:
|
||||
for name in names:
|
||||
pf = files.get(name)
|
||||
if not pf:
|
||||
continue
|
||||
pending_icon = pending_label = None
|
||||
section = ""
|
||||
expect_section = False
|
||||
for ln in read_lines(pf):
|
||||
if SECTION_RE.match(ln):
|
||||
expect_section = True
|
||||
continue
|
||||
ml = LABEL_RE.match(ln)
|
||||
if ml:
|
||||
if expect_section:
|
||||
section = ml.group(1)
|
||||
expect_section = False
|
||||
else:
|
||||
pending_label = ml.group(1)
|
||||
continue
|
||||
mi = ICON_RE.match(ln)
|
||||
if mi:
|
||||
pending_icon = mi.group(1)
|
||||
mo = re.search(r"openSubPage\((\d+)\)", ln)
|
||||
if mo:
|
||||
pos = int(mo.group(1))
|
||||
nav_children.setdefault(name, {})[pos] = (
|
||||
pending_icon or "tune",
|
||||
pending_label or "",
|
||||
section,
|
||||
)
|
||||
pending_icon = pending_label = None
|
||||
|
||||
nav: dict[str, dict] = {}
|
||||
for top_idx, names in enumerate(comps):
|
||||
if not names:
|
||||
continue
|
||||
main = names[0]
|
||||
main_icon, main_label = top_meta.get(top_idx, ("tune", main))
|
||||
nav[main] = {"pageIdx": top_idx, "subPath": [],
|
||||
"crumbIcons": [main_icon], "crumbLabels": [main_label]}
|
||||
children = dict(nav_children.get(main, {}))
|
||||
opened_via_subpage = set()
|
||||
for owner, kids in nav_children.items():
|
||||
owner_group = next((ns for ns in comps if owner in ns), None)
|
||||
if not owner_group:
|
||||
continue
|
||||
for kpos in kids:
|
||||
if kpos < len(owner_group):
|
||||
opened_via_subpage.add(owner_group[kpos])
|
||||
for pos in range(1, len(names)):
|
||||
if pos not in children and names[pos] not in opened_via_subpage:
|
||||
label = re.sub(r"(Detail)?Page$", "", names[pos])
|
||||
label = re.sub(r"(?<!^)(?=[A-Z])", " ", label)
|
||||
children[pos] = (main_icon, label, "")
|
||||
for pos, (icon, label, section) in children.items():
|
||||
if pos >= len(names):
|
||||
continue
|
||||
child = names[pos]
|
||||
labels = [main_label] + ([section] if section else []) + [label]
|
||||
icons = [main_icon] + ([icon] if section else []) + [icon]
|
||||
labels, icons = dedup_crumbs(labels, icons)
|
||||
nav[child] = {"pageIdx": top_idx, "subPath": [pos],
|
||||
"crumbIcons": icons,
|
||||
"crumbLabels": labels}
|
||||
for gpos, (gicon, glabel, gsection) in nav_children.get(child, {}).items():
|
||||
if gpos >= len(names):
|
||||
continue
|
||||
glabels = labels + ([gsection] if gsection else []) + [glabel]
|
||||
gicons = icons + ([gicon] if gsection else []) + [gicon]
|
||||
glabels, gicons = dedup_crumbs(glabels, gicons)
|
||||
nav[names[gpos]] = {
|
||||
"pageIdx": top_idx, "subPath": [pos, gpos],
|
||||
"crumbIcons": gicons,
|
||||
"crumbLabels": glabels}
|
||||
return nav
|
||||
nav: dict[str, dict] = {}
|
||||
for top_idx, names in enumerate(comps):
|
||||
if not names:
|
||||
continue
|
||||
main = names[0]
|
||||
main_icon, main_label = top_meta.get(top_idx, ("tune", main))
|
||||
nav[main] = {
|
||||
"pageIdx": top_idx,
|
||||
"subPath": [],
|
||||
"crumbIcons": [main_icon],
|
||||
"crumbLabels": [main_label],
|
||||
}
|
||||
children = dict(nav_children.get(main, {}))
|
||||
opened_via_subpage = set()
|
||||
for owner, kids in nav_children.items():
|
||||
owner_group = next((ns for ns in comps if owner in ns), None)
|
||||
if not owner_group:
|
||||
continue
|
||||
for kpos in kids:
|
||||
if kpos < len(owner_group):
|
||||
opened_via_subpage.add(owner_group[kpos])
|
||||
for pos in range(1, len(names)):
|
||||
if pos not in children and names[pos] not in opened_via_subpage:
|
||||
label = re.sub(r"(Detail)?Page$", "", names[pos])
|
||||
label = re.sub(r"(?<!^)(?=[A-Z])", " ", label)
|
||||
children[pos] = (main_icon, label, "")
|
||||
for pos, (icon, label, section) in children.items():
|
||||
if pos >= len(names):
|
||||
continue
|
||||
child = names[pos]
|
||||
labels = [main_label] + ([section] if section else []) + [label]
|
||||
icons = [main_icon] + ([icon] if section else []) + [icon]
|
||||
labels, icons = dedup_crumbs(labels, icons)
|
||||
nav[child] = {
|
||||
"pageIdx": top_idx,
|
||||
"subPath": [pos],
|
||||
"crumbIcons": icons,
|
||||
"crumbLabels": labels,
|
||||
}
|
||||
for gpos, (gicon, glabel, gsection) in nav_children.get(
|
||||
child, {}
|
||||
).items():
|
||||
if gpos >= len(names):
|
||||
continue
|
||||
glabels = labels + ([gsection] if gsection else []) + [glabel]
|
||||
gicons = icons + ([gicon] if gsection else []) + [gicon]
|
||||
glabels, gicons = dedup_crumbs(glabels, gicons)
|
||||
nav[names[gpos]] = {
|
||||
"pageIdx": top_idx,
|
||||
"subPath": [pos, gpos],
|
||||
"crumbIcons": gicons,
|
||||
"crumbLabels": glabels,
|
||||
}
|
||||
return nav
|
||||
|
||||
|
||||
def tokenize(text: str) -> list[str]:
|
||||
toks: list[str] = []
|
||||
for word in text.lower().split():
|
||||
parts = [p for p in re.split(r"[^a-z0-9]+", word) if p]
|
||||
for p in parts:
|
||||
if p not in STOPWORDS and p not in toks:
|
||||
toks.append(p)
|
||||
if len(parts) > 1:
|
||||
joined = "".join(parts)
|
||||
if joined not in toks:
|
||||
toks.append(joined)
|
||||
return toks
|
||||
toks: list[str] = []
|
||||
for word in text.lower().split():
|
||||
parts = [p for p in re.split(r"[^a-z0-9]+", word) if p]
|
||||
for p in parts:
|
||||
if p not in STOPWORDS and p not in toks:
|
||||
toks.append(p)
|
||||
if len(parts) > 1:
|
||||
joined = "".join(parts)
|
||||
if joined not in toks:
|
||||
toks.append(joined)
|
||||
return toks
|
||||
|
||||
|
||||
SUBTEXT_RE = re.compile(r'^\s*(?:subtext|status):\s*qsTr\("([^"]+)"\)')
|
||||
SECTION_RE = re.compile(r'^\s*SectionHeader\s*\{')
|
||||
SECTION_RE = re.compile(r"^\s*SectionHeader\s*\{")
|
||||
|
||||
|
||||
def extract_settings(files: dict[str, Path], nav: dict[str, dict]) -> list[dict]:
|
||||
entries: list[dict] = []
|
||||
for comp, meta in nav.items():
|
||||
pf = files.get(comp)
|
||||
if not pf:
|
||||
continue
|
||||
lines = read_lines(pf)
|
||||
section = ""
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
if SECTION_RE.match(lines[i]):
|
||||
for j in range(i + 1, min(i + 4, len(lines))):
|
||||
m = LABEL_RE.match(lines[j])
|
||||
if m:
|
||||
section = m.group(1)
|
||||
break
|
||||
row_match = ROW_RE.match(lines[i])
|
||||
if row_match:
|
||||
row_type = row_match.group(1)
|
||||
label = anchor = subtext = None
|
||||
checked_path = toggled_path = None
|
||||
for j in range(i + 1, min(i + 12, len(lines))):
|
||||
if label is None:
|
||||
m = LABEL_RE.match(lines[j])
|
||||
if m:
|
||||
label = m.group(1)
|
||||
if anchor is None:
|
||||
a = ANCHOR_RE.match(lines[j])
|
||||
if a:
|
||||
anchor = a.group(1)
|
||||
if subtext is None:
|
||||
st = SUBTEXT_RE.match(lines[j])
|
||||
if st:
|
||||
subtext = st.group(1)
|
||||
if checked_path is None:
|
||||
ch = CHECKED_RE.match(lines[j])
|
||||
if ch:
|
||||
checked_path = ch.group(1)
|
||||
if toggled_path is None:
|
||||
tg = ONTOGGLED_RE.match(lines[j])
|
||||
if tg:
|
||||
toggled_path = tg.group(1)
|
||||
toggle_path = (
|
||||
checked_path
|
||||
if row_type == "ToggleRow" and checked_path and checked_path == toggled_path
|
||||
else ""
|
||||
)
|
||||
if label and label not in SKIP_LABELS and anchor:
|
||||
extra = " ".join(meta["crumbLabels"]) + \
|
||||
" " + section + " " + (subtext or "")
|
||||
entries.append({
|
||||
"pageIdx": meta["pageIdx"], "subPath": meta["subPath"],
|
||||
"crumbIcons": meta["crumbIcons"],
|
||||
"crumbLabels": meta["crumbLabels"],
|
||||
"title": label, "anchor": anchor,
|
||||
"section": section,
|
||||
"subtext": subtext or "",
|
||||
"togglePath": toggle_path,
|
||||
"keywords": " ".join(sorted(set(tokenize(label + " " + extra)))),
|
||||
})
|
||||
i += 1
|
||||
return entries
|
||||
def extract_settings(
|
||||
files: dict[str, Path], nav: dict[str, dict]
|
||||
) -> list[dict]:
|
||||
entries: list[dict] = []
|
||||
for comp, meta in nav.items():
|
||||
pf = files.get(comp)
|
||||
if not pf:
|
||||
continue
|
||||
lines = read_lines(pf)
|
||||
section = ""
|
||||
i = 0
|
||||
while i < len(lines):
|
||||
if SECTION_RE.match(lines[i]):
|
||||
for j in range(i + 1, min(i + 4, len(lines))):
|
||||
m = LABEL_RE.match(lines[j])
|
||||
if m:
|
||||
section = m.group(1)
|
||||
break
|
||||
row_match = ROW_RE.match(lines[i])
|
||||
if row_match:
|
||||
row_type = row_match.group(1)
|
||||
label = anchor = subtext = None
|
||||
checked_path = toggled_path = None
|
||||
for j in range(i + 1, min(i + 12, len(lines))):
|
||||
if label is None:
|
||||
m = LABEL_RE.match(lines[j])
|
||||
if m:
|
||||
label = m.group(1)
|
||||
if anchor is None:
|
||||
a = ANCHOR_RE.match(lines[j])
|
||||
if a:
|
||||
anchor = a.group(1)
|
||||
if subtext is None:
|
||||
st = SUBTEXT_RE.match(lines[j])
|
||||
if st:
|
||||
subtext = st.group(1)
|
||||
if checked_path is None:
|
||||
ch = CHECKED_RE.match(lines[j])
|
||||
if ch:
|
||||
checked_path = ch.group(1)
|
||||
if toggled_path is None:
|
||||
tg = ONTOGGLED_RE.match(lines[j])
|
||||
if tg:
|
||||
toggled_path = tg.group(1)
|
||||
toggle_path = (
|
||||
checked_path
|
||||
if row_type == "ToggleRow"
|
||||
and checked_path
|
||||
and checked_path == toggled_path
|
||||
else ""
|
||||
)
|
||||
if label and label not in SKIP_LABELS and anchor:
|
||||
extra = (
|
||||
" ".join(meta["crumbLabels"])
|
||||
+ " "
|
||||
+ section
|
||||
+ " "
|
||||
+ (subtext or "")
|
||||
)
|
||||
entries.append(
|
||||
{
|
||||
"pageIdx": meta["pageIdx"],
|
||||
"subPath": meta["subPath"],
|
||||
"crumbIcons": meta["crumbIcons"],
|
||||
"crumbLabels": meta["crumbLabels"],
|
||||
"title": label,
|
||||
"anchor": anchor,
|
||||
"section": section,
|
||||
"subtext": subtext or "",
|
||||
"togglePath": toggle_path,
|
||||
"keywords": " ".join(
|
||||
sorted(set(tokenize(label + " " + extra)))
|
||||
),
|
||||
}
|
||||
)
|
||||
i += 1
|
||||
return entries
|
||||
|
||||
|
||||
def build_inverted_and_ranking(entries: list[dict]):
|
||||
inverted: dict[str, list[int]] = defaultdict(list)
|
||||
ranking: dict[str, dict[int, float]] = defaultdict(dict)
|
||||
for idx, e in enumerate(entries):
|
||||
fields = {"title": e["title"], "keywords": e["keywords"]}
|
||||
seen: set[str] = set()
|
||||
for field, text in fields.items():
|
||||
weight = FIELD_WEIGHT.get(field, 0.2)
|
||||
for tok in tokenize(text):
|
||||
if idx not in inverted[tok]:
|
||||
inverted[tok].append(idx)
|
||||
ranking[tok][idx] = max(ranking[tok].get(idx, 0.0), weight)
|
||||
seen.add(tok)
|
||||
for tok, ids in inverted.items():
|
||||
ids.sort(key=lambda i: ranking[tok][i], reverse=True)
|
||||
return inverted, {t: {str(k): v for k, v in d.items()} for t, d in ranking.items()}
|
||||
inverted: dict[str, list[int]] = defaultdict(list)
|
||||
ranking: dict[str, dict[int, float]] = defaultdict(dict)
|
||||
for idx, e in enumerate(entries):
|
||||
fields = {"title": e["title"], "keywords": e["keywords"]}
|
||||
seen: set[str] = set()
|
||||
for field, text in fields.items():
|
||||
weight = FIELD_WEIGHT.get(field, 0.2)
|
||||
for tok in tokenize(text):
|
||||
if idx not in inverted[tok]:
|
||||
inverted[tok].append(idx)
|
||||
ranking[tok][idx] = max(ranking[tok].get(idx, 0.0), weight)
|
||||
seen.add(tok)
|
||||
for tok, ids in inverted.items():
|
||||
ids.sort(key=lambda i: ranking[tok][i], reverse=True)
|
||||
return inverted, {
|
||||
t: {str(k): v for k, v in d.items()} for t, d in ranking.items()
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) != 3:
|
||||
print(__doc__)
|
||||
return 1
|
||||
settings = Path(sys.argv[1])
|
||||
out = Path(sys.argv[2])
|
||||
files = discover_files(settings)
|
||||
nav = build_nav_map(settings, files)
|
||||
entries = extract_settings(files, nav)
|
||||
inverted, ranking = build_inverted_and_ranking(entries)
|
||||
for e in entries:
|
||||
e.pop("keywords", None)
|
||||
out.write_text(json.dumps({
|
||||
"version": 2,
|
||||
"entries": entries,
|
||||
"inverted": inverted,
|
||||
"ranking": ranking,
|
||||
}, ensure_ascii=False, indent=2))
|
||||
print(f"settings index: {len(entries)} entries, "
|
||||
f"{len(inverted)} tokens -> {out}")
|
||||
print("files:", len(files))
|
||||
print("comps:", len(parse_page_comps(settings)))
|
||||
print("registry:", len(parse_page_registry(settings)))
|
||||
print("nav:", len(nav))
|
||||
print("entries:", len(entries))
|
||||
return 0
|
||||
if len(sys.argv) != 3:
|
||||
print(__doc__)
|
||||
return 1
|
||||
settings = Path(sys.argv[1])
|
||||
out = Path(sys.argv[2])
|
||||
files = discover_files(settings)
|
||||
nav = build_nav_map(settings, files)
|
||||
entries = extract_settings(files, nav)
|
||||
inverted, ranking = build_inverted_and_ranking(entries)
|
||||
for e in entries:
|
||||
e.pop("keywords", None)
|
||||
out.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"version": 2,
|
||||
"entries": entries,
|
||||
"inverted": inverted,
|
||||
"ranking": ranking,
|
||||
},
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
)
|
||||
print(
|
||||
f"settings index: {len(entries)} entries, "
|
||||
f"{len(inverted)} tokens -> {out}"
|
||||
)
|
||||
print("files:", len(files))
|
||||
print("comps:", len(parse_page_comps(settings)))
|
||||
print("registry:", len(parse_page_registry(settings)))
|
||||
print("nav:", len(nav))
|
||||
print("entries:", len(entries))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
sys.exit(main())
|
||||
|
||||
Reference in New Issue
Block a user