Created
June 21, 2024 14:52
-
-
Save EleotleCram/e917f48f1cc5c9284ea68d9a7c654ebd to your computer and use it in GitHub Desktop.
Manage Tasmota UFS from python
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python3 | |
import sys | |
import os | |
import requests | |
# Function to show help message | |
def show_help(): | |
print("Usage: tasmota_ufs.py --hostname TASMOTA_DEVICE [OPTIONS] CMD [ARGS]") | |
print("") | |
print("Commands:") | |
print(" PUT FILE DEST_PATH Upload a file to the specified destination path.") | |
print(" GET FILE_PATH [-o OUTPUT_FILE] Download a file from the specified path.") | |
print(" GET FILE_PATH TARGET_DIR Download a file from the specified path to the target directory.") | |
print(" DEL FILE_PATH Delete a file at the specified path.") | |
print("") | |
print("Options:") | |
print(" --hostname TASMOTA_DEVICE Specify the Tasmota device hostname.") | |
print(" -q, --quiet Enable quiet mode, suppressing success or failure messages.") | |
print(" --help Show this help message and exit.") | |
sys.exit(1) | |
# Function to upload a file | |
def put_file(hostname, file, dest_path, quiet): | |
try: | |
url = f"http://{hostname}/ufsd?download={dest_path}" | |
response = requests.get(url, verify=False) | |
if response.status_code == 200: | |
url = f"http://{hostname}/ufsu" | |
with open(file, 'rb') as f: | |
files = {'ufsu': (os.path.basename(file), f)} | |
response = requests.post(url, files=files, verify=False) | |
if response.status_code == 200: | |
if not quiet: | |
print("File uploaded successfully.") | |
return 0 | |
else: | |
if not quiet: | |
print("File upload failed.") | |
return 1 | |
else: | |
if not quiet: | |
print("File upload failed.") | |
return 1 | |
except Exception: | |
return 1 | |
# Function to download a file | |
def get_file(hostname, file_path, output_file, quiet): | |
try: | |
url = f"http://{hostname}/ufsd?download={file_path}" | |
response = requests.get(url, stream=True, verify=False) | |
if response.status_code == 200: | |
with open(output_file, 'wb') as f: | |
for chunk in response.iter_content(chunk_size=8192): | |
f.write(chunk) | |
if not quiet: | |
print("File downloaded successfully.") | |
return 0 | |
else: | |
if not quiet: | |
print("File download failed.") | |
return 1 | |
except Exception: | |
if not quiet: | |
print("File download failed.") | |
return 1 | |
# Function to download a file to a target directory | |
def get_file_to_dir(hostname, file_path, target_dir, quiet): | |
output_file = os.path.join(target_dir, os.path.basename(file_path)) | |
return get_file(hostname, file_path, output_file, quiet) | |
# Function to delete a file | |
def del_file(hostname, file_path, quiet): | |
try: | |
url = f"http://{hostname}/ufsd?delete={file_path}" | |
response = requests.get(url, verify=False) | |
if response.status_code == 200: | |
if not quiet: | |
print("File deleted successfully.") | |
return 0 | |
else: | |
if not quiet: | |
print("File deletion failed.") | |
return 1 | |
except Exception: | |
return 1 | |
def main(): | |
if len(sys.argv) < 2: | |
show_help() | |
hostname = None | |
quiet = False | |
cmd = None | |
cmd_args = [] | |
# Parse arguments manually | |
i = 1 | |
while i < len(sys.argv): | |
arg = sys.argv[i] | |
if arg == "--hostname": | |
i += 1 | |
if i < len(sys.argv): | |
hostname = sys.argv[i] | |
else: | |
show_help() | |
elif arg == "-q" or arg == "--quiet": | |
quiet = True | |
elif arg == "--help": | |
show_help() | |
elif arg in ["PUT", "GET", "DEL"]: | |
cmd = arg | |
cmd_args = sys.argv[i + 1:] | |
break | |
else: | |
print(f"Unknown option: {arg}") | |
show_help() | |
i += 1 | |
if not hostname: | |
print("Error: --hostname option is required.") | |
show_help() | |
if not cmd: | |
print("Error: No command specified.") | |
show_help() | |
if cmd == "PUT": | |
if len(cmd_args) != 2: | |
print("Error: PUT requires 2 arguments (FILE DEST_PATH)") | |
show_help() | |
file, dest_path = cmd_args | |
sys.exit(put_file(hostname, file, dest_path, quiet)) | |
elif cmd == "GET": | |
if len(cmd_args) == 1: | |
file_path = cmd_args[0] | |
output_file = os.path.basename(file_path) | |
sys.exit(get_file(hostname, file_path, output_file, quiet)) | |
elif len(cmd_args) == 3 and cmd_args[1] == "-o": | |
file_path, _, output_file = cmd_args | |
sys.exit(get_file(hostname, file_path, output_file, quiet)) | |
elif len(cmd_args) == 2 and os.path.isdir(cmd_args[1]): | |
file_path, target_dir = cmd_args | |
sys.exit(get_file_to_dir(hostname, file_path, target_dir, quiet)) | |
else: | |
print("Error: Invalid arguments for GET command.") | |
show_help() | |
elif cmd == "DEL": | |
if len(cmd_args) != 1: | |
print("Error: DEL requires 1 argument (FILE_PATH)") | |
show_help() | |
file_path = cmd_args[0] | |
sys.exit(del_file(hostname, file_path, quiet)) | |
if __name__ == "__main__": | |
main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment