nyx_space/od/msr/trackingdata/
python.rs

1/*
2    Nyx, blazing fast astrodynamics
3    Copyright (C) 2018-onwards Christopher Rabotin <christopher.rabotin@gmail.com>
4
5    This program is free software: you can redistribute it and/or modify
6    it under the terms of the GNU Affero General Public License as published
7    by the Free Software Foundation, either version 3 of the License, or
8    (at your option) any later version.
9
10    This program is distributed in the hope that it will be useful,
11    but WITHOUT ANY WARRANTY; without even the implied warranty of
12    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13    GNU Affero General Public License for more details.
14
15    You should have received a copy of the GNU Affero General Public License
16    along with this program.  If not, see <https://www.gnu.org/licenses/>.
17*/
18
19use super::{Measurement, MeasurementType, TrackingDataArc};
20use crate::io::{ExportCfg, InputOutputError};
21use hifitime::{Duration, Epoch};
22use pyo3::prelude::*;
23use pyo3::types::PyType;
24use std::collections::HashMap;
25use std::ops::Bound::{Excluded, Included, Unbounded};
26
27#[pymethods]
28impl TrackingDataArc {
29    #[new]
30    fn py_new(measurements: Vec<Measurement>) -> Self {
31        let mut trk_data = Self {
32            measurements,
33            source: None,
34            moduli: None,
35            force_reject: false,
36        };
37
38        trk_data.sort();
39
40        trk_data
41    }
42
43    /// Initializes a new Almanac from a file path to CCSDS OEM file, after converting to to SPICE SPK/BSP
44    ///
45    /// :type path: str
46    /// :type aliases: dict
47    /// :rtype: nyx_space.od.TrackingDataArc
48    #[classmethod]
49    #[pyo3(name = "from_ccsds_tdm")]
50    fn py_from_ccsds_tdm_file(
51        _cls: Bound<'_, PyType>,
52        path: &str,
53        aliases: Option<HashMap<String, String>>,
54    ) -> Result<Self, InputOutputError> {
55        TrackingDataArc::from_tdm(path, aliases)
56    }
57
58    /// Load TrackingDataArc from a parquet file.
59    ///
60    /// :type path: str
61    /// :rtype: TrackingDataArc
62    #[classmethod]
63    #[pyo3(name = "from_parquet")]
64    fn py_from_parquet(_cls: Bound<'_, PyType>, path: &str) -> Result<Self, InputOutputError> {
65        Self::from_parquet(path)
66    }
67
68    /// Write tracking data in CCSDS TDM format.
69    ///
70    /// :type spacecraft_name: str
71    /// :type aliases: dict | None
72    /// :type path: str
73    /// :rtype: str
74    #[pyo3(name = "write_ccsds_tdm")]
75    fn py_write_ccsds_tdm(
76        &self,
77        spacecraft_name: String,
78        aliases: Option<HashMap<String, String>>,
79        path: &str,
80    ) -> Result<String, InputOutputError> {
81        Ok(self
82            .clone()
83            .to_tdm_file(spacecraft_name, aliases, path, ExportCfg::default())?
84            .to_str()
85            .unwrap_or("woah_bug_building_path")
86            .to_string())
87    }
88
89    /// :rtype: list[str]
90    #[pyo3(name = "unique_aliases")]
91    fn py_unique_aliases(&self) -> Vec<String> {
92        self.unique_aliases().iter().cloned().collect()
93    }
94    /// :rtype: list[MeasurementType]
95    #[pyo3(name = "unique_types")]
96    fn py_unique_types(&self) -> Vec<MeasurementType> {
97        self.unique_types().iter().cloned().collect()
98    }
99
100    fn __str__(&self) -> String {
101        format!("{self}")
102    }
103
104    fn __repr__(&self) -> String {
105        format!("{self} @ {self:p}")
106    }
107
108    #[getter]
109    fn get_force_reject(&self) -> bool {
110        self.force_reject
111    }
112
113    #[setter]
114    fn set_force_reject(&mut self, reject: bool) {
115        self.force_reject = reject;
116    }
117
118    /// Filter measurements by epoch range.
119    ///
120    /// :type start: Epoch | None
121    /// :type end: Epoch | None
122    /// :rtype: TrackingDataArc
123    #[pyo3(name = "filter_by_epoch")]
124    fn py_filter_by_epoch(&self, start: Option<Epoch>, end: Option<Epoch>) -> Self {
125        let start_bound = start.map(Included).unwrap_or(Unbounded);
126        let end_bound = end.map(Excluded).unwrap_or(Unbounded);
127        self.clone().filter_by_epoch((start_bound, end_bound))
128    }
129
130    /// Filter measurements by duration offset.
131    ///
132    /// :type start: Duration | None
133    /// :type end: Duration | None
134    /// :rtype: TrackingDataArc
135    #[pyo3(name = "filter_by_offset")]
136    fn py_filter_by_offset(&self, start: Option<Duration>, end: Option<Duration>) -> Self {
137        let start_bound = match start {
138            Some(s) => Included(s),
139            None => Unbounded,
140        };
141        let end_bound = match end {
142            Some(e) => Excluded(e),
143            None => Unbounded,
144        };
145        self.clone().filter_by_offset((start_bound, end_bound))
146    }
147
148    /// Filter measurements by tracker alias.
149    ///
150    /// :type tracker: str
151    /// :rtype: TrackingDataArc
152    #[pyo3(name = "filter_by_tracker")]
153    fn py_filter_by_tracker(&self, tracker: String) -> Self {
154        self.clone().filter_by_tracker(tracker)
155    }
156
157    /// Filter measurements by measurement type.
158    ///
159    /// :type msr_type: MeasurementType
160    /// :rtype: TrackingDataArc
161    #[pyo3(name = "filter_by_measurement_type")]
162    fn py_filter_by_measurement_type(&self, msr_type: MeasurementType) -> Self {
163        self.clone().filter_by_measurement_type(msr_type)
164    }
165
166    /// Exclude measurements by tracker alias.
167    ///
168    /// :type tracker: str
169    /// :rtype: TrackingDataArc
170    #[pyo3(name = "exclude_tracker")]
171    fn py_exclude_tracker(&self, tracker: String) -> Self {
172        self.clone().exclude_tracker(tracker)
173    }
174
175    /// Exclude measurements by epoch range.
176    ///
177    /// :type start: Epoch | None
178    /// :type end: Epoch | None
179    /// :rtype: TrackingDataArc
180    #[pyo3(name = "exclude_by_epoch")]
181    fn py_exclude_by_epoch(&self, start: Option<Epoch>, end: Option<Epoch>) -> Self {
182        let start_bound = match start {
183            Some(s) => Included(s),
184            None => Unbounded,
185        };
186        let end_bound = match end {
187            Some(e) => Excluded(e),
188            None => Unbounded,
189        };
190        self.clone().exclude_by_epoch((start_bound, end_bound))
191    }
192
193    /// Exclude measurements by measurement type.
194    ///
195    /// :type msr_type: MeasurementType
196    /// :rtype: TrackingDataArc
197    #[pyo3(name = "exclude_measurement_type")]
198    fn py_exclude_measurement_type(&self, msr_type: MeasurementType) -> Self {
199        self.clone().exclude_measurement_type(msr_type)
200    }
201
202    /// :rtype: TrackingDataArc
203    #[pyo3(name = "resid_vs_ref_check")]
204    fn py_resid_vs_ref_check(&self) -> Self {
205        self.clone().resid_vs_ref_check()
206    }
207
208    /// Write tracking data arc to a parquet file.
209    ///
210    /// :type path: str
211    /// :type cfg: ExportCfg
212    /// :rtype: str
213    #[pyo3(name = "to_parquet")]
214    fn py_to_parquet(&self, path: String, cfg: ExportCfg) -> Result<String, InputOutputError> {
215        self.to_parquet(path, cfg)
216            .map(|pathbuf| pathbuf.to_string_lossy().into_owned())
217    }
218}
nyx_space/od/msr/trackingdata/python.rs

nyx_space/od/msr/trackingdata/
python.rs
