nexuscosmos 0.2.0

A powerful, dataset-agnostic Python library for astronomical data acquisition with built-in caching, rate limiting, async support, and progress tracking

🌌 NexusCosmos

A powerful, dataset-agnostic Python library for astronomical data acquisition.

NexusCosmos provides a unified interface to fetch astronomical data from any API source — whether it's NASA JPL Horizons, Minor Planet Center, ESA GAIA, or your own custom endpoints. Built with modern Python best practices, it handles all the complex infrastructure so you can focus on the science.

---

✨ Key Features

Feature	Description
🔌 Any Data Source	Works with any REST API - just provide the URL
⚡ Async Support	Fetch 100+ objects in parallel (3-10x faster)
📊 Progress Tracking	Visual progress bars for batch operations
💾 Smart Caching	File-based TTL cache reduces redundant API calls
🚦 Rate Limiting	Built-in rate limiter respects API limits
🔄 Auto Retry	Exponential backoff handles transient failures
🛡️ Error Recovery	Falls back to cached data when APIs are unavailable
🔍 Request Inspection	Debug exactly what's being sent/received
📤 Data Export	Export to JSON, CSV, pandas DataFrame, Astropy Table

---

📦 Installation

# Basic installation
pip install nexuscosmos

# With optional dependencies
pip install nexuscosmos[async]      # For async support (aiohttp)
pip install nexuscosmos[progress]   # For progress bars (tqdm)
pip install nexuscosmos[pandas]     # For DataFrame export
pip install nexuscosmos[astropy]    # For Astropy Table export
pip install nexuscosmos[all]        # Everything

---

🚀 Quick Start

Basic Usage (Any API)

from nexuscosmos import GenericAcquisitionClient, QueryConfig

# Connect to ANY astronomical API
client = GenericAcquisitionClient(
    base_url="https://api.example.com/data",
    show_progress=True,    # Show progress bars
    debug=True             # Enable debug logging
)

# Configure your query
config = QueryConfig(
    object_id="asteroid_123",
    start_time="2025-01-01",
    stop_time="2025-01-31",
    step_size="1d"
)

# Fetch data
result = client.fetch_data(config)

if result['success']:
    print(f"Data: {result['data']}")

Batch Operations (Multiple Objects)

# Fetch data for multiple objects at once
configs = [
    QueryConfig(object_id=f"object_{i}", start_time="2025-01-01", stop_time="2025-01-31")
    for i in range(50)
]

# Sequential with progress bar
results = client.fetch_batch(configs, show_progress=True)
# Output: Fetching data: 100%|████████████| 50/50 [00:30<00:00, 1.67obj/s]

Async Operations (3-10x Faster)

import asyncio
from nexuscosmos import AsyncGenericAcquisitionClient, QueryConfig

async def fetch_many():
    client = AsyncGenericAcquisitionClient(
        base_url="https://api.example.com/data",
        max_concurrent=10,    # 10 parallel requests
        show_progress=True
    )
    
    configs = [
        QueryConfig(object_id=f"object_{i}", start_time="2025-01-01", stop_time="2025-01-31")
        for i in range(100)
    ]
    
    # Fetch all in parallel - MUCH faster!
    results = await client.fetch_data_async_batch(configs)
    return results

results = asyncio.run(fetch_many())
print(f"Fetched {len(results)} objects")

---

🔭 Using with JPL Horizons

NexusCosmos includes a ready-to-use client for NASA JPL Horizons:

from nexuscosmos import HorizonsClient, QueryConfig

# Pre-configured for JPL Horizons API
client = HorizonsClient()

# Quick presets for common queries
config = QueryConfig.quick_ephemeris(object_id="1I")  # 1I/'Oumuamua
result = client.fetch_data(config)

# Supports interstellar objects with friendly names
# "1I", "Oumuamua", "2I/Borisov", "3I/ATLAS" all work!

---

💾 Data Export

from nexuscosmos import DataExporter

# After fetching data...
result = client.fetch_data(config)
exporter = DataExporter(result['data'])

# Export to various formats
json_str = exporter.to_json(indent=2)
csv_str = exporter.to_csv()
df = exporter.to_dataframe()           # Requires pandas
table = exporter.to_astropy_table()    # Requires astropy

# Save directly to files
exporter.save_json("output.json")
exporter.save_csv("output.csv")

---

🔧 Advanced Configuration

Client Options

