Docker/pandora-box
Docker/pandora-box
1
0
Fork 0
mirror of https://github.com/dbarzin/pandora-box.git synced 2025-07-19 13:29:42 +02:00
Code Issues Activity

reformating

Browse source
This commit is contained in:
dbarzin 2022-06-11 21:06:59 +02:00
parent 7134251bb3
commit 1a29d021cf
6 changed files with 192 additions and 181 deletions
Download patch file Download diff file Expand all files Collapse all files

82
pandora-box.py
View file

@ -6,9 +6,10 @@ import time
import sys import sys
import pyudev import pyudev
import psutil import psutil
import os
# ----------------------------------------------------------- # -----------------------------------------------------------
# Variables # Config variables
# ----------------------------------------------------------- # -----------------------------------------------------------
NO_SCAN = True NO_SCAN = True
@ -20,6 +21,7 @@ PANDORA_ROOT_URL="http://127.0.0.1:6100"
# Screen # Screen
# ----------------------------------------------------------- # -----------------------------------------------------------
def intitCurses(): def intitCurses():
global screen global screen
screen = curses.initscr() screen = curses.initscr()
@ -30,32 +32,38 @@ def intitCurses():
curses.noecho() curses.noecho()
screen.clear() screen.clear()
def printStatus(strStatus): def printStatus(strStatus):
screen.addstr(12,0,'Status : %-32s' % strStatus) screen.addstr(12, 0, "Status : %-32s" % strStatus)
screen.refresh() screen.refresh()
def printFSLabel(strLabel): def printFSLabel(strLabel):
screen.addstr(13,0,'Device : %-32s' % strLabel) screen.addstr(13, 0, "Device : %-32s" % strLabel)
screen.refresh() screen.refresh()
def printAction(strAction): def printAction(strAction):
screen.addstr(14,0,'Action : %-64s' % strAction) screen.addstr(14, 0, "Action : %-64s" % strAction)
screen.refresh() screen.refresh()
def initBar(): def initBar():
global progress_win global progress_win
progress_win = curses.newwin(3, 62, 3, 16) progress_win = curses.newwin(3, 62, 3, 16)
progress_win.border(0) progress_win.border(0)
def updateBar(progress): def updateBar(progress):
global progress_win global progress_win
rangex = (60 / float(100)) * progress rangex = (60 / float(100)) * progress
pos = int(rangex) pos = int(rangex)
display = '#' display = "#"
if pos != 0: if pos != 0:
progress_win.addstr(1, pos, "{}".format(display)) progress_win.addstr(1, pos, "{}".format(display))
progress_win.refresh() progress_win.refresh()
def printScreen(): def printScreen():
screen.addstr(1, 0, " ██▓███ ▄▄▄ ███▄ █ ▓█████▄ ▒█████ ██▀███ ▄▄▄ ▄▄▄▄ ▒█████ ▒██ ██▒") screen.addstr(1, 0, " ██▓███ ▄▄▄ ███▄ █ ▓█████▄ ▒█████ ██▀███ ▄▄▄ ▄▄▄▄ ▒█████ ▒██ ██▒")
screen.addstr(2, 0, " ▓██░ ██▒▒████▄ ██ ▀█ █ ▒██▀ ██▌▒██▒ ██▒▓██ ▒ ██▒▒████▄ ▓█████▄ ▒██▒ ██▒▒▒ █ █ ▒░") screen.addstr(2, 0, " ▓██░ ██▒▒████▄ ██ ▀█ █ ▒██▀ ██▌▒██▒ ██▒▓██ ▒ ██▒▒████▄ ▓█████▄ ▒██▒ ██▒▒▒ █ █ ▒░")
@ -68,20 +76,22 @@ def printScreen():
screen.addstr(9, 0, " ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ") screen.addstr(9, 0, " ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ")
screen.addstr(10, 0, " ░ ░ ") screen.addstr(10, 0, " ░ ░ ")
printStatus("WAITING") printStatus("WAITING")
printFSLabel('') printFSLabel("")
printAction('') printAction("")
initBar() initBar()
updateBar(1) updateBar(1)
def endCurses(): def endCurses():
curses.endwin() curses.endwin()
curses.flushinp() curses.flushinp()
# ----------------------------------------------------------- # -----------------------------------------------------------
# device # device
# ----------------------------------------------------------- # -----------------------------------------------------------
# mount device
def mountDevice(device): def mountDevice(device):
if USB_AUTO_MOUNT: if USB_AUTO_MOUNT:
found = False found = False
@ -99,71 +109,77 @@ def mountDevice(device):
res = os.system("pmount " + device.device_node + " box") res = os.system("pmount " + device.device_node + " box")
# print("Return type: ", res) # print("Return type: ", res)
# unmount device
def umountDevice(): def umountDevice():
if not USB_AUTO_MOUNT: if not USB_AUTO_MOUNT:
printAction("unmount device /media/box") printAction("unmount device /media/box")
res = os.system("pumount /media/box") res = os.system("pumount /media/box")
# print("Return type: ", res) # print("Return type: ", res)
def deviceLoop(): def deviceLoop():
context = pyudev.Context() context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context) monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by('block') monitor.filter_by("block")
for device in iter(monitor.poll, None): for device in iter(monitor.poll, None):
if 'ID_FS_TYPE' in device: if "ID_FS_TYPE" in device:
if device.action == 'add': if device.action == "add":
if device.device_node[5:7] == 'sd' and device.get('DEVTYPE')=='partition': if device.device_node[5:7] == "sd" and device.get("DEVTYPE") == "partition":
# print("New device {}".format(device.device_node)) # print("New device {}".format(device.device_node))
mountDevice(device) mountDevice(device)
# display device type # display device type
printStatus("KEY INSERTED") printStatus("KEY INSERTED")
printFSLabel(device.get('ID_FS_LABEL')) printFSLabel(device.get("ID_FS_LABEL"))
if device.action == 'remove': if device.action == "remove":
if device.device_node[5:7] == 'sd' and device.get('DEVTYPE')=='partition': if device.device_node[5:7] == "sd" and device.get("DEVTYPE") == "partition":
printStatus("WAITING") printStatus("WAITING")
printAction('Device removed') printAction("Device removed")
printFSLabel('') printFSLabel("")
umountDevice() umountDevice()
# ----------------------------------------------------------- # -----------------------------------------------------------
# pandora # pandora
# ----------------------------------------------------------- # -----------------------------------------------------------
# scan device at mountPoint
def scan(mountPoint): def scan(mountPoint):
pp = pypandora.PyPandora(root_url=PANDORA_ROOT_URL) pp = pypandora.PyPandora(root_url=PANDORA_ROOT_URL)
for arg in sys.argv[1:]: for arg in sys.argv[1:]:
print(arg,end='',flush=True) print(arg, end="", flush=True)
print(":",end='',flush=True) print(":", end="", flush=True)
res = pp.submit_from_disk(arg) res = pp.submit_from_disk(arg)
while True: while True:
print('.',end='',flush=True) print(".", end="", flush=True)
time.sleep(1) time.sleep(1)
res = pp.task_status(res['taskId']) res = pp.task_status(res["taskId"])
if res['status']!='WAITING': if res["status"] != "WAITING":
break break
print(res['status']) print(res["status"])
# -------------------------------------- # --------------------------------------
def pandoraBox():
try:
intitCurses() intitCurses()
printScreen() printScreen()
deviceLoop() deviceLoop()
while True: finally:
key = screen.getch()
if key == curses.KEY_MOUSE:
break
if key == 27:
break
endCurses() endCurses()
# --------------------------------------
if __name__ == "__main__":
pandoraBox()

