from pathlib import Path
import math
import pandas as pd
import yaml
import logging
from typing import Optional, Union
from dataclasses import dataclass
from GUIBRUSHR.General_Constants.FunctionsAndConstants.Constant_Variables import ConstantVariables
from GUIBRUSHR.General_Constants.FunctionsAndConstants.general_functions import get_csv_value, compute_transit_durations
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Constants
RADIATION_MODES = ConstantVariables.LIST_RAD_MODE
DEFAULT_RADIATION_MODE = "Transmission"
[docs]
@dataclass
class TargetParameters:
"""Data class to store target parameters with default values."""
[docs]
class Target:
"""
A class representing a target object with various physical and orbital parameters.
This class handles loading and managing target parameters from either CSV or YAML files.
It supports both transmission and emission radiation modes.
Attributes:
radiation_mode (str): The radiation mode of the target ("Transmission" or "Emission")
"""
[docs]
def __init__(
self,
path_target_file: Optional[Union[str, Path]] = None,
rad_mode: str = DEFAULT_RADIATION_MODE
) -> None:
"""
Initialize a Target object.
Args:
path_target_file: Path to the target parameter file (CSV or YAML)
rad_mode: Radiation mode ("Transmission" or "Emission")
Raises:
ValueError: If radiation mode is invalid
FileNotFoundError: If target file doesn't exist
ValueError: If file format is invalid or required parameters are missing
"""
if rad_mode not in RADIATION_MODES:
raise ValueError(f"Invalid radiation mode. Must be one of {RADIATION_MODES}")
self.name: Optional[str] = None
self.radius: Optional[float] = None
self.mass: Optional[float] = None
self.gravity: Optional[float] = None
self.equilibrium_temperature: Optional[float] = None
self.stellar_radius: Optional[float] = None
self.stellar_mass: Optional[float] = None
self.pressure_at_layer_0: Optional[float] = None
self.hjd0: Optional[float] = None
self.period: Optional[float] = None
self.systemic_velocity: Optional[float] = None
self.limit_phase_t14: Optional[float] = None
self.limit_phase_t12: Optional[float] = None
self.limit_phase_t23: Optional[float] = None
self.t14_hours: Optional[float] = None
self.t12_hours: Optional[float] = None
self.t23_hours: Optional[float] = None
self.kp: Optional[float] = None
self.kp_ecc: Optional[float] = None
self.ks: Optional[float] = None
self.eccentricity: Optional[float] = None
self.argument_of_periastron: Optional[float] = None
self.stellar_effective_temperature: Optional[float] = None
self.ra: Optional[float] = None
self.dec: Optional[float] = None
self.a_Rs_ratio: Optional[float] = None
self.projected_obliquity: Optional[float] = None
self.inclination: Optional[float] = None
self.v_sini: Optional[float] = None
self.radiation_mode = rad_mode
if path_target_file is not None:
self._load_parameters(path_target_file)
self._set_limit_phase_emission()
self._compute_kp_ecc_if_missing()
def _load_parameters(self, path_target_file: Union[str, Path]) -> None:
"""
Load parameters from either CSV or YAML file.
Path resolution mirrors the ``read_general_info`` auto-detection
introduced by commit d5c01f8: callers may pass either the legacy
``df_general_info.csv`` filename or its YAML sibling, and the loader
accepts whichever actually exists. This keeps GUI plot helpers (which
still hardcode ``.csv``) working against modern YAML-only retrievals.
Args:
path_target_file: Path to the parameter file (or to its sibling
in the alternate format).
Raises:
FileNotFoundError: If neither the requested file nor a sibling
in the alternate format exists.
ValueError: If the file format is invalid or required parameters
are missing.
"""
path = Path(path_target_file)
if not path.exists():
sibling = None
suffix = path.suffix.lower()
if suffix == '.csv':
sibling = path.with_suffix('.yaml')
if not sibling.exists():
sibling = path.with_suffix('.yml')
if not sibling.exists():
sibling = None
elif suffix in ('.yaml', '.yml'):
sibling = path.with_suffix('.csv')
if not sibling.exists():
sibling = None
if sibling is None:
raise FileNotFoundError(
f"Target file not found: {path_target_file} "
"(no CSV/YAML sibling found either)"
)
path = sibling
try:
if path.suffix.lower() == '.csv':
self._load_from_csv(path)
elif path.suffix.lower() in ['.yaml', '.yml']:
self._load_from_yaml(path)
else:
raise ValueError(f"Unsupported file format: {path.suffix}")
except Exception as e:
logger.error(f"Error loading target parameters: {str(e)}")
raise
def _populate_from_general_info(self, yaml_data: dict) -> None:
"""Populate target fields from a df_general_info.yaml dict.
The dict uses ``target_*``-prefixed keys (mirroring the legacy
``df_general_info.csv`` columns). Missing optional fields fall
back to the same defaults as ``_load_from_csv``.
"""
def _opt(key, default=None, cast=float):
v = yaml_data.get(key)
if v is None:
return default
return cast(v) if cast is not None else v
self.name = yaml_data["target_name"]
self.radius = float(yaml_data["target_radius"])
self.mass = float(yaml_data["target_mass"])
self.gravity = float(yaml_data["target_gravity"])
self.equilibrium_temperature = float(yaml_data["target_t_eq"])
self.stellar_radius = float(yaml_data["target_stellar_radius"])
self.stellar_mass = float(yaml_data["target_stellar_mass"])
self.pressure_at_layer_0 = float(yaml_data["target_p0"])
self.hjd0 = float(yaml_data["target_hjd0"])
self.period = float(yaml_data["target_period"])
self.systemic_velocity = float(yaml_data["target_vsys"])
if yaml_data.get("target_limph_t23") is not None:
self.limit_phase_t14 = float(yaml_data["target_limph_t14"])
self.limit_phase_t12 = float(yaml_data["target_limph_t12"])
self.limit_phase_t23 = float(yaml_data["target_limph_t23"])
self.t14_hours = float(yaml_data["target_t14"])
self.t12_hours = float(yaml_data["target_t12"])
self.t23_hours = float(yaml_data["target_t23"])
else:
old_limph = float(yaml_data["target_limph"])
self.limit_phase_t14 = old_limph
self.limit_phase_t12 = old_limph
self.limit_phase_t23 = old_limph
self.t14_hours = 0.0
self.t12_hours = 0.0
self.t23_hours = 0.0
self.kp = float(yaml_data["target_kp"])
self.kp_ecc = _opt("target_kp_ecc", default=None)
self.ks = _opt("target_ks", default=0)
self.eccentricity = float(yaml_data["target_ecc"])
self.argument_of_periastron = float(yaml_data["target_opi"])
self.stellar_effective_temperature = float(yaml_data["target_stellar_teff"])
self.ra = _opt("target_ra", default=0)
self.dec = _opt("target_dec", default=0)
self.a_Rs_ratio = _opt("target_a_Rs_ratio", default=0)
self.projected_obliquity = _opt("target_projected_obliquity", default=0)
self.inclination = _opt("target_inclination", default=90)
self.v_sini = _opt("target_v_sini", default=0)
def _load_from_csv(self, path: Path) -> None:
"""Load parameters from CSV file."""
try:
df = pd.read_csv(path)
self.name = get_csv_value(df, "target_name")
self.radius = float(get_csv_value(df, "target_radius"))
self.mass = float(get_csv_value(df, "target_mass"))
self.gravity = float(get_csv_value(df, "target_gravity"))
self.equilibrium_temperature = float(get_csv_value(df, "target_t_eq"))
self.stellar_radius = float(get_csv_value(df, "target_stellar_radius"))
self.stellar_mass = float(get_csv_value(df, "target_stellar_mass"))
self.pressure_at_layer_0 = float(get_csv_value(df, "target_p0"))
self.hjd0 = float(get_csv_value(df, "target_hjd0"))
self.period = float(get_csv_value(df, "target_period"))
self.systemic_velocity = float(get_csv_value(df, "target_vsys"))
limph_t23 = get_csv_value(df, "target_limph_t23")
if limph_t23 is not None:
self.limit_phase_t14 = float(get_csv_value(df, "target_limph_t14"))
self.limit_phase_t12 = float(get_csv_value(df, "target_limph_t12"))
self.limit_phase_t23 = float(get_csv_value(df, "target_limph_t23"))
self.t14_hours = float(get_csv_value(df, "target_t14"))
self.t12_hours = float(get_csv_value(df, "target_t12"))
self.t23_hours = float(get_csv_value(df, "target_t23"))
else:
old_limph = float(get_csv_value(df, "target_limph"))
self.limit_phase_t14 = old_limph
self.limit_phase_t12 = old_limph
self.limit_phase_t23 = old_limph
self.t14_hours = 0.0
self.t12_hours = 0.0
self.t23_hours = 0.0
self.kp = float(get_csv_value(df, "target_kp"))
kp_ecc = get_csv_value(df, "target_kp_ecc")
self.kp_ecc = None if kp_ecc is None else float(kp_ecc)
ks = get_csv_value(df, "target_ks")
self.ks = 0 if ks is None else float(ks)
self.eccentricity = float(get_csv_value(df, "target_ecc"))
self.argument_of_periastron = float(get_csv_value(df, "target_opi"))
self.stellar_effective_temperature = float(get_csv_value(df, "target_stellar_teff"))
ra = get_csv_value(df, "target_ra")
self.ra = 0 if ra is None else float(ra)
dec = get_csv_value(df, "target_dec")
self.dec = 0 if dec is None else float(dec)
a_Rs_ratio = get_csv_value(df, "target_a_Rs_ratio")
self.a_Rs_ratio = 0 if a_Rs_ratio is None else float(a_Rs_ratio)
projected_obliquity = get_csv_value(df, "target_projected_obliquity")
self.projected_obliquity = 0 if projected_obliquity is None else float(projected_obliquity)
inclination = get_csv_value(df, "target_inclination")
self.inclination = 90 if inclination is None else float(inclination)
v_sini = get_csv_value(df, "target_v_sini")
self.v_sini = 0 if v_sini is None else float(v_sini)
except Exception as e:
raise ValueError(f"Error parsing CSV file: {str(e)}")
def _load_from_yaml(self, path: Path) -> None:
"""Load parameters from YAML file.
Two YAML schemas are supported and auto-detected by the keys
present at the top level:
1. **Standalone target YAML** (``info.yaml``-style, used by tests
and for hand-authored target definitions): bare keys such as
``name``, ``radius_jup``, ``stellar_teff_K``.
2. **df_general_info.yaml** (written by ``retrieval_io.write_general_info``
since commit d5c01f8): same fields prefixed with ``target_`` and
with the column-name spelling used by the legacy CSV layout
(``target_name``, ``target_radius``, ``target_stellar_teff``).
The retrieval pipeline passes the latter to ``Target(...)``; the
GUI test harness writes the former.
"""
try:
with open(path, 'r') as f:
yaml_data = yaml.safe_load(f)
if "target_name" in yaml_data:
self._populate_from_general_info(yaml_data)
return
self.name = yaml_data["name"]
self.radius = float(yaml_data["radius_jup"])
self.mass = float(yaml_data["mass_jup"])
self.gravity = float(yaml_data["gravity_cm_s2"])
self.equilibrium_temperature = float(yaml_data["t_eq_K"])
self.stellar_radius = float(yaml_data["stellar_radius_sun"])
self.stellar_mass = float(yaml_data["stellar_mass_sun"])
self.pressure_at_layer_0 = float(yaml_data["p0_log10_bar"])
self.hjd0 = float(yaml_data["hjd0_days"])
self.period = float(yaml_data["period_days"])
self.systemic_velocity = float(yaml_data["v_system_km_s"])
self.kp = float(yaml_data["kp_km_s"])
kp_ecc_present = "kp_ecc_km_s" in yaml_data.keys()
self.kp_ecc = float(yaml_data["kp_ecc_km_s"]) if kp_ecc_present else None
self.ks = float(yaml_data["ks_km_s"]) if "ks_km_s" in yaml_data.keys() else 0
self.eccentricity = float(yaml_data["ecc"])
self.argument_of_periastron = float(yaml_data["periastron_argument"])
self.stellar_effective_temperature = float(yaml_data["stellar_teff_K"])
self.ra = float(yaml_data["ra"]) if "ra" in yaml_data.keys() else 0
self.dec = float(yaml_data["dec"]) if "dec" in yaml_data.keys() else 0
self.a_Rs_ratio = float(yaml_data["a_Rs_ratio"]) if "a_Rs_ratio" in yaml_data.keys() else 0
self.projected_obliquity = float(yaml_data["projected_obliquity_deg"]) if "projected_obliquity_deg" in yaml_data.keys() else 0
self.inclination = float(yaml_data["inclination_deg"]) if "inclination_deg" in yaml_data.keys() else 90
self.v_sini = float(yaml_data["v_sini_km_s"]) if "v_sini_km_s" in yaml_data.keys() else 0
if "Limit_Phase_T23" in yaml_data:
self.limit_phase_t14 = float(yaml_data["Limit_Phase_T14"])
self.limit_phase_t12 = float(yaml_data["Limit_Phase_T12"])
self.limit_phase_t23 = float(yaml_data["Limit_Phase_T23"])
self.t14_hours = float(yaml_data["t14_hours"])
self.t12_hours = float(yaml_data["t12_hours"])
self.t23_hours = float(yaml_data["t23_hours"])
if not kp_ecc_present:
self._compute_kp_ecc_if_missing()
self._rewrite_yaml(path)
logger.info(f"Added missing kp_ecc_km_s to YAML for {self.name}: "
f"kp_ecc_km_s={self.kp_ecc}")
else:
old_t14 = float(yaml_data["t14_hours"]) if "t14_hours" in yaml_data else 0.0
phase_dict = compute_transit_durations(
self.period, self.a_Rs_ratio,
(self.radius * ConstantVariables.R_JUP_MEAN) / (self.stellar_radius * ConstantVariables.R_SUN),
float(self.inclination), ecc=float(self.eccentricity),
omega_planet=float(self.argument_of_periastron)
)
self.limit_phase_t14 = phase_dict["limph_T14"]
self.limit_phase_t12 = phase_dict["limph_T12"]
self.limit_phase_t23 = phase_dict["limph_T23"]
self.t14_hours = phase_dict["T14_hours"]
self.t12_hours = phase_dict["T12_hours"]
self.t23_hours = phase_dict["T23_hours"]
self._compute_kp_ecc_if_missing()
self._rewrite_yaml(path)
logger.info(f"Migrated old YAML format for {self.name}: "
f"limph_T14={self.limit_phase_t14}, limph_T12={self.limit_phase_t12}, limph_T23={self.limit_phase_t23}, "
f"t14={self.t14_hours}h, t12={self.t12_hours}h, t23={self.t23_hours}h")
except Exception as e:
raise ValueError(f"Error parsing YAML file: {str(e)}")
def _compute_kp_ecc_if_missing(self) -> None:
"""Compute and store the eccentric semi-amplitude when absent from the target file.
``kp_ecc`` is the planetary radial-velocity semi-amplitude corrected for
eccentricity, ``kp_ecc = kp / sqrt(1 - e**2)`` (it coincides with ``kp`` for a
circular orbit, ``e == 0``). It is used in place of ``kp`` only when the
eccentricity option is enabled (see ``rv_planet_and_star``). When a loaded target
file does not provide it, the correct value is computed here and then persisted to
the file, so it is stored once and simply read back on later loads.
"""
if self.kp_ecc is not None or self.kp is None:
return
ecc = self.eccentricity if self.eccentricity is not None else 0.0
if 0.0 < ecc < 1.0:
self.kp_ecc = self.kp / math.sqrt(1.0 - ecc ** 2)
else:
self.kp_ecc = self.kp
def _set_limit_phase_emission(self) -> None:
"""Set the limit phase based on radiation mode."""
if self.radiation_mode == "Emission":
self.limit_phase_t14 = 2.0
self.limit_phase_t12 = 2.0
self.limit_phase_t23 = 2.0
logger.info("Set limit phase to 2.0 for emission mode")
def _rewrite_yaml(self, path: Path) -> None:
"""Rewrite the YAML file with the new format including T14/T12 parameters."""
yaml_dict = {
'name': self.name,
'radius_jup': self.radius,
'mass_jup': self.mass,
'gravity_cm_s2': self.gravity,
't_eq_K': self.equilibrium_temperature,
'stellar_radius_sun': self.stellar_radius,
'stellar_mass_sun': self.stellar_mass,
'stellar_teff_K': self.stellar_effective_temperature,
'p0_log10_bar': self.pressure_at_layer_0,
'hjd0_days': self.hjd0,
'period_days': self.period,
't14_hours': float(self.t14_hours),
't12_hours': float(self.t12_hours),
't23_hours': float(self.t23_hours),
'Limit_Phase_T14': float(self.limit_phase_t14),
'Limit_Phase_T12': float(self.limit_phase_t12),
'Limit_Phase_T23': float(self.limit_phase_t23),
'kp_km_s': self.kp,
'kp_ecc_km_s': self.kp_ecc,
'ks_km_s': self.ks,
'v_system_km_s': self.systemic_velocity,
'ecc': self.eccentricity,
'periastron_argument': self.argument_of_periastron,
'ra': self.ra,
'dec': self.dec,
'a_Rs_ratio': self.a_Rs_ratio,
'projected_obliquity_deg': self.projected_obliquity,
'inclination_deg': self.inclination,
'v_sini_km_s': self.v_sini,
}
with open(path, 'w') as f:
yaml.dump(yaml_dict, f, sort_keys=False)
def __str__(self) -> str:
"""Return a string representation of the target."""
return f"Target(name={self.name}, mode={self.radiation_mode})"