#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "CelestrakQueryType")]
#[derive(Clone)]
pub struct PyCelestrakQueryType {
pub(crate) value: celestrak::CelestrakQueryType,
}
#[pymethods]
impl PyCelestrakQueryType {
#[classattr]
#[allow(non_snake_case)]
fn GP() -> Self {
PyCelestrakQueryType {
value: celestrak::CelestrakQueryType::GP,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SUP_GP() -> Self {
PyCelestrakQueryType {
value: celestrak::CelestrakQueryType::SupGP,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SATCAT() -> Self {
PyCelestrakQueryType {
value: celestrak::CelestrakQueryType::SATCAT,
}
}
fn __str__(&self) -> String {
self.value.as_str().to_string()
}
fn __repr__(&self) -> String {
format!("CelestrakQueryType.{:?}", 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 = "CelestrakOutputFormat")]
#[derive(Clone)]
pub struct PyCelestrakOutputFormat {
pub(crate) value: celestrak::CelestrakOutputFormat,
}
#[pymethods]
impl PyCelestrakOutputFormat {
#[classattr]
#[allow(non_snake_case)]
fn TLE() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::Tle,
}
}
#[classattr]
#[allow(non_snake_case)]
fn TWO_LE() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::TwoLe,
}
}
#[classattr]
#[allow(non_snake_case)]
fn THREE_LE() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::ThreeLe,
}
}
#[classattr]
#[allow(non_snake_case)]
fn XML() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::Xml,
}
}
#[classattr]
#[allow(non_snake_case)]
fn KVN() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::Kvn,
}
}
#[classattr]
#[allow(non_snake_case)]
fn JSON() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::Json,
}
}
#[classattr]
#[allow(non_snake_case)]
fn JSON_PRETTY() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::JsonPretty,
}
}
#[classattr]
#[allow(non_snake_case)]
fn CSV() -> Self {
PyCelestrakOutputFormat {
value: celestrak::CelestrakOutputFormat::Csv,
}
}
fn __str__(&self) -> String {
self.value.as_str().to_string()
}
fn __repr__(&self) -> String {
format!("CelestrakOutputFormat.{:?}", 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 = "SupGPSource")]
#[derive(Clone)]
pub struct PySupGPSource {
pub(crate) value: celestrak::SupGPSource,
}
#[pymethods]
impl PySupGPSource {
#[classattr]
#[allow(non_snake_case)]
fn SPACEX() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::SpaceX,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SPACEX_SUP() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::SpaceXSup,
}
}
#[classattr]
#[allow(non_snake_case)]
fn PLANET() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Planet,
}
}
#[classattr]
#[allow(non_snake_case)]
fn ONEWEB() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::OneWeb,
}
}
#[classattr]
#[allow(non_snake_case)]
fn STARLINK() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Starlink,
}
}
#[classattr]
#[allow(non_snake_case)]
fn STARLINK_SUP() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::StarlinkSup,
}
}
#[classattr]
#[allow(non_snake_case)]
fn GEO() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Geo,
}
}
#[classattr]
#[allow(non_snake_case)]
fn GPS() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Gps,
}
}
#[classattr]
#[allow(non_snake_case)]
fn GLONASS() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Glonass,
}
}
#[classattr]
#[allow(non_snake_case)]
fn METEOSAT() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Meteosat,
}
}
#[classattr]
#[allow(non_snake_case)]
fn INTELSAT() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Intelsat,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SES() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Ses,
}
}
#[classattr]
#[allow(non_snake_case)]
fn IRIDIUM() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Iridium,
}
}
#[classattr]
#[allow(non_snake_case)]
fn IRIDIUM_NEXT() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::IridiumNext,
}
}
#[classattr]
#[allow(non_snake_case)]
fn ORBCOMM() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Orbcomm,
}
}
#[classattr]
#[allow(non_snake_case)]
fn GLOBALSTAR() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Globalstar,
}
}
#[classattr]
#[allow(non_snake_case)]
fn SWARM_TECHNOLOGIES() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::SwarmTechnologies,
}
}
#[classattr]
#[allow(non_snake_case)]
fn AMATEUR() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Amateur,
}
}
#[classattr]
#[allow(non_snake_case)]
fn CELESTRAK() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::CelesTrak,
}
}
#[classattr]
#[allow(non_snake_case)]
fn KUIPER() -> Self {
PySupGPSource {
value: celestrak::SupGPSource::Kuiper,
}
}
fn __str__(&self) -> String {
self.value.as_str().to_string()
}
fn __repr__(&self) -> String {
format!("SupGPSource.{:?}", 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 = "CelestrakQuery")]
#[derive(Clone)]
pub struct PyCelestrakQuery {
pub(crate) inner: celestrak::CelestrakQuery,
}
#[pymethods]
impl PyCelestrakQuery {
#[classattr]
fn gp() -> Self {
PyCelestrakQuery {
inner: celestrak::CelestrakQuery::gp(),
}
}
#[classattr]
fn sup_gp() -> Self {
PyCelestrakQuery {
inner: celestrak::CelestrakQuery::sup_gp(),
}
}
#[classattr]
fn satcat() -> Self {
PyCelestrakQuery {
inner: celestrak::CelestrakQuery::satcat(),
}
}
fn group(&self, name: &str) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().group(name),
}
}
fn catnr(&self, id: u32) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().catnr(id),
}
}
fn intdes(&self, intdes: &str) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().intdes(intdes),
}
}
fn name_search(&self, name: &str) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().name_search(name),
}
}
fn special(&self, special: &str) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().special(special),
}
}
fn source(&self, source: &PySupGPSource) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().source(source.value),
}
}
fn file(&self, file: &str) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().file(file),
}
}
fn payloads(&self, enabled: bool) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().payloads(enabled),
}
}
fn on_orbit(&self, enabled: bool) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().on_orbit(enabled),
}
}
fn active(&self, enabled: bool) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().active(enabled),
}
}
fn max(&self, count: u32) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().max(count),
}
}
#[pyo3(name = "format")]
fn set_format(&self, fmt: &PyCelestrakOutputFormat) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().format(fmt.value),
}
}
fn filter(&self, field: &str, value: &str) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().filter(field, value),
}
}
fn order_by(&self, field: &str, ascending: bool) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().order_by(field, ascending),
}
}
fn limit(&self, count: u32) -> Self {
PyCelestrakQuery {
inner: self.inner.clone().limit(count),
}
}
fn build_url(&self) -> String {
self.inner.build_url()
}
fn __str__(&self) -> String {
self.inner.build_url()
}
fn __repr__(&self) -> String {
format!(
"CelestrakQuery({:?}, \"{}\")",
self.inner.query_type(),
self.inner.build_url()
)
}
}
#[pyclass(module = "brahe._brahe", from_py_object)]
#[pyo3(name = "CelestrakSATCATRecord")]
#[derive(Clone)]
pub struct PyCelestrakSATCATRecord {
inner: celestrak::CelestrakSATCATRecord,
}
#[pymethods]
impl PyCelestrakSATCATRecord {
#[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 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 ops_status_code(&self) -> Option<String> { self.inner.ops_status_code.clone() }
#[getter] fn owner(&self) -> Option<String> { self.inner.owner.clone() }
#[getter] fn launch_date(&self) -> Option<String> { self.inner.launch_date.clone() }
#[getter] fn launch_site(&self) -> Option<String> { self.inner.launch_site.clone() }
#[getter] fn decay_date(&self) -> Option<String> { self.inner.decay_date.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 rcs(&self) -> Option<String> { self.inner.rcs.clone() }
#[getter] fn data_status_code(&self) -> Option<String> { self.inner.data_status_code.clone() }
#[getter] fn orbit_center(&self) -> Option<String> { self.inner.orbit_center.clone() }
#[getter] fn orbit_type(&self) -> Option<String> { self.inner.orbit_type.clone() }
fn __str__(&self) -> String {
format!(
"CelestrakSATCATRecord(name={:?}, norad_id={:?})",
self.inner.object_name, self.inner.norad_cat_id
)
}
fn __repr__(&self) -> String {
self.__str__()
}
}
#[pyclass(module = "brahe._brahe")]
#[pyo3(name = "CelestrakClient")]
pub struct PyCelestrakClient {
inner: celestrak::CelestrakClient,
}
#[pymethods]
impl PyCelestrakClient {
#[new]
#[pyo3(signature = (base_url=None, cache_max_age=None, max_retries=None))]
fn new(
base_url: Option<&str>,
cache_max_age: Option<f64>,
max_retries: Option<u32>,
) -> Self {
let mut client = match (base_url, cache_max_age) {
(Some(url), Some(age)) => {
celestrak::CelestrakClient::with_base_url_and_cache_age(url, age)
}
(Some(url), None) => celestrak::CelestrakClient::with_base_url(url),
(None, Some(age)) => celestrak::CelestrakClient::with_cache_age(age),
(None, None) => celestrak::CelestrakClient::new(),
};
if let Some(n) = max_retries {
client = client.max_retries(n);
}
PyCelestrakClient { inner: client }
}
#[pyo3(signature = (*, catnr=None, group=None, name=None, intdes=None))]
fn get_gp(
&self,
catnr: Option<u32>,
group: Option<&str>,
name: Option<&str>,
intdes: Option<&str>,
) -> PyResult<Vec<PyGPRecord>> {
let count = catnr.is_some() as u8
+ group.is_some() as u8
+ name.is_some() as u8
+ intdes.is_some() as u8;
if count != 1 {
return Err(exceptions::PyValueError::new_err(
"Provide exactly one of: catnr, group, name, intdes",
));
}
let records = if let Some(id) = catnr {
self.inner.get_gp_by_catnr(id)
} else if let Some(g) = group {
self.inner.get_gp_by_group(g)
} else if let Some(n) = name {
self.inner.get_gp_by_name(n)
} else {
self.inner.get_gp_by_intdes(intdes.unwrap())
}
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records
.into_iter()
.map(|r| PyGPRecord { inner: r })
.collect())
}
fn get_sup_gp(&self, source: &PySupGPSource) -> PyResult<Vec<PyGPRecord>> {
let records = self
.inner
.get_sup_gp(source.value)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records
.into_iter()
.map(|r| PyGPRecord { inner: r })
.collect())
}
#[pyo3(signature = (*, catnr=None, active=None, payloads=None, on_orbit=None))]
fn get_satcat(
&self,
catnr: Option<u32>,
active: Option<bool>,
payloads: Option<bool>,
on_orbit: Option<bool>,
) -> PyResult<Vec<PyCelestrakSATCATRecord>> {
if catnr.is_none() && active.is_none() && payloads.is_none() && on_orbit.is_none() {
return Err(exceptions::PyValueError::new_err(
"Provide at least one of: catnr, active, payloads, on_orbit",
));
}
let mut query = celestrak::CelestrakQuery::satcat();
if let Some(id) = catnr {
query = query.catnr(id);
}
if let Some(a) = active {
query = query.active(a);
}
if let Some(p) = payloads {
query = query.payloads(p);
}
if let Some(o) = on_orbit {
query = query.on_orbit(o);
}
let records = self
.inner
.query_satcat(&query)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(records
.into_iter()
.map(|r| PyCelestrakSATCATRecord { inner: r })
.collect())
}
#[pyo3(signature = (*, catnr, step_size=60.0))]
fn get_sgp_propagator(&self, catnr: u32, step_size: f64) -> PyResult<PySGPPropagator> {
let propagator = self
.inner
.get_sgp_propagator_by_catnr(catnr, step_size)
.map_err(|e| BraheError::new_err(e.to_string()))?;
Ok(PySGPPropagator { propagator })
}
fn query<'py>(
&self,
py: Python<'py>,
query: &PyCelestrakQuery,
) -> PyResult<Py<PyAny>> {
match query.inner.query_type() {
celestrak::CelestrakQueryType::GP | celestrak::CelestrakQueryType::SupGP => {
let records = self
.inner
.query_gp(&query.inner)
.map_err(|e| BraheError::new_err(e.to_string()))?;
let py_records: Vec<PyGPRecord> = records
.into_iter()
.map(|r| PyGPRecord { inner: r })
.collect();
py_records.into_py_any(py)
}
celestrak::CelestrakQueryType::SATCAT => {
let records = self
.inner
.query_satcat(&query.inner)
.map_err(|e| BraheError::new_err(e.to_string()))?;
let py_records: Vec<PyCelestrakSATCATRecord> = records
.into_iter()
.map(|r| PyCelestrakSATCATRecord { inner: r })
.collect();
py_records.into_py_any(py)
}
}
}
fn query_raw(&self, query: &PyCelestrakQuery) -> PyResult<String> {
self.inner
.query_raw(&query.inner)
.map_err(|e| BraheError::new_err(e.to_string()))
}
fn download(&self, query: &PyCelestrakQuery, filepath: &str) -> PyResult<()> {
self.inner
.download(&query.inner, Path::new(filepath))
.map_err(|e| BraheError::new_err(e.to_string()))
}
}