astro-mmdc 0.1.1

Python SDK for the MMDC astrophysics platform — SED data access and blazar broadband emission modeling

astro-mmdc

Python SDK for the MMDC astrophysics platform — programmatic access to multi-wavelength SED data and blazar broadband emission modeling.

MMDC provides APIs for querying astrophysical databases, preparing Spectral Energy Distribution (SED) data from multiple catalogs, and running physics simulations for blazar emission modeling using SSC, EIC, and hadronic models.

Installation

pip install astro-mmdc

Or with uv:

uv pip install astro-mmdc

Requires Python 3.10+.

---

Quick Start

Get SED data for a source in 4 lines:

from astro_mmdc import MMDC

client = MMDC()
job = client.sed.prepare_and_wait(ra=187.28, dec=2.05, database_name="3C273")
data = client.sed.get_data(job.uuid)

Run an SSC model fit:

result = client.modeling.batch_infer(
    "observations.csv",
    z=0.158,
    ebl=True,
    model_type="SSC",
)
print(result.pdf_link)
print(result.best_parameters)

End-to-end pipeline — from sky coordinates to model fit:

from astro_mmdc import MMDC

client = MMDC()

# Fetch SED data
job = client.sed.prepare_and_wait(
    ra=166.11, dec=38.21, database_name="Mkn421", source_name="Mkn 421"
)
info = client.sed.get_info(job.uuid)
client.sed.download_csv(job.uuid, "mkn421_sed.csv")

# Fit SSC model
result = client.modeling.batch_infer(
    "mkn421_sed.csv",
    z=info.redshift,
    ebl=True,
    model_type="SSC",
)

for name, param in result.best_parameters.items():
    print(f"{name}: {param.value:.4f} +/- {param.error:.4f}")

---

Guide

Creating a Client

from astro_mmdc import MMDC

client = MMDC()

# Custom request timeout (seconds)
client = MMDC(timeout=60.0)

# As a context manager (auto-closes HTTP connection)
with MMDC() as client:
    ...

The client provides two resource namespaces:

• client.sed — SED data preparation and retrieval
• client.modeling — blazar emission modeling (SSC, EIC, hadronic)

---

SED Data Preparation

SED preparation fetches multi-wavelength observational data from external catalogs for a given sky position.

job = client.sed.prepare_and_wait(
    ra=187.28,              # Right Ascension in degrees
    dec=2.05,               # Declination in degrees
    database_name="3C273",  # Database identifier
    source_name="3C 273",   # Optional display name
)
print(job.status)  # "done", "no_data", or "error"
print(job.uuid)    # Use this UUID for all subsequent data retrieval

If data already exists for a sky position, MMDC returns the cached result. Use force=True to re-fetch:

job = client.sed.prepare_and_wait(ra=187.28, dec=2.05, database_name="3C273", force=True)

Retrieving SED Data

Once a job is complete, retrieve the frequency/flux data:

data = client.sed.get_data(job.uuid)

# data.uuid, data.ra, data.dec, data.source_name
# data.data — nested dict of frequency ranges, catalogs, and flux measurements

Filter by time range, catalogs, or convert units:

data = client.sed.get_data(
    job.uuid,
    mjd_start=50000.0,                      # Filter by MJD time range
    mjd_end=60000.0,
    exclude_catalogs=["WISE", "2MASS"],      # Exclude specific catalogs
    exclude_freq_ranges=["radio"],           # Exclude frequency ranges
    x_axis="freq_ev",                        # Convert frequency to eV
    y_axis="flux_ev",                        # Convert flux to eV/cm²/s
)

Available axis conversions:

x_axis	Description
"freq_ev"	Hz to eV

y_axis	Description
"flux_ev"	erg/cm²/s to eV/cm²/s
"flux_norm"	Normalized eV/cm²/s
"flux_jyhz"	Jy·Hz
"flux_wm2"	W/m²
"nufnu_fnu_jy"	Fν in Jy

Source Metadata

info = client.sed.get_info(job.uuid)
print(info.source_name)   # "3C 273"
print(info.ra, info.dec)  # Coordinates
print(info.redshift)      # Redshift (float or None)
print(info.gal_lat)       # Galactic latitude
print(info.gal_long)      # Galactic longitude
print(info.W_peak)        # Peak frequency indicator

Download as CSV

client.sed.download_csv(job.uuid, "sed_data.csv")

# With filters (same options as get_data)
client.sed.download_csv(
    job.uuid, "filtered.csv",
    mjd_start=55000.0,
    exclude_catalogs=["WISE"],
)

