File size: 2,574 Bytes
1738d47 9a51830 1738d47 |
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 |
import pandas as pd
from ctgan import CTGAN
from sklearn.preprocessing import LabelEncoder
import os
import json
import requests
import streamlit as st
def train_and_generate_synthetic(real_data, schema, output_path):
"""Trains a CTGAN model and generates synthetic data."""
categorical_cols = [col for col, dtype in zip(schema['columns'], schema['types']) if dtype == 'string']
# Store label encoders
label_encoders = {}
for col in categorical_cols:
le = LabelEncoder()
real_data[col] = le.fit_transform(real_data[col])
label_encoders[col] = le
# Train CTGAN
gan = CTGAN(epochs=300)
gan.fit(real_data, categorical_cols)
# Generate synthetic data
synthetic_data = gan.sample(schema['size'])
# Decode categorical columns
for col in categorical_cols:
synthetic_data[col] = label_encoders[col].inverse_transform(synthetic_data[col])
# Save to CSV
os.makedirs('outputs', exist_ok=True)
synthetic_data.to_csv(output_path, index=False)
print(f"✅ Synthetic data saved to {output_path}")
def generate_schema(prompt):
"""Fetches schema from an external API and validates JSON."""
API_URL = "https://infinitymatter-synthetic-data-generator.hf.space/"
headers = {"Authorization": f"Bearer {st.secrets['hf_token']}"}
try:
response = requests.post(API_URL, json={"prompt": prompt}, headers=headers)
print("🔍 Raw API Response:", response.text) # Debugging line
schema = response.json()
# Validate required keys
if 'columns' not in schema or 'types' not in schema or 'size' not in schema:
raise ValueError("❌ Invalid schema format! Expected keys: 'columns', 'types', 'size'")
print("✅ Valid Schema Received:", schema) # Debugging line
return schema
except json.JSONDecodeError:
print("❌ Failed to parse JSON response. API might be down or returning non-JSON data.")
return None
except requests.exceptions.RequestException as e:
print(f"❌ API request failed: {e}")
return None
def fetch_data(domain):
"""Fetches real data for the given domain and ensures it's a valid DataFrame."""
data_path = f"datasets/{domain}.csv"
if os.path.exists(data_path):
df = pd.read_csv(data_path)
if not isinstance(df, pd.DataFrame) or df.empty:
raise ValueError("❌ Loaded data is invalid!")
return df
else:
raise FileNotFoundError(f"❌ Dataset for {domain} not found.")
|