#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "RequestController")]
#[derive(Clone)]
pub struct PyRequestController {
pub(crate) value: spacetrack::RequestController,
}
#[pymethods]
impl PyRequestController {
#[classattr]
#[allow(non_snake_case)]
fn BASIC_SPACE_DATA() -> Self {
PyRequestController {
value: spacetrack::RequestController::BasicSpaceData,
}
}
#[classattr]
#[allow(non_snake_case)]
fn EXPANDED_SPACE_DATA() -> Self {
PyRequestController {
value: spacetrack::RequestController::ExpandedSpaceData,
}
}
#[classattr]
#[allow(non_snake_case)]
fn FILE_SHARE() -> Self {
PyRequestController {
value: spacetrack::RequestController::FileShare,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SP_EPHEMERIS() -> Self {
PyRequestController {
value: spacetrack::RequestController::SPEphemeris,
}
}
#[classattr]
#[allow(non_snake_case)]
fn PUBLIC_FILES() -> Self {
PyRequestController {
value: spacetrack::RequestController::PublicFiles,
}
}
fn __str__(&self) -> String {
format!("{:?}", self.value)
}
fn __repr__(&self) -> String {
format!("RequestController.{:?}", self.value)
}
fn __richcmp__(&self, other: &Self, op: CompareOp) -> PyResult<bool> {
match op {
CompareOp::Eq => Ok(self.value == other.value),
CompareOp::Ne => Ok(self.value != other.value),
_ => Err(exceptions::PyNotImplementedError::new_err(
"Comparison not supported",
)),
}
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "RequestClass")]
#[derive(Clone)]
pub struct PyRequestClass {
pub(crate) value: spacetrack::RequestClass,
}
#[pymethods]
impl PyRequestClass {
#[classattr]
#[allow(non_snake_case)]
fn GP() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::GP,
}
}
#[classattr]
#[allow(non_snake_case)]
fn GP_HISTORY() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::GPHistory,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SATCAT() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::SATCAT,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SATCAT_CHANGE() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::SATCATChange,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SATCAT_DEBUT() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::SATCATDebut,
}
}
#[classattr]
#[allow(non_snake_case)]
fn DECAY() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::Decay,
}
}
#[classattr]
#[allow(non_snake_case)]
fn TIP() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::TIP,
}
}
#[classattr]
#[allow(non_snake_case)]
fn CDM_PUBLIC() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::CDMPublic,
}
}
#[classattr]
#[allow(non_snake_case)]
fn BOXSCORE() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::Boxscore,
}
}
#[classattr]
#[allow(non_snake_case)]
fn ANNOUNCEMENT() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::Announcement,
}
}
#[classattr]
#[allow(non_snake_case)]
fn LAUNCH_SITE() -> Self {
PyRequestClass {
value: spacetrack::RequestClass::LaunchSite,
}
}
fn __str__(&self) -> String {
format!("{:?}", self.value)
}
fn __repr__(&self) -> String {
format!("RequestClass.{:?}", self.value)
}
fn __richcmp__(&self, other: &Self, op: CompareOp) -> PyResult<bool> {
match op {
CompareOp::Eq => Ok(self.value == other.value),
CompareOp::Ne => Ok(self.value != other.value),
_ => Err(exceptions::PyNotImplementedError::new_err(
"Comparison not supported",
)),
}
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "SortOrder")]
#[derive(Clone)]
pub struct PySortOrder {
pub(crate) value: spacetrack::SortOrder,
}
#[pymethods]
impl PySortOrder {
#[classattr]
#[allow(non_snake_case)]
fn ASC() -> Self {
PySortOrder {
value: spacetrack::SortOrder::Asc,
}
}
#[classattr]
#[allow(non_snake_case)]
fn DESC() -> Self {
PySortOrder {
value: spacetrack::SortOrder::Desc,
}
}
fn __str__(&self) -> String {
format!("{:?}", self.value)
}
fn __repr__(&self) -> String {
format!("SortOrder.{:?}", self.value)
}
fn __richcmp__(&self, other: &Self, op: CompareOp) -> PyResult<bool> {
match op {
CompareOp::Eq => Ok(self.value == other.value),
CompareOp::Ne => Ok(self.value != other.value),
_ => Err(exceptions::PyNotImplementedError::new_err(
"Comparison not supported",
)),
}
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "OutputFormat")]
#[derive(Clone)]
pub struct PyOutputFormat {
pub(crate) value: spacetrack::OutputFormat,
}
#[pymethods]
impl PyOutputFormat {
#[classattr]
#[allow(non_snake_case)]
fn JSON() -> Self {
PyOutputFormat {
value: spacetrack::OutputFormat::JSON,
}
}
#[classattr]
#[allow(non_snake_case)]
fn XML() -> Self {
PyOutputFormat {
value: spacetrack::OutputFormat::XML,
}
}
#[classattr]
#[allow(non_snake_case)]
fn HTML() -> Self {
PyOutputFormat {
value: spacetrack::OutputFormat::HTML,
}
}
#[classattr]
#[allow(non_snake_case)]
fn CSV() -> Self {
PyOutputFormat {
value: spacetrack::OutputFormat::CSV,
}
}
#[classattr]
#[allow(non_snake_case)]
fn TLE() -> Self {
PyOutputFormat {
value: spacetrack::OutputFormat::TLE,
}
}
#[classattr]
#[allow(non_snake_case)]
fn THREE_LE() -> Self {
PyOutputFormat {
value: spacetrack::OutputFormat::ThreeLe,
}
}
#[classattr]
#[allow(non_snake_case)]
fn KVN() -> Self {
PyOutputFormat {
value: spacetrack::OutputFormat::KVN,
}
}
fn __str__(&self) -> String {
match self.value {
spacetrack::OutputFormat::ThreeLe => "3LE".to_string(),
_ => format!("{:?}", self.value),
}
}
fn __repr__(&self) -> String {
match self.value {
spacetrack::OutputFormat::ThreeLe => "OutputFormat.3LE".to_string(),
_ => format!("OutputFormat.{:?}", self.value),
}
}
fn __richcmp__(&self, other: &Self, op: CompareOp) -> PyResult<bool> {
match op {
CompareOp::Eq => Ok(self.value == other.value),
CompareOp::Ne => Ok(self.value != other.value),
_ => Err(exceptions::PyNotImplementedError::new_err(
"Comparison not supported",
)),
}
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "SpaceTrackQuery")]
#[derive(Clone)]
pub struct PySpaceTrackQuery {
inner: spacetrack::SpaceTrackQuery,
}
#[pymethods]
impl PySpaceTrackQuery {
#[new]
fn new(request_class: &PyRequestClass) -> Self {
PySpaceTrackQuery {
inner: spacetrack::SpaceTrackQuery::new(request_class.value),
}
}
fn controller(&self, controller: &PyRequestController) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().controller(controller.value),
}
}
fn filter(&self, field: &str, value: &str) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().filter(field, value),
}
}
fn order_by(&self, field: &str, order: &PySortOrder) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().order_by(field, order.value),
}
}
fn limit(&self, count: u32) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().limit(count),
}
}
fn limit_offset(&self, count: u32, offset: u32) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().limit_offset(count, offset),
}
}
#[pyo3(name = "format")]
fn set_format(&self, fmt: &PyOutputFormat) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().format(fmt.value),
}
}
fn predicates_filter(&self, fields: Vec<String>) -> Self {
let field_refs: Vec<&str> = fields.iter().map(|s| s.as_str()).collect();
PySpaceTrackQuery {
inner: self.inner.clone().predicates_filter(&field_refs),
}
}
fn metadata(&self, enabled: bool) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().metadata(enabled),
}
}
fn distinct(&self, enabled: bool) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().distinct(enabled),
}
}
fn empty_result(&self, enabled: bool) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().empty_result(enabled),
}
}
fn favorites(&self, favorites: &str) -> Self {
PySpaceTrackQuery {
inner: self.inner.clone().favorites(favorites),
}
}
fn build(&self) -> String {
self.inner.build()
}
fn __str__(&self) -> String {
self.inner.build()
}
fn __repr__(&self) -> String {
format!("SpaceTrackQuery(\"{}\")", self.inner.build())
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "GPRecord")]
#[derive(Clone)]
#[allow(clippy::upper_case_acronyms)]
pub struct PyGPRecord {
pub(crate) inner: spacetrack::GPRecord,
}
#[pymethods]
impl PyGPRecord {
#[classmethod]
#[pyo3(signature = (json_str))]
fn from_json(_cls: &Bound<'_, PyType>, json_str: &str) -> PyResult<Self> {
let record: spacetrack::GPRecord = serde_json::from_str(json_str)
.map_err(|e| BraheError::new_err(format!("Failed to parse GPRecord JSON: {}", e)))?;
Ok(PyGPRecord { inner: record })
}
#[getter] fn ccsds_omm_vers(&self) -> Option<String> { self.inner.ccsds_omm_vers.clone() }
#[getter] fn comment(&self) -> Option<String> { self.inner.comment.clone() }
#[getter] fn creation_date(&self) -> Option<String> { self.inner.creation_date.clone() }
#[getter] fn originator(&self) -> Option<String> { self.inner.originator.clone() }
#[getter] fn object_name(&self) -> Option<String> { self.inner.object_name.clone() }
#[getter] fn object_id(&self) -> Option<String> { self.inner.object_id.clone() }
#[getter] fn center_name(&self) -> Option<String> { self.inner.center_name.clone() }
#[getter] fn ref_frame(&self) -> Option<String> { self.inner.ref_frame.clone() }
#[getter] fn time_system(&self) -> Option<String> { self.inner.time_system.clone() }
#[getter] fn mean_element_theory(&self) -> Option<String> { self.inner.mean_element_theory.clone() }
#[getter] fn epoch(&self) -> Option<String> { self.inner.epoch.clone() }
#[getter] fn mean_motion(&self) -> Option<f64> { self.inner.mean_motion }
#[getter] fn eccentricity(&self) -> Option<f64> { self.inner.eccentricity }
#[getter] fn inclination(&self) -> Option<f64> { self.inner.inclination }
#[getter] fn ra_of_asc_node(&self) -> Option<f64> { self.inner.ra_of_asc_node }
#[getter] fn arg_of_pericenter(&self) -> Option<f64> { self.inner.arg_of_pericenter }
#[getter] fn mean_anomaly(&self) -> Option<f64> { self.inner.mean_anomaly }
#[getter] fn ephemeris_type(&self) -> Option<u8> { self.inner.ephemeris_type }
#[getter] fn classification_type(&self) -> Option<String> { self.inner.classification_type.clone() }
#[getter] fn norad_cat_id(&self) -> Option<u32> { self.inner.norad_cat_id }
#[getter] fn element_set_no(&self) -> Option<u16> { self.inner.element_set_no }
#[getter] fn rev_at_epoch(&self) -> Option<u32> { self.inner.rev_at_epoch }
#[getter] fn bstar(&self) -> Option<f64> { self.inner.bstar }
#[getter] fn mean_motion_dot(&self) -> Option<f64> { self.inner.mean_motion_dot }
#[getter] fn mean_motion_ddot(&self) -> Option<f64> { self.inner.mean_motion_ddot }
#[getter] fn semimajor_axis(&self) -> Option<f64> { self.inner.semimajor_axis }
#[getter] fn period(&self) -> Option<f64> { self.inner.period }
#[getter] fn apoapsis(&self) -> Option<f64> { self.inner.apoapsis }
#[getter] fn periapsis(&self) -> Option<f64> { self.inner.periapsis }
#[getter] fn object_type(&self) -> Option<String> { self.inner.object_type.clone() }
#[getter] fn rcs_size(&self) -> Option<String> { self.inner.rcs_size.clone() }
#[getter] fn country_code(&self) -> Option<String> { self.inner.country_code.clone() }
#[getter] fn launch_date(&self) -> Option<String> { self.inner.launch_date.clone() }
#[getter] fn site(&self) -> Option<String> { self.inner.site.clone() }
#[getter] fn decay_date(&self) -> Option<String> { self.inner.decay_date.clone() }
#[getter] fn file(&self) -> Option<u64> { self.inner.file }
#[getter] fn gp_id(&self) -> Option<u32> { self.inner.gp_id }
#[getter] fn tle_line0(&self) -> Option<String> { self.inner.tle_line0.clone() }
#[getter] fn tle_line1(&self) -> Option<String> { self.inner.tle_line1.clone() }
#[getter] fn tle_line2(&self) -> Option<String> { self.inner.tle_line2.clone() }
#[pyo3(signature = (step_size=60.0))]
fn to_sgp_propagator(&self, step_size: f64) -> PyResult<PySGPPropagator> {
let propagator = propagators::SGPPropagator::from_gp_record(&self.inner, step_size)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(PySGPPropagator { propagator })
}
#[pyo3(signature = (force_config, propagation_config=None, params=None))]
fn to_numerical_orbit_propagator(
&self,
force_config: &PyForceModelConfig,
propagation_config: Option<&PyNumericalPropagationConfig>,
params: Option<PyReadonlyArray1<f64>>,
) -> PyResult<PyNumericalOrbitPropagator> {
let params_dvec = params.map(|p| {
let vec = p.to_vec().unwrap();
DVector::from_vec(vec)
});
let prop_config = propagation_config.map(|c| c.config.clone());
let propagator = propagators::DNumericalOrbitPropagator::from_gp_record(
&self.inner,
force_config.config.clone(),
prop_config,
params_dvec,
)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(PyNumericalOrbitPropagator { propagator })
}
fn to_omm(&self) -> PyResult<PyOMM> {
let omm = self
.inner
.to_omm()
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(PyOMM { inner: omm })
}
fn to_json(&self) -> PyResult<String> {
serde_json::to_string(&self.inner)
.map_err(|e| BraheError::new_err(format!("Failed to serialize GPRecord to JSON: {}", e)))
}
fn to_dict<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, pyo3::types::PyDict>> {
let dict = pyo3::types::PyDict::new(py);
dict.set_item("CCSDS_OMM_VERS", self.inner.ccsds_omm_vers.as_deref())?;
dict.set_item("COMMENT", self.inner.comment.as_deref())?;
dict.set_item("CREATION_DATE", self.inner.creation_date.as_deref())?;
dict.set_item("ORIGINATOR", self.inner.originator.as_deref())?;
dict.set_item("OBJECT_NAME", self.inner.object_name.as_deref())?;
dict.set_item("OBJECT_ID", self.inner.object_id.as_deref())?;
dict.set_item("CENTER_NAME", self.inner.center_name.as_deref())?;
dict.set_item("REF_FRAME", self.inner.ref_frame.as_deref())?;
dict.set_item("TIME_SYSTEM", self.inner.time_system.as_deref())?;
dict.set_item("MEAN_ELEMENT_THEORY", self.inner.mean_element_theory.as_deref())?;
dict.set_item("EPOCH", self.inner.epoch.as_deref())?;
dict.set_item("MEAN_MOTION", self.inner.mean_motion)?;
dict.set_item("ECCENTRICITY", self.inner.eccentricity)?;
dict.set_item("INCLINATION", self.inner.inclination)?;
dict.set_item("RA_OF_ASC_NODE", self.inner.ra_of_asc_node)?;
dict.set_item("ARG_OF_PERICENTER", self.inner.arg_of_pericenter)?;
dict.set_item("MEAN_ANOMALY", self.inner.mean_anomaly)?;
dict.set_item("BSTAR", self.inner.bstar)?;
dict.set_item("MEAN_MOTION_DOT", self.inner.mean_motion_dot)?;
dict.set_item("MEAN_MOTION_DDOT", self.inner.mean_motion_ddot)?;
dict.set_item("SEMIMAJOR_AXIS", self.inner.semimajor_axis)?;
dict.set_item("PERIOD", self.inner.period)?;
dict.set_item("APOAPSIS", self.inner.apoapsis)?;
dict.set_item("PERIAPSIS", self.inner.periapsis)?;
dict.set_item("EPHEMERIS_TYPE", self.inner.ephemeris_type)?;
dict.set_item("NORAD_CAT_ID", self.inner.norad_cat_id)?;
dict.set_item("ELEMENT_SET_NO", self.inner.element_set_no)?;
dict.set_item("REV_AT_EPOCH", self.inner.rev_at_epoch)?;
dict.set_item("FILE", self.inner.file)?;
dict.set_item("GP_ID", self.inner.gp_id)?;
dict.set_item("CLASSIFICATION_TYPE", self.inner.classification_type.as_deref())?;
dict.set_item("OBJECT_TYPE", self.inner.object_type.as_deref())?;
dict.set_item("RCS_SIZE", self.inner.rcs_size.as_deref())?;
dict.set_item("COUNTRY_CODE", self.inner.country_code.as_deref())?;
dict.set_item("LAUNCH_DATE", self.inner.launch_date.as_deref())?;
dict.set_item("SITE", self.inner.site.as_deref())?;
dict.set_item("DECAY_DATE", self.inner.decay_date.as_deref())?;
dict.set_item("TLE_LINE0", self.inner.tle_line0.as_deref())?;
dict.set_item("TLE_LINE1", self.inner.tle_line1.as_deref())?;
dict.set_item("TLE_LINE2", self.inner.tle_line2.as_deref())?;
Ok(dict)
}
fn __str__(&self) -> String {
format!(
"GPRecord(name={:?}, norad_id={:?}, epoch={:?})",
self.inner.object_name, self.inner.norad_cat_id, self.inner.epoch
)
}
fn __repr__(&self) -> String {
self.__str__()
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "SATCATRecord")]
#[derive(Clone)]
#[allow(clippy::upper_case_acronyms)]
pub struct PySATCATRecord {
inner: spacetrack::SATCATRecord,
}
#[pymethods]
impl PySATCATRecord {
#[getter] fn intldes(&self) -> Option<String> { self.inner.intldes.clone() }
#[getter] fn norad_cat_id(&self) -> Option<u32> { self.inner.norad_cat_id }
#[getter] fn object_type(&self) -> Option<String> { self.inner.object_type.clone() }
#[getter] fn satname(&self) -> Option<String> { self.inner.satname.clone() }
#[getter] fn country(&self) -> Option<String> { self.inner.country.clone() }
#[getter] fn launch(&self) -> Option<String> { self.inner.launch.clone() }
#[getter] fn site(&self) -> Option<String> { self.inner.site.clone() }
#[getter] fn decay(&self) -> Option<String> { self.inner.decay.clone() }
#[getter] fn period(&self) -> Option<String> { self.inner.period.clone() }
#[getter] fn inclination(&self) -> Option<String> { self.inner.inclination.clone() }
#[getter] fn apogee(&self) -> Option<String> { self.inner.apogee.clone() }
#[getter] fn perigee(&self) -> Option<String> { self.inner.perigee.clone() }
#[getter] fn comment(&self) -> Option<String> { self.inner.comment.clone() }
#[getter] fn commentcode(&self) -> Option<String> { self.inner.commentcode.clone() }
#[getter] fn rcsvalue(&self) -> Option<String> { self.inner.rcsvalue.clone() }
#[getter] fn rcs_size(&self) -> Option<String> { self.inner.rcs_size.clone() }
#[getter] fn file(&self) -> Option<String> { self.inner.file.clone() }
#[getter] fn launch_year(&self) -> Option<String> { self.inner.launch_year.clone() }
#[getter] fn launch_num(&self) -> Option<String> { self.inner.launch_num.clone() }
#[getter] fn launch_piece(&self) -> Option<String> { self.inner.launch_piece.clone() }
#[getter] fn current(&self) -> Option<String> { self.inner.current.clone() }
#[getter] fn object_name(&self) -> Option<String> { self.inner.object_name.clone() }
#[getter] fn object_id(&self) -> Option<String> { self.inner.object_id.clone() }
#[getter] fn object_number(&self) -> Option<String> { self.inner.object_number.clone() }
fn __str__(&self) -> String {
format!(
"SATCATRecord(name={:?}, norad_id={:?})",
self.inner.satname, self.inner.norad_cat_id
)
}
fn __repr__(&self) -> String {
self.__str__()
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "FileShareFileRecord")]
#[derive(Clone)]
pub struct PyFileShareFileRecord {
inner: spacetrack::FileShareFileRecord,
}
#[pymethods]
impl PyFileShareFileRecord {
#[getter] fn file_id(&self) -> Option<String> { self.inner.file_id.clone() }
#[getter] fn file_name(&self) -> Option<String> { self.inner.file_name.clone() }
#[getter] fn file_link(&self) -> Option<String> { self.inner.file_link.clone() }
#[getter] fn file_size(&self) -> Option<String> { self.inner.file_size.clone() }
#[getter] fn file_conttype(&self) -> Option<String> { self.inner.file_conttype.clone() }
#[getter] fn folder_id(&self) -> Option<String> { self.inner.folder_id.clone() }
#[getter] fn created(&self) -> Option<String> { self.inner.created.clone() }
fn __str__(&self) -> String {
format!(
"FileShareFileRecord(id={:?}, name={:?})",
self.inner.file_id, self.inner.file_name
)
}
fn __repr__(&self) -> String {
self.__str__()
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "FolderRecord")]
#[derive(Clone)]
pub struct PyFolderRecord {
inner: spacetrack::FolderRecord,
}
#[pymethods]
impl PyFolderRecord {
#[getter] fn folder_id(&self) -> Option<String> { self.inner.folder_id.clone() }
#[getter] fn folder_name(&self) -> Option<String> { self.inner.folder_name.clone() }
#[getter] fn parent_folder_id(&self) -> Option<String> { self.inner.parent_folder_id.clone() }
#[getter] fn created(&self) -> Option<String> { self.inner.created.clone() }
fn __str__(&self) -> String {
format!(
"FolderRecord(id={:?}, name={:?})",
self.inner.folder_id, self.inner.folder_name
)
}
fn __repr__(&self) -> String {
self.__str__()
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "SPEphemerisFileRecord")]
#[derive(Clone)]
pub struct PySPEphemerisFileRecord {
inner: spacetrack::SPEphemerisFileRecord,
}
#[pymethods]
impl PySPEphemerisFileRecord {
#[getter] fn file_id(&self) -> Option<String> { self.inner.file_id.clone() }
#[getter] fn norad_cat_id(&self) -> Option<u32> { self.inner.norad_cat_id }
#[getter] fn file_name(&self) -> Option<String> { self.inner.file_name.clone() }
#[getter] fn file_link(&self) -> Option<String> { self.inner.file_link.clone() }
#[getter] fn file_size(&self) -> Option<String> { self.inner.file_size.clone() }
#[getter] fn created(&self) -> Option<String> { self.inner.created.clone() }
#[getter] fn epoch_start(&self) -> Option<String> { self.inner.epoch_start.clone() }
#[getter] fn epoch_stop(&self) -> Option<String> { self.inner.epoch_stop.clone() }
fn __str__(&self) -> String {
format!(
"SPEphemerisFileRecord(id={:?}, norad_id={:?}, name={:?})",
self.inner.file_id, self.inner.norad_cat_id, self.inner.file_name
)
}
fn __repr__(&self) -> String {
self.__str__()
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "RateLimitConfig")]
#[derive(Clone)]
pub struct PyRateLimitConfig {
pub(crate) inner: spacetrack::RateLimitConfig,
}
#[pymethods]
impl PyRateLimitConfig {
#[new]
#[pyo3(signature = (max_per_minute=25, max_per_hour=250))]
fn new(max_per_minute: u32, max_per_hour: u32) -> Self {
PyRateLimitConfig {
inner: spacetrack::RateLimitConfig {
max_per_minute,
max_per_hour,
},
}
}
#[staticmethod]
fn disabled() -> Self {
PyRateLimitConfig {
inner: spacetrack::RateLimitConfig::disabled(),
}
}
#[getter]
fn max_per_minute(&self) -> u32 {
self.inner.max_per_minute
}
#[getter]
fn max_per_hour(&self) -> u32 {
self.inner.max_per_hour
}
fn __str__(&self) -> String {
format!(
"RateLimitConfig(max_per_minute={}, max_per_hour={})",
self.inner.max_per_minute, self.inner.max_per_hour
)
}
fn __repr__(&self) -> String {
self.__str__()
}
fn __richcmp__(&self, other: &Self, op: CompareOp) -> PyResult<bool> {
match op {
CompareOp::Eq => Ok(self.inner == other.inner),
CompareOp::Ne => Ok(self.inner != other.inner),
_ => Err(exceptions::PyNotImplementedError::new_err(
"Comparison not supported",
)),
}
}
}
#[pyclass(module = "brahe._brahe")]
#[pyo3(name = "SpaceTrackClient")]
pub struct PySpaceTrackClient {
inner: spacetrack::SpaceTrackClient,
}
#[pymethods]
impl PySpaceTrackClient {
#[new]
#[pyo3(signature = (identity, password, base_url=None, rate_limit=None))]
fn new(
identity: &str,
password: &str,
base_url: Option<&str>,
rate_limit: Option<&PyRateLimitConfig>,
) -> Self {
let config = rate_limit
.map(|rl| rl.inner.clone())
.unwrap_or_default();
let client = match base_url {
Some(url) => spacetrack::SpaceTrackClient::with_base_url_and_rate_limit(
identity, password, url, config,
),
None => spacetrack::SpaceTrackClient::with_rate_limit(identity, password, config),
};
PySpaceTrackClient { inner: client }
}
fn authenticate(&self) -> PyResult<()> {
self.inner.authenticate().map_err(|e| BraheError::new_err(e.to_string()))
}
fn query_raw(&self, query: &PySpaceTrackQuery) -> PyResult<String> {
self.inner
.query_raw(&query.inner)
.map_err(|e| BraheError::new_err(e.to_string()))
}
fn query_json(&self, py: Python<'_>, query: &PySpaceTrackQuery) -> PyResult<Py<PyAny>> {
let values = self
.inner
.query_json(&query.inner)
.map_err(|e| BraheError::new_err(e.to_string()))?;
let json_str = serde_json::to_string(&values)
.map_err(|e| BraheError::new_err(format!("JSON serialization failed: {}", e)))?;
let json_module = py.import("json")?;
json_module
.call_method1("loads", (json_str,))
.map(|obj| obj.into())
}
fn query_gp(&self, query: &PySpaceTrackQuery) -> PyResult<Vec<PyGPRecord>> {
let records = self
.inner
.query_gp(&query.inner)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records.into_iter().map(|r| PyGPRecord { inner: r }).collect())
}
fn query_satcat(&self, query: &PySpaceTrackQuery) -> PyResult<Vec<PySATCATRecord>> {
let records = self
.inner
.query_satcat(&query.inner)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records
.into_iter()
.map(|r| PySATCATRecord { inner: r })
.collect())
}
fn fileshare_upload(
&self,
folder_id: &str,
file_name: &str,
file_data: &[u8],
) -> PyResult<String> {
self.inner
.fileshare_upload(folder_id, file_name, file_data)
.map_err(|e| BraheError::new_err(e.to_string()))
}
fn fileshare_download<'py>(
&self,
py: Python<'py>,
file_id: &str,
) -> PyResult<Bound<'py, pyo3::types::PyBytes>> {
let data = self
.inner
.fileshare_download(file_id)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(pyo3::types::PyBytes::new(py, &data))
}
fn fileshare_download_folder<'py>(
&self,
py: Python<'py>,
folder_id: &str,
) -> PyResult<Bound<'py, pyo3::types::PyBytes>> {
let data = self
.inner
.fileshare_download_folder(folder_id)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(pyo3::types::PyBytes::new(py, &data))
}
fn fileshare_list_files(&self) -> PyResult<Vec<PyFileShareFileRecord>> {
let records = self
.inner
.fileshare_list_files()
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records
.into_iter()
.map(|r| PyFileShareFileRecord { inner: r })
.collect())
}
fn fileshare_list_folders(&self) -> PyResult<Vec<PyFolderRecord>> {
let records = self
.inner
.fileshare_list_folders()
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records
.into_iter()
.map(|r| PyFolderRecord { inner: r })
.collect())
}
fn fileshare_delete(&self, file_id: &str) -> PyResult<String> {
self.inner
.fileshare_delete(file_id)
.map_err(|e| BraheError::new_err(e.to_string()))
}
fn spephemeris_download<'py>(
&self,
py: Python<'py>,
file_id: &str,
) -> PyResult<Bound<'py, pyo3::types::PyBytes>> {
let data = self
.inner
.spephemeris_download(file_id)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(pyo3::types::PyBytes::new(py, &data))
}
fn spephemeris_list_files(&self) -> PyResult<Vec<PySPEphemerisFileRecord>> {
let records = self
.inner
.spephemeris_list_files()
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records
.into_iter()
.map(|r| PySPEphemerisFileRecord { inner: r })
.collect())
}
fn spephemeris_file_history(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
let values = self
.inner
.spephemeris_file_history()
.map_err(|e| BraheError::new_err(e.to_string()))?;
let json_str = serde_json::to_string(&values)
.map_err(|e| BraheError::new_err(format!("JSON serialization failed: {}", e)))?;
let json_module = py.import("json")?;
json_module
.call_method1("loads", (json_str,))
.map(|obj| obj.into())
}
fn publicfiles_download<'py>(
&self,
py: Python<'py>,
file_name: &str,
) -> PyResult<Bound<'py, pyo3::types::PyBytes>> {
let data = self
.inner
.publicfiles_download(file_name)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(pyo3::types::PyBytes::new(py, &data))
}
fn publicfiles_list_dirs(&self, py: Python<'_>) -> PyResult<Py<PyAny>> {
let values = self
.inner
.publicfiles_list_dirs()
.map_err(|e| BraheError::new_err(e.to_string()))?;
let json_str = serde_json::to_string(&values)
.map_err(|e| BraheError::new_err(format!("JSON serialization failed: {}", e)))?;
let json_module = py.import("json")?;
json_module
.call_method1("loads", (json_str,))
.map(|obj| obj.into())
}
}
#[pyfunction]
#[pyo3(name = "spacetrack_greater_than")]
fn py_spacetrack_greater_than(value: &str) -> String {
spacetrack::operators::greater_than(value)
}
#[pyfunction]
#[pyo3(name = "spacetrack_less_than")]
fn py_spacetrack_less_than(value: &str) -> String {
spacetrack::operators::less_than(value)
}
#[pyfunction]
#[pyo3(name = "spacetrack_not_equal")]
fn py_spacetrack_not_equal(value: &str) -> String {
spacetrack::operators::not_equal(value)
}
#[pyfunction]
#[pyo3(name = "spacetrack_inclusive_range")]
fn py_spacetrack_inclusive_range(left: &str, right: &str) -> String {
spacetrack::operators::inclusive_range(left, right)
}
#[pyfunction]
#[pyo3(name = "spacetrack_like")]
fn py_spacetrack_like(value: &str) -> String {
spacetrack::operators::like(value)
}
#[pyfunction]
#[pyo3(name = "spacetrack_startswith")]
fn py_spacetrack_startswith(value: &str) -> String {
spacetrack::operators::startswith(value)
}
#[pyfunction]
#[pyo3(name = "spacetrack_now")]
fn py_spacetrack_now() -> String {
spacetrack::operators::now()
}
#[pyfunction]
#[pyo3(name = "spacetrack_now_offset")]
fn py_spacetrack_now_offset(days: i32) -> String {
spacetrack::operators::now_offset(days)
}
#[pyfunction]
#[pyo3(name = "spacetrack_null_val")]
fn py_spacetrack_null_val() -> String {
spacetrack::operators::null_val()
}
#[pyfunction]
#[pyo3(name = "spacetrack_or_list")]
fn py_spacetrack_or_list(values: Vec<String>) -> String {
let refs: Vec<&str> = values.iter().map(|s| s.as_str()).collect();
spacetrack::operators::or_list(&refs)
}