client = GenericAcquisitionClient(
    base_url="https://api.example.com",
    
    # Rate Limiting
    enforce_rate_limit=True,
    max_requests_per_minute=60,
    
    # Caching
    cache=FileCache(cache_dir="./cache", ttl_seconds=86400),  # 24hr cache
    
    # Resilience
    fallback_to_cache=True,    # Use stale cache if API fails
    timeout_seconds=30,
    
    # Debugging
    debug=True,
    show_progress=True
)

Request Inspection

# After a request, inspect what happened
result = client.fetch_data(config)

print(f"URL called: {client.last_request_url}")
print(f"Parameters: {client.last_request_params}")
print(f"Status code: {client.last_response_status}")
print(f"Response time: {client.last_response_time:.2f}s")
print(f"Error (if any): {client.last_error}")

Custom Rate Limiter & Backoff

from nexuscosmos import RateLimiter, ExponentialBackoff, FileCache

client = GenericAcquisitionClient(
    base_url="https://api.example.com",
    rate_limiter=RateLimiter(max_attempts=100, window_minutes=1),
    backoff=ExponentialBackoff(base_delay=2.0, max_delay=120.0, multiplier=3.0),
    cache=FileCache(cache_dir="./my_cache", ttl_seconds=3600)
)

---

🏗️ Extending for Custom APIs

Create your own client by extending the base class:

from nexuscosmos import BaseAcquisitionClient, BaseTextParser

class MySpaceAgencyClient(BaseAcquisitionClient):
    """Client for My Space Agency's API."""
    
    BASE_URL = "https://api.myspaceagency.org/v2"
    
    def __init__(self, api_key: str, **kwargs):
        super().__init__(**kwargs)
        self.api_key = api_key
        self.base_url = self.BASE_URL
    
    def fetch_data(self, query_config):
        # Your custom implementation
        params = {
            "target": query_config.object_id,
            "start": query_config.start_time,
            "end": query_config.stop_time,
            "api_key": self.api_key
        }
        # ... fetch and parse

---

📊 Data Models

EphemerisData

Flexible container for astronomical observations:

from nexuscosmos import EphemerisData

ephem = EphemerisData(
    datetime_str="2025-01-15 12:00:00",
    RA=45.5,                    # Right Ascension (degrees)
    DEC=30.2,                   # Declination (degrees)
    delta=1.5,                  # Distance from observer (AU)
    r=1.2,                      # Heliocentric distance (AU)
    V_mag=15.5,                 # Visual magnitude
    # ... many more optional fields
)

OrbitalElements

Container for orbital parameters:

from nexuscosmos import OrbitalElements

orbit = OrbitalElements(
    epoch="2025-01-15",
    a=2.5,      # Semi-major axis (AU)
    e=0.95,     # Eccentricity
    i=135.0,    # Inclination (degrees)
    # ...
)

# Automatic orbit type detection
orbit.is_elliptical()    # True if e < 1
orbit.is_hyperbolic()    # True if e > 1
orbit.is_parabolic()     # True if e ≈ 1

---

🧪 Testing

# Run tests
pytest tests/ -v

# With coverage
pytest tests/ --cov=nexuscosmos --cov-report=html

---

📁 Project Structure

nexuscosmos/
├── __init__.py              # Package exports
├── acquisition_base.py      # Generic & Async clients
├── config.py                # QueryConfig, ClientConfig
├── models.py                # EphemerisData, OrbitalElements
├── cache.py                 # FileCache with TTL
├── utils.py                 # RateLimiter, ExponentialBackoff
├── export.py                # DataExporter (JSON, CSV, DataFrame)
├── parsers_base.py          # Base parser classes
└── datasets/
    └── horizons/            # JPL Horizons implementation
        ├── client.py        # HorizonsClient
        └── parser.py        # HorizonsParser

---

🤝 Contributing

Contributions are welcome! Areas we'd love help with:

• New Dataset Clients: Minor Planet Center, ESA GAIA, etc.
• Documentation: Tutorials, examples, API docs
• Testing: More test coverage
• Features: See Issues

---

📄 License

MIT License - see LICENSE for details.

---

🙏 Acknowledgments

• NASA JPL Horizons for their excellent ephemeris API
• The Python astronomy community
• All contributors and users

---

Made with ❤️ for the astronomy community

"The cosmos is within us. We are made of star-stuff." — Carl Sagan