Spaces:
Running
on
Zero
Running
on
Zero
File size: 3,475 Bytes
3427608 243f5a6 3427608 243f5a6 3427608 243f5a6 3427608 243f5a6 3427608 243f5a6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 |
import paramiko
import numpy as np
import io, os
from PIL import Image
import requests
import json
from .constants import SSH_SERVER, SSH_PORT, SSH_USER, SSH_PASSWORD, SSH_LOG
def create_ssh_client(server, port, user, password):
ssh = paramiko.SSHClient()
ssh.load_system_host_keys()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(server, port, user, password)
return ssh
def get_image_from_url(image_url):
response = requests.get(image_url)
response.raise_for_status() # success
return Image.open(io.BytesIO(response.content))
# def upload_image_via_sftp(ssh, image, remote_image_path):
# if isinstance(image, str):
# print("get url image")
# image = get_image_from_url(image)
# with ssh.open_sftp() as sftp:
# with io.BytesIO() as image_byte_stream:
# image.save(image_byte_stream, format='JPEG')
# image_byte_stream.seek(0)
# sftp.putfo(image_byte_stream, remote_image_path)
# print(f"Successfully uploaded image to {remote_image_path}")
# def upload_json_via_sftp(ssh, data, remote_json_path):
# json_data = json.dumps(data, indent=4)
# with ssh.open_sftp() as sftp:
# with io.BytesIO(json_data.encode('utf-8')) as json_byte_stream:
# sftp.putfo(json_byte_stream, remote_json_path)
# print(f"Successfully uploaded JSON data to {remote_json_path}")
def upload_image(states, output_dir):
pass
# ssh = create_ssh_client(SSH_SERVER, SSH_PORT, SSH_USER, SSH_PASSWORD)
# for i in range(len(states)):
# output_file = os.path.join(output_dir, f"{i}.jpg")
# upload_image_via_sftp(ssh, states[i].output, output_file)
def upload_informance(data, data_path):
pass
# ssh = create_ssh_client(SSH_SERVER, SSH_PORT, SSH_USER, SSH_PASSWORD)
# upload_json_via_sftp(ssh, data, data_path)
def create_remote_directory(remote_directory):
ssh = create_ssh_client(SSH_SERVER, SSH_PORT, SSH_USER, SSH_PASSWORD)
stdin, stdout, stderr = ssh.exec_command(f'mkdir -p {SSH_LOG}/{remote_directory}')
error = stderr.read().decode('utf-8')
if error:
print(f"Error: {error}")
else:
print(f"Directory {remote_directory} created successfully.")
# ssh.close()
return f'{SSH_LOG}/{remote_directory}'
def upload_ssh_all(states, output_dir, data, data_path):
ssh = create_ssh_client(SSH_SERVER, SSH_PORT, SSH_USER, SSH_PASSWORD)
output_file_list = []
image_list = []
for i in range(len(states)):
output_file = os.path.join(output_dir, f"{i}.jpg")
output_file_list.append(output_file)
image_list.append(states[i].output)
with ssh.open_sftp() as sftp:
for i in range(len(output_file_list)):
if isinstance(image_list[i], str):
print("get url image")
image_list[i] = get_image_from_url(image_list[i])
with io.BytesIO() as image_byte_stream:
image_list[i].save(image_byte_stream, format='JPEG')
image_byte_stream.seek(0)
sftp.putfo(image_byte_stream, output_file_list[i])
print(f"Successfully uploaded image to {output_file_list[i]}")
json_data = json.dumps(data, indent=4)
with io.BytesIO(json_data.encode('utf-8')) as json_byte_stream:
sftp.putfo(json_byte_stream, data_path)
print(f"Successfully uploaded JSON data to {data_path}") |