This commit is contained in:
2026-07-30 22:31:36 +02:00
parent 21e902bd61
commit d1d18627cc
7 changed files with 135 additions and 76 deletions
@@ -7,12 +7,12 @@
# Requirements: # Requirements:
# - colorthief (python3 package) # too lazy to implement color extraction myself :D # - colorthief (python3 package) # too lazy to implement color extraction myself :D
import os
import sys
import argparse import argparse
import os
import subprocess import subprocess
from pathlib import Path import sys
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from threading import Lock from threading import Lock
MAX_WORKERS = 8 MAX_WORKERS = 8
@@ -112,7 +112,7 @@ def extract_color(image_path: str) -> str:
max_score = -1.0 max_score = -1.0
for color in palette: for color in palette:
h, s, v = rgb2hsv(*color) _, s, v = rgb2hsv(*color)
# Filter out undesirable colors # Filter out undesirable colors
# Too dark # Too dark
@@ -140,7 +140,7 @@ def match_color(color: str, palette: dict[str, str]) -> str:
"""Match the given #rrggbb color to the closest flavor in the palette.""" """Match the given #rrggbb color to the closest flavor in the palette."""
color = color.lower().strip().removeprefix("#") color = color.lower().strip().removeprefix("#")
target_rgb = hex2rgb(color) target_rgb = hex2rgb(color)
target_h, target_s, target_v = rgb2hsv(*target_rgb) target_h, target_s, _ = rgb2hsv(*target_rgb)
# Warn if not representative (nearly grayscale) # Warn if not representative (nearly grayscale)
if target_s < 5: if target_s < 5:
@@ -150,7 +150,7 @@ def match_color(color: str, palette: dict[str, str]) -> str:
def get_weighted_distance(hex_val: str) -> float: def get_weighted_distance(hex_val: str) -> float:
p_rgb = hex2rgb(hex_val) p_rgb = hex2rgb(hex_val)
p_h, p_s, p_v = rgb2hsv(*p_rgb) p_h, _, _ = rgb2hsv(*p_rgb)
# RGB distance with weighting # RGB distance with weighting
rmean = (target_rgb[0] + p_rgb[0]) / 2 rmean = (target_rgb[0] + p_rgb[0]) / 2
+29 -22
View File
@@ -1,21 +1,18 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import sys
import glob
import argparse import argparse
import select import glob
import os import os
import select
import struct import struct
import sys
EVENT_FORMAT = '@llHHi' EVENT_FORMAT = "@llHHi"
EVENT_SIZE = struct.calcsize(EVENT_FORMAT) EVENT_SIZE = struct.calcsize(EVENT_FORMAT)
EV_LED = 0x11 EV_LED = 0x11
LED_CODES = { LED_CODES = {"numlock": 0x00, "capslock": 0x01, "scrolllock": 0x02}
'numlock': 0x00,
'capslock': 0x01,
'scrolllock': 0x02
}
def get_led_state(led_type: str) -> int: def get_led_state(led_type: str) -> int:
pattern = f"/sys/class/leds/*::{led_type}/brightness" pattern = f"/sys/class/leds/*::{led_type}/brightness"
@@ -24,24 +21,26 @@ def get_led_state(led_type: str) -> int:
return 0 return 0
for path in paths: for path in paths:
try: try:
with open(path, 'r') as f: with open(path, "r") as f:
if int(f.read().strip()) > 0: if int(f.read().strip()) > 0:
return 1 return 1
except (IOError, ValueError, OSError): except (ValueError, OSError):
continue continue
return 0 return 0
def has_led_capability(event_path: str) -> bool: def has_led_capability(event_path: str) -> bool:
try: try:
basename = os.path.basename(event_path) basename = os.path.basename(event_path)
cap_path = f"/sys/class/input/{basename}/device/capabilities/led" cap_path = f"/sys/class/input/{basename}/device/capabilities/led"
if os.path.exists(cap_path): if os.path.exists(cap_path):
with open(cap_path, 'r') as f: with open(cap_path, "r") as f:
return f.read().strip() != "0" return f.read().strip() != "0"
except OSError: except OSError:
pass pass
return False return False
class DeviceMonitor: class DeviceMonitor:
def __init__(self, target_led: str): def __init__(self, target_led: str):
self.target_led = target_led self.target_led = target_led
@@ -64,7 +63,7 @@ class DeviceMonitor:
def scan_devices(self) -> None: def scan_devices(self) -> None:
current_fds = set(self.active_fds.keys()) current_fds = set(self.active_fds.keys())
paths = glob.glob('/dev/input/event*') paths = glob.glob("/dev/input/event*")
found_fds = set() found_fds = set()
for path in paths: for path in paths:
@@ -117,12 +116,17 @@ class DeviceMonitor:
events_count = len(data) // EVENT_SIZE events_count = len(data) // EVENT_SIZE
for i in range(events_count): for i in range(events_count):
chunk = data[i * EVENT_SIZE : (i + 1) * EVENT_SIZE] chunk = data[i * EVENT_SIZE : (i + 1) * EVENT_SIZE]
_, _, ev_type, ev_code, ev_value = struct.unpack(EVENT_FORMAT, chunk) _, _, ev_type, ev_code, ev_value = struct.unpack(
EVENT_FORMAT, chunk
)
if ev_type == EV_LED and ev_code == self.target_led_code: if (
if ev_value != self.last_state: ev_type == EV_LED
self.emit_state(ev_value) and ev_code == self.target_led_code
self.last_state = ev_value and ev_value != self.last_state
):
self.emit_state(ev_value)
self.last_state = ev_value
except BlockingIOError: except BlockingIOError:
pass pass
@@ -157,19 +161,22 @@ class DeviceMonitor:
except Exception: except Exception:
pass pass
def main(): def main():
parser = argparse.ArgumentParser(description="Zero-polling keyboard LED monitor.") parser = argparse.ArgumentParser(description="Zero-polling keyboard LED monitor.")
parser.add_argument( parser.add_argument(
'-l', '--led', "-l",
"--led",
type=str, type=str,
default='capslock', default="capslock",
choices=['capslock', 'numlock', 'scrolllock'], choices=["capslock", "numlock", "scrolllock"],
help="Target LED to monitor" help="Target LED to monitor",
) )
args = parser.parse_args() args = parser.parse_args()
monitor = DeviceMonitor(args.led) monitor = DeviceMonitor(args.led)
monitor.run() monitor.run()
if __name__ == "__main__": if __name__ == "__main__":
main() main()
+1 -1
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import sys
import subprocess import subprocess
import sys
def has_ping6() -> bool: def has_ping6() -> bool:
+20 -19
View File
@@ -11,12 +11,12 @@
# - glib bindings for python # - glib bindings for python
import argparse import argparse
import fcntl
import subprocess import subprocess
import time import time
import fcntl
from os import environ
from datetime import datetime from datetime import datetime
from enum import Enum from enum import Enum
from os import environ
from pathlib import Path from pathlib import Path
from shutil import copy2 from shutil import copy2
@@ -52,7 +52,7 @@ def take_screenshot(filepath: Path, typeStr: str):
lockFD = open("/tmp/screenshot-script.lock", "w") lockFD = open("/tmp/screenshot-script.lock", "w")
try: try:
fcntl.flock(lockFD, fcntl.LOCK_EX | fcntl.LOCK_NB) fcntl.flock(lockFD, fcntl.LOCK_EX | fcntl.LOCK_NB)
except IOError: except OSError:
lockFD.close() lockFD.close()
raise RuntimeError("Another screenshot is currently being taken.") raise RuntimeError("Another screenshot is currently being taken.")
@@ -66,11 +66,11 @@ def take_screenshot(filepath: Path, typeStr: str):
ScreenshotType.AREA: f"hyprshot -z -m region -o {SCREENSHOT_DIR} -f ", ScreenshotType.AREA: f"hyprshot -z -m region -o {SCREENSHOT_DIR} -f ",
ScreenshotType.WINDOW: f"hyprshot -z -m window -o {SCREENSHOT_DIR} -f ", ScreenshotType.WINDOW: f"hyprshot -z -m window -o {SCREENSHOT_DIR} -f ",
} }
process = subprocess.run(f"{cmd[type]}{filepath.name}", shell=True) subprocess.run(f"{cmd[type]}{filepath.name}", shell=True, check=True)
if process.returncode != 0:
raise RuntimeError("Failed to take screenshot: hyprshot command failed.")
if not wait_until_file_exists(filepath): if not wait_until_file_exists(filepath):
raise RuntimeError("Failed to take screenshot: output file not found after hyprshot command.") raise RuntimeError(
"Failed to take screenshot: output file not found after hyprshot command."
)
elif "niri" in currentDesktop: elif "niri" in currentDesktop:
niriScreenshotPath = SCREENSHOT_DIR / ".niri_screenshot.png" niriScreenshotPath = SCREENSHOT_DIR / ".niri_screenshot.png"
@@ -79,23 +79,24 @@ def take_screenshot(filepath: Path, typeStr: str):
# and the selection ui is drawn inside of niri without its state exposed to external programs. # and the selection ui is drawn inside of niri without its state exposed to external programs.
# so we use grim + slurp for area mode and niri's built-in commands for others. # so we use grim + slurp for area mode and niri's built-in commands for others.
ScreenshotType.FULL: "niri msg action screenshot-screen", ScreenshotType.FULL: "niri msg action screenshot-screen",
ScreenshotType.AREA: f" grim -g \"$(slurp)\" -t png {niriScreenshotPath} && cat {niriScreenshotPath} | wl-copy", ScreenshotType.AREA: f' grim -g "$(slurp)" -t png {niriScreenshotPath} && cat {niriScreenshotPath} | wl-copy',
ScreenshotType.WINDOW: "niri msg action screenshot-window", ScreenshotType.WINDOW: "niri msg action screenshot-window",
} }
if niriScreenshotPath.exists(): if niriScreenshotPath.exists():
niriScreenshotPath.unlink() niriScreenshotPath.unlink()
process = subprocess.run(cmd[type], shell=True) subprocess.run(cmd[type], shell=True, check=True)
if process.returncode != 0:
print(process.returncode)
raise RuntimeError("Failed to take screenshot: niri screenshot command failed.")
if wait_until_file_exists(niriScreenshotPath): if wait_until_file_exists(niriScreenshotPath):
# niriScreenshotPath.rename(filepath) # niriScreenshotPath.rename(filepath)
copy2(niriScreenshotPath, filepath) copy2(niriScreenshotPath, filepath)
else: else:
raise RuntimeError("Failed to take screenshot: output file not found after niri command.") raise RuntimeError(
"Failed to take screenshot: output file not found after niri command."
)
if not wait_until_file_exists(filepath): if not wait_until_file_exists(filepath):
raise RuntimeError("Failed to take screenshot: output file not found after copying.") raise RuntimeError(
"Failed to take screenshot: output file not found after copying."
)
else: else:
# print("Unsupported desktop environment.") # print("Unsupported desktop environment.")
@@ -106,12 +107,11 @@ def take_screenshot(filepath: Path, typeStr: str):
def edit_screenshot(filepath: Path): def edit_screenshot(filepath: Path):
subprocess.run(f"gradia {filepath}", shell=True) subprocess.run(f"gradia {filepath}", shell=True, check=True)
# subprocess.run(f"spectacle -l --edit-existing {filepath}", shell=True)
def gen_file_name(prefix="screenshot", ext=".png"): def gen_file_name(prefix="screenshot", ext=".png"):
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") timestamp = datetime.now(tz=datetime.now().astimezone().tzinfo).strftime("%Y-%m-%d_%H-%M-%S")
return f"{prefix}_{timestamp}{ext}" return f"{prefix}_{timestamp}{ext}"
@@ -135,7 +135,7 @@ if __name__ == "__main__":
args = parser.parse_args() args = parser.parse_args()
filepath: Path = Path() filepath: Path = Path()
if not args.type == ScreenshotType.EDIT.value: if args.type != ScreenshotType.EDIT.value:
# file path # file path
SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True) SCREENSHOT_DIR.mkdir(parents=True, exist_ok=True)
filename = gen_file_name() filename = gen_file_name()
@@ -166,7 +166,6 @@ if __name__ == "__main__":
# callback on close # callback on close
def close_callback(n): def close_callback(n):
global editing
if not editing: if not editing:
loop.quit() loop.quit()
@@ -175,6 +174,7 @@ if __name__ == "__main__":
"Click to edit", "Click to edit",
str(filepath), str(filepath),
) )
n.set_hint("transient", GLib.Variant("i", 1))
n.add_action( n.add_action(
# so default action is used, which will be triggered on simply clicking the notification card # so default action is used, which will be triggered on simply clicking the notification card
"default", "default",
@@ -193,5 +193,6 @@ if __name__ == "__main__":
n = Notify.Notification.new( n = Notify.Notification.new(
"Screenshot Error", "Screenshot Error",
str(e), str(e),
history=False,
) )
n.show() n.show()
+76 -25
View File
@@ -7,8 +7,8 @@ import shlex
import shutil import shutil
import subprocess import subprocess
import sys import sys
import tempfile
import tarfile import tarfile
import tempfile
import zipfile import zipfile
from pathlib import Path from pathlib import Path
@@ -20,14 +20,15 @@ SIZE_PATTERN = re.compile(r"^[1-9][0-9]*[KMGkmg]?$")
def download_to(url: str, dest_dir: Path) -> Path: def download_to(url: str, dest_dir: Path) -> Path:
# Import when needed # Import when needed
import requests import requests
local_filename = url.split('/')[-1]
local_filename = url.split("/")[-1]
dest_path = dest_dir / local_filename dest_path = dest_dir / local_filename
print(f"Downloading '{url}' to '{dest_path}'...") print(f"Downloading '{url}' to '{dest_path}'...")
try: try:
with requests.get(url, stream=True) as r: with requests.get(url, stream=True) as r:
r.raise_for_status() r.raise_for_status()
with open(dest_path, 'wb') as f: with open(dest_path, "wb") as f:
for chunk in r.iter_content(chunk_size=8192): for chunk in r.iter_content(chunk_size=8192):
f.write(chunk) f.write(chunk)
except Exception as e: except Exception as e:
@@ -59,7 +60,9 @@ def get_strip_count(archive_path: Path, is_tar: bool, is_zip: bool) -> int:
if is_tar: if is_tar:
cmd = ["tar", "-tf", str(archive_path)] cmd = ["tar", "-tf", str(archive_path)]
with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1) as proc: with subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, bufsize=1
) as proc:
if proc is None or proc.stdout is None: if proc is None or proc.stdout is None:
sys.exit("Error: Failed to list archive contents.") sys.exit("Error: Failed to list archive contents.")
for line in proc.stdout: for line in proc.stdout:
@@ -67,7 +70,7 @@ def get_strip_count(archive_path: Path, is_tar: bool, is_zip: bool) -> int:
if not line: if not line:
continue continue
current_top = line.split('/', 1)[0] current_top = line.split("/", 1)[0]
if top_level is None: if top_level is None:
top_level = current_top top_level = current_top
@@ -78,9 +81,9 @@ def get_strip_count(archive_path: Path, is_tar: bool, is_zip: bool) -> int:
proc.wait() proc.wait()
elif is_zip: elif is_zip:
with zipfile.ZipFile(archive_path, 'r') as zf: with zipfile.ZipFile(archive_path, "r") as zf:
for name in zf.namelist(): for name in zf.namelist():
current_top = name.split('/', 1)[0] current_top = name.split("/", 1)[0]
if top_level is None: if top_level is None:
top_level = current_top top_level = current_top
elif top_level != current_top: elif top_level != current_top:
@@ -101,29 +104,56 @@ def extract_archive(archive_path: Path, dest_dir: Path, strip_components: int =
elif zipfile.is_zipfile(archive_path): elif zipfile.is_zipfile(archive_path):
if strip_components > 0: if strip_components > 0:
with tempfile.TemporaryDirectory() as tmp_ext: with tempfile.TemporaryDirectory() as tmp_ext:
subprocess.run(["unzip", "-q", str(archive_path), "-d", tmp_ext], check=True) subprocess.run(
["unzip", "-q", str(archive_path), "-d", tmp_ext], check=True
)
top_dir = next(Path(tmp_ext).iterdir()) top_dir = next(Path(tmp_ext).iterdir())
for item in top_dir.iterdir(): for item in top_dir.iterdir():
shutil.move(str(item), str(dest_dir)) shutil.move(str(item), str(dest_dir))
else: else:
subprocess.run(["unzip", "-q", str(archive_path), "-d", str(dest_dir)], check=True) subprocess.run(
["unzip", "-q", str(archive_path), "-d", str(dest_dir)], check=True
)
def main(): def main():
parser = argparse.ArgumentParser(description="Extract an archive to a directory and spawn a shell.") parser = argparse.ArgumentParser(
description="Extract an archive to a directory and spawn a shell."
)
parser.add_argument("archive", type=str, help="Path to the tarball or zip file") parser.add_argument("archive", type=str, help="Path to the tarball or zip file")
parser.add_argument("--exec", "-e", dest="cmd", default=DEFAULT_SHELL, parser.add_argument(
help=f"Command to spawn (default: '{DEFAULT_SHELL}')") "--exec",
parser.add_argument("--target", "-t", type=Path, help="Target directory for extraction.") "-e",
parser.add_argument("--disable-tmpfs", "-d", action="store_true", help="Disable tmpfs mounting, extract directly to dir") dest="cmd",
parser.add_argument("--size", "-s", default=DEFAULT_TMPFS_SIZE, default=DEFAULT_SHELL,
help=f"Size of the tmpfs if used (default: {DEFAULT_TMPFS_SIZE})") help=f"Command to spawn (default: '{DEFAULT_SHELL}')",
parser.add_argument("--no-cleanup", "-n", action="store_true", help="Disable cleanup (unmount and remove dir)") )
parser.add_argument(
"--target", "-t", type=Path, help="Target directory for extraction."
)
parser.add_argument(
"--disable-tmpfs",
"-d",
action="store_true",
help="Disable tmpfs mounting, extract directly to dir",
)
parser.add_argument(
"--size",
"-s",
default=DEFAULT_TMPFS_SIZE,
help=f"Size of the tmpfs if used (default: {DEFAULT_TMPFS_SIZE})",
)
parser.add_argument(
"--no-cleanup",
"-n",
action="store_true",
help="Disable cleanup (unmount and remove dir)",
)
args = parser.parse_args() args = parser.parse_args()
archive = args.archive.strip() archive = args.archive.strip()
if archive.startswith(('https://', 'http://')): if archive.startswith(("https://", "http://")):
archive = download_to(archive, Path.cwd()) archive = download_to(archive, Path.cwd())
else: else:
archive = Path(args.archive).resolve() archive = Path(args.archive).resolve()
@@ -132,7 +162,9 @@ def main():
sys.exit(f"Error: Archive '{archive}' does not exist or is not a file.") sys.exit(f"Error: Archive '{archive}' does not exist or is not a file.")
if not SIZE_PATTERN.match(args.size): if not SIZE_PATTERN.match(args.size):
sys.exit(f"Error: Invalid size format '{args.size}'. Expected format like '4G', '500M'.") sys.exit(
f"Error: Invalid size format '{args.size}'. Expected format like '4G', '500M'."
)
is_tar = tarfile.is_tarfile(archive) is_tar = tarfile.is_tarfile(archive)
is_zip = zipfile.is_zipfile(archive) is_zip = zipfile.is_zipfile(archive)
@@ -155,14 +187,26 @@ def main():
else: else:
target_dir.mkdir(parents=True, exist_ok=True) target_dir.mkdir(parents=True, exist_ok=True)
strip_components = strip_components = get_strip_count(archive, is_tar, is_zip) strip_components = get_strip_count(archive, is_tar, is_zip)
try: try:
if use_tmpfs: if use_tmpfs:
print(f"Mounting tmpfs at '{target_dir}' with size {args.size}...") print(f"Mounting tmpfs at '{target_dir}' with size {args.size}...")
uid, gid = os.getuid(), os.getgid() uid, gid = os.getuid(), os.getgid()
mount_opts = f"size={args.size},uid={uid},gid={gid},mode=0700" mount_opts = f"size={args.size},uid={uid},gid={gid},mode=0700"
subprocess.run(["sudo", "mount", "-t", "tmpfs", "-o", mount_opts, "tmpfs", str(target_dir)], check=True) subprocess.run(
[
"sudo",
"mount",
"-t",
"tmpfs",
"-o",
mount_opts,
"tmpfs",
str(target_dir),
],
check=True,
)
print(f"Extracting '{archive}' to '{target_dir}'...") print(f"Extracting '{archive}' to '{target_dir}'...")
extract_archive(archive, target_dir, strip_components) extract_archive(archive, target_dir, strip_components)
@@ -170,19 +214,26 @@ def main():
print(f"Spawning '{args.cmd}' in {target_dir}...") print(f"Spawning '{args.cmd}' in {target_dir}...")
parsed_cmd = shlex.split(args.cmd) parsed_cmd = shlex.split(args.cmd)
subprocess.run(parsed_cmd, cwd=str(target_dir)) subprocess.run(parsed_cmd, cwd=str(target_dir), check=True)
finally: finally:
if do_cleanup: if do_cleanup:
print("Cleaning up...") print("Cleaning up...")
if use_tmpfs: if use_tmpfs:
subprocess.run(["sudo", "umount", str(target_dir)], stderr=subprocess.DEVNULL) subprocess.run(
["sudo", "umount", str(target_dir)],
stderr=subprocess.DEVNULL,
check=True,
)
if not dir_existed_before: if not dir_existed_before:
try: try:
shutil.rmtree(target_dir) shutil.rmtree(target_dir)
except PermissionError: except PermissionError:
if shutil.which("sudo"): subprocess.run(
subprocess.run(["sudo", "rm", "-rf", str(target_dir)], stderr=subprocess.DEVNULL) ["sudo", "rm", "-rf", str(target_dir)],
stderr=subprocess.DEVNULL,
check=True,
)
else: else:
print(f"Cleanup disabled. Extracted contents are left at: {target_dir}") print(f"Cleanup disabled. Extracted contents are left at: {target_dir}")
+1 -1
View File
@@ -7,7 +7,7 @@
zmodload zsh/datetime zmodload zsh/datetime
: ${uy_done_min_cmd_duration:=10} : ${uy_done_min_cmd_duration:=10}
: ${uy_done_exclude:='^(nvim|helix|hx|vim|vi|nano|less|more|man|ssh|top|htop|btop|sudoedit)$'} : ${uy_done_exclude:='^(nvim|helix|hx|vim|vi|nano|less|more|man|ssh|top|htop|btop|sudoedit|yazi)$'}
# Returns the id in $REPLY, empty if there is none or niri did not answer. # Returns the id in $REPLY, empty if there is none or niri did not answer.
if zmodload zsh/net/socket 2>/dev/null; then if zmodload zsh/net/socket 2>/dev/null; then
+2 -2
View File
@@ -1,11 +1,11 @@
[[plugin.deps]] [[plugin.deps]]
use = "yazi-rs/plugins:git" use = "yazi-rs/plugins:git"
rev = "4c63ed3" rev = "9014ed2"
hash = "88e56a64b7ce7c4314427452343fef17" hash = "88e56a64b7ce7c4314427452343fef17"
[[plugin.deps]] [[plugin.deps]]
use = "yazi-rs/plugins:smart-enter" use = "yazi-rs/plugins:smart-enter"
rev = "4c63ed3" rev = "9014ed2"
hash = "187cc58ba7ac3befd49c342129e6f1b6" hash = "187cc58ba7ac3befd49c342129e6f1b6"
[[plugin.deps]] [[plugin.deps]]