48
tests/detect-usb.py
View file

@ -7,32 +7,34 @@ import os
context = pyudev.Context() context = pyudev.Context()
monitor = pyudev.Monitor.from_netlink(context) monitor = pyudev.Monitor.from_netlink(context)
monitor.filter_by('block') monitor.filter_by("block")
autoMount = False autoMount = False
def printDeviceInfo(dev): def printDeviceInfo(dev):
print('') print("")
print('<BLOCK INFORMATION>') print("<BLOCK INFORMATION>")
print('Device name: %s' % dev.get('DEVNAME')) print("Device name: %s" % dev.get("DEVNAME"))
print('Device type: %s' % dev.get('DEVTYPE')) print("Device type: %s" % dev.get("DEVTYPE"))
print('Bus system: %s' % dev.get('ID_BUS')) print("Bus system: %s" % dev.get("ID_BUS"))
print('Partition label: %s' % dev.get('ID_FS_LABEL')) print("Partition label: %s" % dev.get("ID_FS_LABEL"))
print('FS: %s' % dev.get('ID_FS_SYSTEM_ID')) print("FS: %s" % dev.get("ID_FS_SYSTEM_ID"))
print('FS type: %s' % dev.get('ID_FS_TYPE')) print("FS type: %s" % dev.get("ID_FS_TYPE"))
print('Device usage: %s' % dev.get('ID_FS_USAGE')) print("Device usage: %s" % dev.get("ID_FS_USAGE"))
print('Device model: %s' % dev.get('ID_MODEL')) print("Device model: %s" % dev.get("ID_MODEL"))
print('Partition type: %s' % dev.get('ID_PART_TABLE_TYPE')) print("Partition type: %s" % dev.get("ID_PART_TABLE_TYPE"))
print('USB driver: %s' % dev.get('ID_USB_DRIVER')) print("USB driver: %s" % dev.get("ID_USB_DRIVER"))
print('Path id: %s' % dev.get('ID_PATH')) print("Path id: %s" % dev.get("ID_PATH"))
# print('Capacity: %s' % stdOut[0].strip()) # print('Capacity: %s' % stdOut[0].strip())
print('</BLOCK INFORMATION>') print("</BLOCK INFORMATION>")
# enumerate at device connection # enumerate at device connection
for device in iter(monitor.poll, None): for device in iter(monitor.poll, None):
if 'ID_FS_TYPE' in device: if "ID_FS_TYPE" in device:
if device.action == 'add': if device.action == "add":
if device.device_node[5:7] == 'sd' and device.get('DEVTYPE')=='partition': if device.device_node[5:7] == "sd" and device.get("DEVTYPE") == "partition":
printDeviceInfo(device) printDeviceInfo(device)
print("New device {}".format(device.device_node)) print("New device {}".format(device.device_node))
# loop until device is mounted # loop until device is mounted
@ -52,14 +54,10 @@ for device in iter(monitor.poll, None):
res = os.system("pmount " + device.device_node + " box") res = os.system("pmount " + device.device_node + " box")
print("Return type: ", res) print("Return type: ", res)
if device.action == 'remove': if device.action == "remove":
if device.device_node[5:7] == 'sd' and device.get('DEVTYPE')=='partition': if device.device_node[5:7] == "sd" and device.get("DEVTYPE") == "partition":
print('Device removed') print("Device removed")
if not autoMount: if not autoMount:
print("unmount device /media/box") print("unmount device /media/box")
res = os.system("pumount /media/box") res = os.system("pumount /media/box")
print("Return type: ", res) print("Return type: ", res)

