|
|
@ -36,10 +36,18 @@ class MyListener(ServiceListener): |
|
|
|
info = zc.get_service_info(type_, name) |
|
|
|
info = zc.get_service_info(type_, name) |
|
|
|
print(f"Service {name} added, service info: {info}") |
|
|
|
print(f"Service {name} added, service info: {info}") |
|
|
|
if info and info.addresses: |
|
|
|
if info and info.addresses: |
|
|
|
ip = socket.inet_ntoa(info.addresses[0]) |
|
|
|
# Only add if TXT record contains 'MiniDexed' |
|
|
|
if ip not in self.ip_list: |
|
|
|
txt_records = info.properties |
|
|
|
self.ip_list.append(ip) |
|
|
|
if txt_records: |
|
|
|
self.name_list.append(info.server.rstrip('.')) |
|
|
|
for k, v in txt_records.items(): |
|
|
|
|
|
|
|
# v may be bytes, decode if needed |
|
|
|
|
|
|
|
val = v.decode() if isinstance(v, bytes) else v |
|
|
|
|
|
|
|
if (b"MiniDexed" in k or b"MiniDexed" in v) or ("MiniDexed" in str(k) or "MiniDexed" in str(val)): |
|
|
|
|
|
|
|
ip = socket.inet_ntoa(info.addresses[0]) |
|
|
|
|
|
|
|
if ip not in self.ip_list: |
|
|
|
|
|
|
|
self.ip_list.append(ip) |
|
|
|
|
|
|
|
self.name_list.append(info.server.rstrip('.')) |
|
|
|
|
|
|
|
break |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Constants |
|
|
|
# Constants |
|
|
@ -82,6 +90,8 @@ def extract_zip(zip_path): |
|
|
|
if __name__ == "__main__": |
|
|
|
if __name__ == "__main__": |
|
|
|
parser = argparse.ArgumentParser(description="MiniDexed Updater") |
|
|
|
parser = argparse.ArgumentParser(description="MiniDexed Updater") |
|
|
|
parser.add_argument("-v", action="store_true", help="Enable verbose FTP debug output") |
|
|
|
parser.add_argument("-v", action="store_true", help="Enable verbose FTP debug output") |
|
|
|
|
|
|
|
parser.add_argument("--ip", type=str, help="IP address of the device to upload to (skip mDNS discovery)") |
|
|
|
|
|
|
|
parser.add_argument("--version", type=int, choices=[1,2,3], help="Version to upload: 1=Latest official, 2=Continuous, 3=Local build (skip prompt)") |
|
|
|
args = parser.parse_args() |
|
|
|
args = parser.parse_args() |
|
|
|
|
|
|
|
|
|
|
|
import time |
|
|
|
import time |
|
|
@ -97,16 +107,24 @@ if __name__ == "__main__": |
|
|
|
] |
|
|
|
] |
|
|
|
if has_local_build: |
|
|
|
if has_local_build: |
|
|
|
release_options.append(("Local build (from src/)", None)) |
|
|
|
release_options.append(("Local build (from src/)", None)) |
|
|
|
print("Which release do you want to update?") |
|
|
|
|
|
|
|
for idx, (desc, _) in enumerate(release_options): |
|
|
|
if args.version: |
|
|
|
print(f" [{idx+1}] {desc}") |
|
|
|
selected_idx = args.version - 1 |
|
|
|
while True: |
|
|
|
if selected_idx < 0 or selected_idx >= len(release_options): |
|
|
|
choice = input(f"Enter the number of your choice (1-{len(release_options)}): ").strip() |
|
|
|
print(f"Invalid version selection: {args.version}") |
|
|
|
if choice.isdigit() and 1 <= int(choice) <= len(release_options): |
|
|
|
sys.exit(1) |
|
|
|
selected_idx = int(choice)-1 |
|
|
|
github_url = release_options[selected_idx][1] |
|
|
|
github_url = release_options[selected_idx][1] |
|
|
|
else: |
|
|
|
break |
|
|
|
print("Which release do you want to update?") |
|
|
|
print("Invalid selection. Please enter a valid number.") |
|
|
|
for idx, (desc, _) in enumerate(release_options): |
|
|
|
|
|
|
|
print(f" [{idx+1}] {desc}") |
|
|
|
|
|
|
|
while True: |
|
|
|
|
|
|
|
choice = input(f"Enter the number of your choice (1-{len(release_options)}): ").strip() |
|
|
|
|
|
|
|
if choice.isdigit() and 1 <= int(choice) <= len(release_options): |
|
|
|
|
|
|
|
selected_idx = int(choice)-1 |
|
|
|
|
|
|
|
github_url = release_options[selected_idx][1] |
|
|
|
|
|
|
|
break |
|
|
|
|
|
|
|
print("Invalid selection. Please enter a valid number.") |
|
|
|
|
|
|
|
|
|
|
|
# If local build is selected, skip all GitHub/zip logic and do not register cleanup |
|
|
|
# If local build is selected, skip all GitHub/zip logic and do not register cleanup |
|
|
|
use_local_build = has_local_build and selected_idx == len(release_options)-1 |
|
|
|
use_local_build = has_local_build and selected_idx == len(release_options)-1 |
|
|
@ -167,29 +185,35 @@ if __name__ == "__main__": |
|
|
|
# Using mDNS to find the IP address of the device(s) that advertise the FTP service "_ftp._tcp." |
|
|
|
# Using mDNS to find the IP address of the device(s) that advertise the FTP service "_ftp._tcp." |
|
|
|
ip_addresses = [] |
|
|
|
ip_addresses = [] |
|
|
|
device_names = [] |
|
|
|
device_names = [] |
|
|
|
zeroconf = Zeroconf() |
|
|
|
selected_ip = None |
|
|
|
listener = MyListener(ip_addresses, device_names) |
|
|
|
selected_name = None |
|
|
|
browser = ServiceBrowser(zeroconf, "_ftp._tcp.local.", listener) |
|
|
|
if args.ip: |
|
|
|
try: |
|
|
|
selected_ip = args.ip |
|
|
|
print("Searching for devices...") |
|
|
|
selected_name = args.ip |
|
|
|
time.sleep(10) |
|
|
|
else: |
|
|
|
if ip_addresses: |
|
|
|
zeroconf = Zeroconf() |
|
|
|
print("Devices found:") |
|
|
|
listener = MyListener(ip_addresses, device_names) |
|
|
|
for idx, (name, ip) in enumerate(zip(device_names, ip_addresses)): |
|
|
|
browser = ServiceBrowser(zeroconf, "_ftp._tcp.local.", listener) |
|
|
|
print(f" [{idx+1}] {name} ({ip})") |
|
|
|
try: |
|
|
|
while True: |
|
|
|
print("Searching for devices...") |
|
|
|
selection = input(f"Enter the number of the device to upload to (1-{len(ip_addresses)}): ").strip() |
|
|
|
time.sleep(10) |
|
|
|
if selection.isdigit() and 1 <= int(selection) <= len(ip_addresses): |
|
|
|
if ip_addresses: |
|
|
|
selected_ip = ip_addresses[int(selection)-1] |
|
|
|
print("Devices found:") |
|
|
|
selected_name = device_names[int(selection)-1] |
|
|
|
for idx, (name, ip) in enumerate(zip(device_names, ip_addresses)): |
|
|
|
break |
|
|
|
print(f" [{idx+1}] {name} ({ip})") |
|
|
|
print("Invalid selection. Please enter a valid number.") |
|
|
|
while True: |
|
|
|
else: |
|
|
|
selection = input(f"Enter the number of the device to upload to (1-{len(ip_addresses)}): ").strip() |
|
|
|
print("No devices found.") |
|
|
|
if selection.isdigit() and 1 <= int(selection) <= len(ip_addresses): |
|
|
|
sys.exit(1) |
|
|
|
selected_ip = ip_addresses[int(selection)-1] |
|
|
|
finally: |
|
|
|
selected_name = device_names[int(selection)-1] |
|
|
|
zeroconf.close() |
|
|
|
break |
|
|
|
print("Devices found:", list(zip(device_names, ip_addresses))) |
|
|
|
print("Invalid selection. Please enter a valid number.") |
|
|
|
|
|
|
|
else: |
|
|
|
|
|
|
|
print("No devices found.") |
|
|
|
|
|
|
|
sys.exit(1) |
|
|
|
|
|
|
|
finally: |
|
|
|
|
|
|
|
zeroconf.close() |
|
|
|
|
|
|
|
print("Devices found:", list(zip(device_names, ip_addresses))) |
|
|
|
|
|
|
|
|
|
|
|
# Log into the selected device and upload the new version of MiniDexed |
|
|
|
# Log into the selected device and upload the new version of MiniDexed |
|
|
|
print(f"Connecting to {selected_name} ({selected_ip})...") |
|
|
|
print(f"Connecting to {selected_name} ({selected_ip})...") |
|
|
@ -319,7 +343,15 @@ if __name__ == "__main__": |
|
|
|
with open(local_path, 'rb') as f: |
|
|
|
with open(local_path, 'rb') as f: |
|
|
|
ftp.storbinary(f'STOR {remote_path}', f, 8192, callback=progress_callback) |
|
|
|
ftp.storbinary(f'STOR {remote_path}', f, 8192, callback=progress_callback) |
|
|
|
print(f"\nUploaded {file} to {selected_ip}.") |
|
|
|
print(f"\nUploaded {file} to {selected_ip}.") |
|
|
|
|
|
|
|
except ftplib.all_errors as e: |
|
|
|
|
|
|
|
print(f"FTP error: {e}") |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
try: |
|
|
|
ftp.sendcmd("BYE") |
|
|
|
ftp.sendcmd("BYE") |
|
|
|
print(f"Disconnected from {selected_ip}.") |
|
|
|
print(f"Disconnected from {selected_ip}.") |
|
|
|
except ftplib.all_errors as e: |
|
|
|
except ftplib.all_errors as e: |
|
|
|
print(f"FTP error: {e}") |
|
|
|
if str(e).strip().lower() == "timed out": |
|
|
|
|
|
|
|
# Suppress expected timeout after BYE |
|
|
|
|
|
|
|
print(f"Disconnected from {selected_ip}.") |
|
|
|
|
|
|
|
else: |
|
|
|
|
|
|
|
print(f"FTP error after BYE: {e}") |
|
|
|