CSV columns: frequency, flux, flux_err, MJD_start, MJD_end, flag, catalog, reference

---

Blazar Emission Modeling

MMDC supports three blazar broadband emission models:

Model	Description
SSC	Synchrotron Self-Compton
EIC	External Inverse Compton
HADRONIC	Hadronic emission model

Input Data Format

All modeling endpoints expect a CSV file with three columns (case-sensitive, lowercase):

frequency,flux,flux_err
1.00e+09,2.50e-14,3.00e-15
4.85e+09,3.10e-14,2.80e-15
...

Validate CSV Before Submitting

validation = client.modeling.validate_csv("observations.csv")

print(validation.success)          # True/False
print(validation.data_points)      # Number of valid rows
print(validation.columns)          # ["frequency", "flux", "flux_err"]
print(validation.frequency_range)  # [min, max]
print(validation.flux_range)       # [min, max]

SSC Model Fitting

result = client.modeling.batch_infer(
    "observations.csv",
    z=0.158,            # Redshift (0 < z <= 10)
    ebl=True,           # EBL absorption correction
    model_type="SSC",
)

print(result.pdf_link)                   # URL to PDF report
print(result.csv_best_parameters_link)   # URL to best-fit parameters CSV
print(result.csv_best_model_link)        # URL to best-fit model CSV
print(result.best_parameters)            # Dict of parameter name -> {value, error}

Fixed Parameters

Fix specific model parameters instead of fitting them:

result = client.modeling.batch_infer(
    "observations.csv",
    z=0.158,
    ebl=True,
    model_type="SSC",
    fixed_parameters={
        "log_B": -1.5,
        "lorentz_factor": 20.0,
    },
)

SSC parameters: log_B, log_electron_luminosity, log_gamma_cut, log_gamma_min, log_radius, lorentz_factor, spectral_index

EIC parameters: log_B, log_Ld, log_MBH, log_electron_luminosity, log_gamma_cut, log_gamma_min, log_radius, lorentz_factor, spectral_index, log_nu_BLR, log_nu_DT

HADRONIC parameters: log_B, log_Le, log_gamma_e_cut, log_gamma_e_min, log_gamma_p_cut, log_Lp, log_R, lorentz_factor, pe, pp

EIC Model Fitting

result = client.modeling.batch_infer(
    "observations.csv",
    z=0.5,
    ebl=True,
    model_type="EIC",
    fixed_parameters={"log_nu_BLR": 15.0, "log_nu_DT": 13.5},
)

HADRONIC Model with Neutrino Parameters

Hadronic models require additional neutrino likelihood parameters. Choose either Poisson or chi-square likelihood:

Poisson likelihood:

result = client.modeling.batch_infer(
    "observations.csv",
    z=1.0,
    ebl=True,
    model_type="HADRONIC",
    likelihood_type="poisson",
    n_icecube=3,       # Number of IceCube neutrino events
    dt=12.0,           # Observation period in months
)

Chi-square likelihood:

result = client.modeling.batch_infer(
    "observations.csv",
    z=1.0,
    ebl=True,
    model_type="HADRONIC",
    likelihood_type="chi2",
    x1=100.0,          # First neutrino energy (TeV)
    x2=200.0,          # Second neutrino energy (TeV)
    y=-12.0,           # Neutrino flux log value
)

Working with Results

result = client.modeling.batch_infer(...)

# Best-fit parameters
for name, param in result.best_parameters.items():
    print(f"{name}: {param.value} +/- {param.error}")

# Fixed parameters
if result.fixed_parameters:
    for name, param in result.fixed_parameters.items():
        print(f"{name} (fixed): {param.value}")

# Download links
print(result.pdf_link)                   # PDF report with plots
print(result.csv_best_parameters_link)   # Best-fit parameters as CSV
print(result.csv_best_model_link)        # Best-fit model curve as CSV

Download result files:

import httpx

if result.pdf_link:
    pdf = httpx.get(result.pdf_link)
    with open("report.pdf", "wb") as f:
        f.write(pdf.content)

Error Handling

from astro_mmdc import (
    MMDC,
    MMDCError,           # Base exception for all SDK errors
    APIError,            # Non-2xx HTTP response
    NotFoundError,       # 404 response
    ValidationError,     # 422 response (CSV/parameter validation)
    PollingTimeoutError, # Polling exceeded max wait time
)

client = MMDC()

try:
    result = client.modeling.batch_infer("data.csv", z=0.5, ebl=True, model_type="SSC")
