File size: 5,228 Bytes
31c0f24 eb2aa42 31c0f24 eb2aa42 31c0f24 | 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 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | import requests
import json
import os
import time
from datetime import datetime
from huggingface_hub import InferenceClient
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class WaveDataPuller:
def __init__(self, space_url="https://huggingface.co/spaces/nakas/NWPS_SWAN"):
self.space_url = space_url
self.api_url = f"{space_url}/api/predict"
self.output_dir = os.getenv('OUTPUT_DIR', '/tmp/wave_data')
self.poll_interval = int(os.getenv('POLL_INTERVAL', '3600')) # 1 hour default
# Create output directory with error handling for restricted environments
try:
os.makedirs(self.output_dir, exist_ok=True)
except PermissionError:
# Fall back to /tmp if we can't create in the specified location
self.output_dir = '/tmp/wave_data'
os.makedirs(self.output_dir, exist_ok=True)
# Initialize Hugging Face client
self.client = InferenceClient(model="nakas/NWPS_SWAN")
def fetch_wave_data(self):
"""Fetch wave data from the Hugging Face space"""
try:
logger.info("Fetching wave data from Hugging Face space...")
# Try different endpoints that might be available
endpoints_to_try = [
f"{self.space_url}/api/predict",
f"{self.space_url}/api/data",
f"{self.space_url}/gradio_api/call/predict",
]
for endpoint in endpoints_to_try:
try:
response = requests.get(endpoint, timeout=30)
if response.status_code == 200:
return response.json()
except Exception as e:
logger.debug(f"Failed to fetch from {endpoint}: {e}")
continue
# If direct API calls fail, try using the inference client
try:
result = self.client.text_generation("get_wave_data", max_new_tokens=100)
return {"data": result, "source": "inference_client"}
except Exception as e:
logger.error(f"Inference client failed: {e}")
logger.warning("No accessible endpoints found, returning mock data")
return self.generate_mock_data()
except Exception as e:
logger.error(f"Error fetching wave data: {e}")
return None
def generate_mock_data(self):
"""Generate mock SWAN wave data for testing"""
import random
return {
"timestamp": datetime.utcnow().isoformat(),
"location": {"lat": 40.0, "lon": -74.0},
"wave_data": {
"significant_wave_height": round(random.uniform(0.5, 3.0), 2),
"peak_wave_period": round(random.uniform(4.0, 12.0), 2),
"wave_direction": round(random.uniform(0, 360), 1),
"wind_speed": round(random.uniform(2.0, 15.0), 2),
"wind_direction": round(random.uniform(0, 360), 1)
},
"model": "NWPS_SWAN",
"forecast_hours": [0, 6, 12, 18, 24]
}
def save_data(self, data):
"""Save wave data to file"""
if not data:
return
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
filename = f"wave_data_{timestamp}.json"
filepath = os.path.join(self.output_dir, filename)
try:
with open(filepath, 'w') as f:
json.dump(data, f, indent=2)
logger.info(f"Data saved to {filepath}")
except Exception as e:
logger.error(f"Error saving data: {e}")
def run_continuous(self):
"""Run continuous data pulling"""
logger.info(f"Starting continuous wave data pulling every {self.poll_interval} seconds")
while True:
try:
data = self.fetch_wave_data()
if data:
self.save_data(data)
logger.info("Wave data fetched and saved successfully")
else:
logger.warning("No data received")
time.sleep(self.poll_interval)
except KeyboardInterrupt:
logger.info("Stopping wave data puller...")
break
except Exception as e:
logger.error(f"Unexpected error: {e}")
time.sleep(60) # Wait 1 minute before retrying
def run_once(self):
"""Run a single data pull"""
logger.info("Fetching wave data once...")
data = self.fetch_wave_data()
if data:
self.save_data(data)
logger.info("Wave data fetched and saved successfully")
else:
logger.error("Failed to fetch wave data")
if __name__ == "__main__":
puller = WaveDataPuller()
# Check if we should run once or continuously
run_mode = os.getenv('RUN_MODE', 'once')
if run_mode.lower() == 'continuous':
puller.run_continuous()
else:
puller.run_once() |