7
tests/mouse-click.py
View file

@ -13,15 +13,12 @@ screen.clear()
while True: while True:
key = screen.getch() key = screen.getch()
screen.clear() screen.clear()
screen.addstr(0, 0, 'key: {}'.format(key)) screen.addstr(0, 0, "key: {}".format(key))
if key == curses.KEY_MOUSE: if key == curses.KEY_MOUSE:
_, x, y, _, button = curses.getmouse() _, x, y, _, button = curses.getmouse()
screen.addstr(1, 0, 'x, y, button = {}, {}, {}'.format(x, y, button)) screen.addstr(1, 0, "x, y, button = {}, {}, {}".format(x, y, button))
elif key == 27: elif key == 27:
break break
curses.endwin() curses.endwin()
curses.flushinp() curses.flushinp()

6
tests/progress-bar.py
View file

@ -5,20 +5,23 @@ import time
curses.initscr() curses.initscr()
def initBar(): def initBar():
global progress_win global progress_win
progress_win = curses.newwin(3, 62, 3, 10) progress_win = curses.newwin(3, 62, 3, 10)
progress_win.border(0) progress_win.border(0)
def updateBar(progress): def updateBar(progress):
global progress_win global progress_win
rangex = (60 / float(100)) * progress rangex = (60 / float(100)) * progress
pos = int(rangex) pos = int(rangex)
display = '#' display = "#"
if pos != 0: if pos != 0:
progress_win.addstr(1, pos, "{}".format(display)) progress_win.addstr(1, pos, "{}".format(display))
progress_win.refresh() progress_win.refresh()
initBar() initBar()
loading = 0 loading = 0
while loading < 100: while loading < 100:
@ -30,4 +33,3 @@ time.sleep(1)
curses.endwin() curses.endwin()
curses.flushinp() curses.flushinp()

16
tests/scan.py
View file

@ -4,23 +4,21 @@ import pypandora
import time import time
import sys import sys
pp = pypandora.PyPandora(root_url= 'http://127.0.0.1:6100') pp = pypandora.PyPandora(root_url="http://127.0.0.1:6100")
for arg in sys.argv[1:]: for arg in sys.argv[1:]:
print(arg,end='',flush=True) print(arg, end="", flush=True)
print(":",end='',flush=True) print(":", end="", flush=True)
res = pp.submit_from_disk(arg) res = pp.submit_from_disk(arg)
while True: while True:
print('.',end='',flush=True) print(".", end="", flush=True)
time.sleep(1) time.sleep(1)
res = pp.task_status(res['taskId']) res = pp.task_status(res["taskId"])
if res['status']!='WAITING': if res["status"] != "WAITING":
break break
print(res['status']) print(res["status"])

2
tests/screen.py
View file

@ -20,7 +20,7 @@ screen.addstr(7,0," ░▒ ░ ▒ ▒▒ ░░ ░░ ░ ▒░
screen.addstr(8, 0, " ░░ ░ ▒ ░ ░ ░ ░ ░ ░ ░ ░ ░ ▒ ░░ ░ ░ ▒ ░ ░ ░ ░ ░ ▒ ░ ░ ") screen.addstr(8, 0, " ░░ ░ ▒ ░ ░ ░ ░ ░ ░ ░ ░ ░ ▒ ░░ ░ ░ ▒ ░ ░ ░ ░ ░ ▒ ░ ░ ")
screen.addstr(9, 0, " ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ") screen.addstr(9, 0, " ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ░ ")
screen.addstr(10, 0, " ░ ░ ") screen.addstr(10, 0, " ░ ░ ")
screen.addstr(11,0,"READY."); screen.addstr(11, 0, "READY.")
while True: while True:
key = screen.getch() key = screen.getch()