...
This commit is contained in:
Sebastian Moser
1
scripts/nav/.nav_db
Normal file
1
scripts/nav/.nav_db
Normal file
@@ -0,0 +1 @@
|
||||
. m main.py
|
||||
@@ -1,5 +1,8 @@
|
||||
* H ~
|
||||
* O ..
|
||||
* W ~/work
|
||||
* / /
|
||||
~ w ~/work
|
||||
~/work c config
|
||||
~/work d downloads
|
||||
|
||||
|
||||
195
scripts/nav/lf-raw-mode-shell-pipe/main.py
Normal file
195
scripts/nav/lf-raw-mode-shell-pipe/main.py
Normal file
@@ -0,0 +1,195 @@
|
||||
|
||||
import shlex
|
||||
from pathlib import Path
|
||||
from os import path
|
||||
import os
|
||||
import argparse
|
||||
import sys, tty, termios
|
||||
import subprocess
|
||||
import signal
|
||||
|
||||
DB_FILE = "/home/me/work/config/scripts/nav/db"
|
||||
|
||||
|
||||
def main():
|
||||
pwd = Path(os.getcwd())
|
||||
db_matches = get_db_matches(pwd, DB_FILE)
|
||||
folder_db_matches = get_folder_db_matches(pwd)
|
||||
folder_matches = get_folder_matches(pwd)
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('-m', '--mode', help='mode of the program', type=str)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.mode == "lf":
|
||||
|
||||
cmd = "ps" # | grep lf | awk '{print $1}'"
|
||||
cmd2 = "ps | grep lf | awk '{print $1}'"
|
||||
#p = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE)
|
||||
|
||||
#result = subprocess.run(cmd2, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
||||
#pid = result.stdout.decode('utf-8').strip()
|
||||
|
||||
#print("hiiiiiiiiii:", b)
|
||||
#a = p.stdout.readlines()
|
||||
#a = subprocess.check_output(["ps"]).decode("utf-8")
|
||||
|
||||
#a = input("hello --" + f"/proc/{pid}/fd/0" + "--")
|
||||
|
||||
#signal.signal(signal)
|
||||
|
||||
out = open("/home/me/p1", "w")
|
||||
sys.stdout = out
|
||||
sys.stderr = out
|
||||
|
||||
|
||||
path = "/proc/2778149/fd/0"
|
||||
|
||||
print("my stdin:", os.readlink('/proc/self/fd/0'))
|
||||
print("lf stdin:", os.readlink(path))
|
||||
print("pid:", os.getpid())
|
||||
input()
|
||||
print("after first input")
|
||||
|
||||
file = open("/dev/pts/16", "r")
|
||||
os.setpgid(os.getpid(), 2778149)
|
||||
|
||||
subprocess.run(["ps", "-j"], stdout=out, stderr=out)
|
||||
|
||||
exec("file.read(1)")
|
||||
#file.read(1)
|
||||
|
||||
a = input()
|
||||
print("input was:", a)
|
||||
|
||||
out.close()
|
||||
file.close()
|
||||
|
||||
exit()
|
||||
|
||||
#out.write("test" + pid + "\n")
|
||||
out.flush()
|
||||
|
||||
#path = f"/proc/{pid}/fd/0"
|
||||
|
||||
|
||||
out.write(f"before path: {path}\n")
|
||||
out.flush()
|
||||
|
||||
os.system("echo from echo $SHELL > /home/me/p2")
|
||||
#os.system("/bin/bash /home/me/work/config/scripts/nav/run.sh 2>/home/me/p2 1>/home/me/p2")
|
||||
os.system(f"python /home/me/work/config/scripts/nav/test.py </home/me/test/file 2>/home/me/p2 1>/home/me/p2")
|
||||
|
||||
#os.system(f"cat {path}")
|
||||
#old_settings = termios.tcgetattr(fd)
|
||||
|
||||
try:
|
||||
|
||||
file = open(path, "r")
|
||||
tty.setraw(file.fileno())
|
||||
|
||||
out.write(f"before path: {file}\n")
|
||||
out.flush()
|
||||
|
||||
b = file.read(1)
|
||||
|
||||
out.write("got: " + b + "\n")
|
||||
out.flush()
|
||||
|
||||
file.close()
|
||||
except e:
|
||||
out.write("ERROR: " + str(e) + "\n")
|
||||
|
||||
#while True:
|
||||
#b = file.read(1)
|
||||
#out.write("got: " + b + "\n")
|
||||
|
||||
out.close()
|
||||
|
||||
|
||||
"""
|
||||
while True:
|
||||
fd = sys.stdin.fileno()
|
||||
print("fd", fd)
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
try:
|
||||
tty.setraw(sys.stdin.fileno())
|
||||
ch = sys.stdin.read(1)
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
if ch == "q":
|
||||
break
|
||||
print("hello: ", ch)
|
||||
"""
|
||||
|
||||
#os.system("echo hello-world")
|
||||
#while True:
|
||||
#print("test", a)
|
||||
#input()
|
||||
#input("end: ")
|
||||
|
||||
|
||||
else:
|
||||
print("test", my_resolve("~"))
|
||||
print("db:", db_matches)
|
||||
print("folder db:", folder_db_matches)
|
||||
print("folder:", folder_matches)
|
||||
|
||||
|
||||
def get_db_matches(directory, db_file):
|
||||
matches = []
|
||||
with open(db_file, "r") as file:
|
||||
for line in file.readlines():
|
||||
line = line.strip()
|
||||
if line == "":
|
||||
continue
|
||||
tmp = shlex.split(line)
|
||||
try:
|
||||
dir_in = tmp[0]
|
||||
shortcut = tmp[1]
|
||||
dest = tmp[2]
|
||||
except:
|
||||
eprint("db parse error on:", line)
|
||||
continue
|
||||
|
||||
if dir_in == "*":
|
||||
matches.append((shortcut, dest))
|
||||
|
||||
elif my_resolve(directory) == my_resolve(dir_in):
|
||||
matches.append((shortcut, dest))
|
||||
|
||||
return matches
|
||||
|
||||
def get_folder_matches(directory):
|
||||
matches = []
|
||||
ls = os.listdir(directory)
|
||||
|
||||
for path in ls:
|
||||
if path[0] == ".":
|
||||
path_as_list = list(str(path))
|
||||
path_as_list.pop(0)
|
||||
path = "".join(path_as_list)
|
||||
|
||||
|
||||
return matches
|
||||
|
||||
def get_folder_db_matches(directory):
|
||||
if os.path.exists(directory / ".nav_db"):
|
||||
return get_db_matches(directory, directory / ".nav_db")
|
||||
else:
|
||||
return []
|
||||
|
||||
def my_resolve(path):
|
||||
if path == ".":
|
||||
return Path(os.getcwd())
|
||||
if str(path)[0] == "~":
|
||||
path_as_list = list(str(path))
|
||||
path_as_list.pop(0)
|
||||
return Path(str(Path.home()) + "".join(path_as_list))
|
||||
|
||||
return path.resolve()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
7
scripts/nav/lf-raw-mode-shell-pipe/run.sh
Executable file
7
scripts/nav/lf-raw-mode-shell-pipe/run.sh
Executable file
@@ -0,0 +1,7 @@
|
||||
#!/bin/bash
|
||||
#
|
||||
echo from script
|
||||
#script=$(cat ./test.py)
|
||||
#python -c "$script"
|
||||
|
||||
python /home/me/work/config/scripts/nav/test.py #2>/home/me/p2 1>/home/me/p2
|
||||
33
scripts/nav/lf-raw-mode-shell-pipe/test.py
Normal file
33
scripts/nav/lf-raw-mode-shell-pipe/test.py
Normal file
@@ -0,0 +1,33 @@
|
||||
|
||||
print("hiiiiiiiiiiiiiiiiiiiii:")
|
||||
|
||||
#import sys
|
||||
import os
|
||||
|
||||
|
||||
#path = input("path: ")
|
||||
path = "/proc/2767310/fd/0"
|
||||
print("path:", path)
|
||||
|
||||
#os.system(f"cat {path}")
|
||||
#print("hi:", os.path.abspath(sys.stdin.name))
|
||||
print("my stdin:", os.readlink('/proc/self/fd/0'))
|
||||
print("lf stdin:", os.readlink(path))
|
||||
|
||||
#os.system(f"cat /dev/pts/16")
|
||||
#os.system("stty -F /proc/2763309/fd/0 -raw -icanon -echo; cat /proc/2763309/fd/0 > /home/me/p2")
|
||||
exit()
|
||||
|
||||
try:
|
||||
file = open(path, "r")
|
||||
|
||||
print(f"before path: {path}\n")
|
||||
print(f"before path: {file}\n")
|
||||
|
||||
b = file.read(1)
|
||||
|
||||
print("got: " + b + "\n")
|
||||
|
||||
file.close()
|
||||
except Exception as e:
|
||||
print("ERROR: " + str(e) + "\n")
|
||||
@@ -3,54 +3,251 @@ import shlex
|
||||
from pathlib import Path
|
||||
from os import path
|
||||
import os
|
||||
import argparse
|
||||
import sys, tty, termios
|
||||
import subprocess
|
||||
from subprocess import Popen
|
||||
import signal
|
||||
|
||||
DB_FILE = "/home/me/work/config/nav/db"
|
||||
DB_FILE = "/home/me/work/config/scripts/nav/db"
|
||||
|
||||
def debug(*args, **kargs):
|
||||
global out
|
||||
print(*args, **kargs, file=out)
|
||||
out.flush()
|
||||
|
||||
def main():
|
||||
pwd = Path(os.getcwd())
|
||||
db_matches = get_db_matches(pwd)
|
||||
folder_matches = get_folder_matches(pwd)
|
||||
|
||||
print(my_resolve("~/work/config"))
|
||||
print("db:", db_matches)
|
||||
print("folder:", folder_matches)
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('-m', '--mode', help='mode of the program', type=str)
|
||||
parser.add_argument('-c', '--char', help='mode of the program', type=str)
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.mode == "lf":
|
||||
pre_cd = None
|
||||
lf_id = os.environ["id"]
|
||||
pwd = Path(os.getcwd())
|
||||
if args.char is not None:
|
||||
dests = do_nav(args.char, pwd)
|
||||
if len(dests) == 1:
|
||||
pre_cd = my_resolve(dests[0], pwd=pwd)
|
||||
pwd = pre_cd
|
||||
|
||||
Popen(["lf", "-remote", f"send {lf_id} echo -- NAV --"])
|
||||
|
||||
# for debug
|
||||
global out
|
||||
#out = open("/home/me/p1", "w")
|
||||
#sys.stderr = out
|
||||
#sys.stdout = out
|
||||
#debug()
|
||||
|
||||
pid_lf = int(os.environ["lf_user_pid"])
|
||||
pgid = int(os.environ["lf_user_pgid"])
|
||||
|
||||
#for debug
|
||||
#print("pid_lf", pid_lf)
|
||||
#print("pgid", pgid)
|
||||
|
||||
path = f"/proc/{pid_lf}/fd/0"
|
||||
|
||||
os.setpgid(os.getpid(), pgid)
|
||||
|
||||
file = open(path, "r")
|
||||
|
||||
#clear filter
|
||||
#os.system(f'lf -remote "send $id setfilter"')
|
||||
#os.system('lf -remote "send $id echo -- NAV --"')
|
||||
|
||||
chars = []
|
||||
while True:
|
||||
print("-- HIIIIIIIIIII --")
|
||||
#Popen(["lf", "-remote", f"send {lf_id} echo -- NAV iiiiiiiiiiiiiiii --"])
|
||||
|
||||
c = file.read(1)
|
||||
#debug("-- HIIIIIIIIIII after --")
|
||||
|
||||
if pre_cd is not None:
|
||||
Popen(["lf", "-remote", f"send {lf_id} cd {pre_cd}"])
|
||||
pre_cd = None
|
||||
|
||||
if c == "\x20" or c == "\x1b": # exit nav mode, when space or ESC is pressed
|
||||
if len(chars) == 0:
|
||||
pass
|
||||
Popen(["lf", "-remote", f"send {lf_id} setfilter"])
|
||||
Popen(["lf", "-remote", f"send {lf_id} echo NAV Done"])
|
||||
print("NAV Done")
|
||||
break
|
||||
|
||||
if c == "\x03":
|
||||
chars = chars[0:len(chars)-1]
|
||||
|
||||
chars.append(c)
|
||||
|
||||
dests = do_nav(chars, pwd)
|
||||
|
||||
#for debug
|
||||
#debug("dests:", dests)
|
||||
|
||||
if len(dests) == 1:
|
||||
path_to_go = my_resolve(dests[0], pwd=pwd)
|
||||
Popen(["lf", "-remote", f"send {lf_id} setfilter"])
|
||||
|
||||
if os.path.isdir(path_to_go):
|
||||
#for debug
|
||||
#print("cding to:", path_to_go)
|
||||
|
||||
Popen(["lf", "-remote", f"send {lf_id} cd {path_to_go}"])
|
||||
print("-- NAV --")
|
||||
#Popen(["lf", "-remote", f"send {lf_id} echo -- NAV --"])
|
||||
else:
|
||||
Popen(["lf", "-remote", f"send {lf_id} select {path_to_go}"])
|
||||
Popen(["lf", "-remote", f"send {lf_id} echo NAV Done"])
|
||||
break
|
||||
|
||||
|
||||
def get_db_matches(directory):
|
||||
chars = []
|
||||
pwd = path_to_go
|
||||
|
||||
elif len(dests) == 0:
|
||||
chars = []
|
||||
|
||||
else:
|
||||
chars_as_string = "".join(chars)
|
||||
Popen(["lf", "-remote", f"send {lf_id} setfilter {chars_as_string}"])
|
||||
print("-- NAV --")
|
||||
#Popen(["lf", "-remote", f"send {lf_id} echo -- NAV --"])
|
||||
|
||||
#debug("got:", c, "chars:", chars)
|
||||
|
||||
file.close()
|
||||
|
||||
# for debug
|
||||
out.close()
|
||||
|
||||
|
||||
else:
|
||||
chars = []
|
||||
while True:
|
||||
fd = sys.stdin.fileno()
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
try:
|
||||
tty.setraw(sys.stdin.fileno())
|
||||
c = sys.stdin.read(1)
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
|
||||
if c == "\x20" or c == "\x1b": # exit nav mode, when space or ESC is pressed
|
||||
break
|
||||
|
||||
if c == "\x03":
|
||||
chars = chars[0:len(chars)-1]
|
||||
|
||||
dests = do_nav(chars, pwd)
|
||||
|
||||
if len(dests) == 1:
|
||||
path_to_go = my_resolve(dests[0], pwd=pwd)
|
||||
os.system("export")
|
||||
|
||||
elif len(dests) == 0:
|
||||
pass
|
||||
|
||||
else:
|
||||
pass
|
||||
|
||||
def do_nav(chars, pwd):
|
||||
db_matches = get_db_matches(pwd, DB_FILE)
|
||||
folder_db_matches = get_folder_db_matches(pwd)
|
||||
folder_items = list_folder(pwd)
|
||||
|
||||
# for debug
|
||||
#print("pwd:", pwd)
|
||||
#print("chars:", chars)
|
||||
|
||||
# first check if the char matches in what is in the main db
|
||||
# immediatly cd (return list with one el) if found
|
||||
for (match, dest) in db_matches:
|
||||
if match == chars[0]:
|
||||
return [dest]
|
||||
|
||||
# next check local db
|
||||
# but also immediatly cd when found
|
||||
for (match, dest) in folder_db_matches:
|
||||
if match == chars[0]:
|
||||
return [dest]
|
||||
|
||||
# then check folder contents
|
||||
# and return all that start with that char
|
||||
dests = []
|
||||
for item in folder_items:
|
||||
item_chars = item[0:len(chars)]
|
||||
search_chars = "".join(chars[0:len(chars)])
|
||||
if item_chars == search_chars:
|
||||
dests.append(item)
|
||||
return dests
|
||||
|
||||
|
||||
def get_db_matches(directory, db_file):
|
||||
matches = []
|
||||
with open(DB_FILE, "r") as file:
|
||||
with open(db_file, "r") as file:
|
||||
for line in file.readlines():
|
||||
line = line.strip()
|
||||
if line == "":
|
||||
continue
|
||||
tmp = shlex.split(line)
|
||||
try:
|
||||
dir_in = tmp[0]
|
||||
shortcut = tmp[1]
|
||||
dest = tmp[2]
|
||||
except:
|
||||
eprint("db parse error on:", line)
|
||||
continue
|
||||
|
||||
if dir_in == "*":
|
||||
matches.append((shortcut, dest))
|
||||
if dir_in == "~":
|
||||
#if directory == Path.home():
|
||||
|
||||
elif my_resolve(directory) == my_resolve(dir_in):
|
||||
matches.append((shortcut, dest))
|
||||
|
||||
return matches
|
||||
|
||||
def get_folder_matches(directory):
|
||||
def list_folder(directory):
|
||||
matches = []
|
||||
ls = os.listdir(directory)
|
||||
|
||||
return matches
|
||||
"""
|
||||
for path in ls:
|
||||
if path[0] == ".":
|
||||
path_as_list = list(str(path))
|
||||
path_as_list.pop(0)
|
||||
path = "".join(path_as_list)
|
||||
|
||||
def my_resolve(path):
|
||||
if str(path)[0] == "~":
|
||||
matches.append(path)
|
||||
"""
|
||||
|
||||
return ls
|
||||
|
||||
def get_folder_db_matches(directory):
|
||||
if os.path.exists(directory / ".nav_db"):
|
||||
return get_db_matches(directory, directory / ".nav_db")
|
||||
else:
|
||||
return []
|
||||
|
||||
def my_resolve(path, pwd=os.getcwd()):
|
||||
if path == ".":
|
||||
return Path(pwd)
|
||||
|
||||
elif str(path)[0] == "~":
|
||||
path_as_list = list(str(path))
|
||||
path_as_list.pop(0)
|
||||
print("path_as_list:", path_as_list)
|
||||
return Path(str(Path.home()) + "".join(path_as_list))
|
||||
|
||||
return path.resolve()
|
||||
elif str(path)[0] == "/":
|
||||
return Path(path)
|
||||
|
||||
else:
|
||||
return Path(pwd) / Path(path)
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user