fbmc-chronos2 / ssh_helper.py
Evgueni Poloukarov
feat: complete Day 3 zero-shot inference pipeline
74bde7a
raw
history blame
2.66 kB
#!/usr/bin/env python3
"""SSH helper using paramiko for HuggingFace Space access"""
import paramiko
import sys
import os
def execute_ssh_command(command, timeout=60):
"""Execute command on HF Space via SSH and return output"""
# SSH configuration
hostname = "ssh.hf.space"
username = "evgueni-p-fbmc-chronos2-forecast"
key_file = "/c/Users/evgue/.ssh/id_ed25519"
# Convert Windows path for paramiko
key_file_win = "C:\\Users\\evgue\\.ssh\\id_ed25519"
try:
# Create SSH client
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# Load private key
private_key = paramiko.Ed25519Key.from_private_key_file(key_file_win)
# Connect
print(f"[*] Connecting to {hostname}...")
client.connect(
hostname=hostname,
username=username,
pkey=private_key,
timeout=30,
look_for_keys=False,
allow_agent=False
)
print("[+] Connected!")
# Execute command
print(f"[*] Executing: {command[:100]}...")
stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
# Get output
output = stdout.read().decode('utf-8')
error = stderr.read().decode('utf-8')
exit_code = stdout.channel.recv_exit_status()
client.close()
return {
'stdout': output,
'stderr': error,
'exit_code': exit_code,
'success': exit_code == 0
}
except Exception as e:
return {
'stdout': '',
'stderr': str(e),
'exit_code': -1,
'success': False
}
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python ssh_helper.py 'command to execute'")
sys.exit(1)
command = sys.argv[1]
result = execute_ssh_command(command)
print("\n" + "="*60)
print("STDOUT:")
print("="*60)
# Handle Unicode encoding for Windows console
try:
print(result['stdout'])
except UnicodeEncodeError:
print(result['stdout'].encode('ascii', errors='replace').decode('ascii'))
if result['stderr']:
print("\n" + "="*60)
print("STDERR:")
print("="*60)
# Handle Unicode encoding for Windows console
try:
print(result['stderr'])
except UnicodeEncodeError:
print(result['stderr'].encode('ascii', errors='replace').decode('ascii'))
print("\n" + "="*60)
print(f"Exit code: {result['exit_code']}")
print("="*60)
sys.exit(result['exit_code'])