except ValidationError as e:
    print(f"CSV validation failed: {e} (type: {e.validation_type})")
except PollingTimeoutError:
    print("Job did not complete in time")
except NotFoundError:
    print("Resource not found")
except APIError as e:
    print(f"HTTP {e.status_code}: {e.detail}")

The SDK automatically retries on transient errors (429, 502, 503, 504) with exponential backoff (up to 3 attempts).

---

Advanced

Manual Job Control

Both SED preparation and batch modeling are asynchronous — you submit a job, then poll for results. The convenience methods (prepare_and_wait, batch_infer) handle polling automatically, but you can manage each step yourself for more control.

This is useful when you want to submit multiple jobs at once and poll them independently, or do other work between submission and result retrieval.

SED: Manual Submit and Poll

# Submit — returns immediately
job = client.sed.prepare(ra=187.28, dec=2.05, database_name="3C273")
print(job.uuid)

# Check status manually
status = client.sed.get_status(job.uuid)
print(status.status)  # "processing", "done", "no_data", or "error"

# Or block until complete with custom polling settings
completed = client.sed.wait_for_completion(
    job.uuid,
    poll_interval=5.0,   # Seconds between checks
    max_minutes=15.0,    # Give up after this
)

Modeling: Manual Submit and Poll

# Submit — returns immediately with a batch_result_id
submission = client.modeling.submit_batch(
    "observations.csv",
    z=0.158,
    ebl=True,
    model_type="SSC",
)
print(submission.batch_result_id)

# Check result manually
result = client.modeling.get_batch_result(submission.batch_result_id)
if result.pdf_link:
    print("Job complete!")
else:
    print("Still processing...")

# Or block until complete with custom polling settings
result = client.modeling.wait_for_batch(
    submission.batch_result_id,
    poll_interval=10.0,
    max_minutes=15.0,
)

Batch Processing Multiple Sources

import time

sources = [
    {"ra": 187.28, "dec": 2.05, "name": "3C273"},
    {"ra": 166.11, "dec": 38.21, "name": "Mkn421"},
    {"ra": 253.47, "dec": 39.76, "name": "Mkn501"},
]

# Submit all jobs first
jobs = []
for src in sources:
    job = client.sed.prepare(ra=src["ra"], dec=src["dec"], database_name=src["name"])
    jobs.append(job)
    print(f"Submitted {src['name']}: {job.uuid}")

# Then wait for all of them
for job in jobs:
    completed = client.sed.wait_for_completion(job.uuid)
    print(f"{completed.source_name}: {completed.status}")

Synchronous Inference (SSC Only)

For quick SSC calculations without queuing:

# Convert CSV to the required JSON format first
json_data = client.modeling.csv_to_json("observations.csv")

# Run synchronous inference
result = client.modeling.infer(json_data["data"])

BatchResult Object Reference

All fields available on a BatchResult:

result.data                       # Model curve data points (dict)
result.best_parameters            # Best-fit parameters (dict of name -> {value, error})
result.fixed_parameters           # Fixed parameters (dict of name -> {value, error})
result.model_type                 # "SSC", "EIC", or "HADRONIC"
result.z                          # Redshift
result.pdf_link                   # URL to PDF report
result.csv_best_parameters_link   # URL to best-fit parameters CSV
result.csv_best_model_link        # URL to best-fit model curve CSV
result.uploaded_file              # Original input data (dict)
result.multinest_stats            # MultiNest sampling statistics (dict)
result.equal_weighted_posterior   # Posterior samples (dict)

---

API Reference

client.sed

Method	Description
prepare(ra, dec, database_name, ...)	Submit SED preparation job
get_status(uuid)	Check job status
wait_for_completion(uuid, ...)	Poll until job finishes
prepare_and_wait(ra, dec, database_name, ...)	Submit and wait
get_data(uuid, ...)	Get frequency/flux data
get_info(uuid)	Get source metadata
download_csv(uuid, dest, ...)	Download data as CSV file

client.modeling

Method	Description
validate_csv(file)	Validate CSV before submission
submit_batch(file, z, ebl, model_type, ...)	Submit batch inference job
get_batch_result(batch_result_id)	Get current job result
wait_for_batch(batch_result_id, ...)	Poll until job completes
batch_infer(file, z, ebl, model_type, ...)	Submit and wait
infer(data)	Synchronous SSC inference
csv_to_json(file)	Convert CSV to JSON format

---

Links

• MMDC Platform: mmdc.am
• Source Code: github.com/ICRANet/mmdc
• Issues: github.com/ICRANet/mmdc/issues