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 merge(&mut self, other: Self) {
101        *self = self.clone() + other;
102    }
103
104    fn __str__(&self) -> String {
105        format!("{self}")
106    }
107
108    fn __repr__(&self) -> String {
109        format!("{self} @ {self:p}")
110    }
111
112    fn __add__(&self, other: Self) -> Self {
113        self.clone() + other
114    }
115
116    #[getter]
117    fn get_force_reject(&self) -> bool {
118        self.force_reject
119    }
120
121    #[setter]
122    fn set_force_reject(&mut self, reject: bool) {
123        self.force_reject = reject;
124    }
125
126    /// Filter measurements by epoch range.
127    ///
128    /// :type start: Epoch | None
129    /// :type end: Epoch | None
130    /// :rtype: TrackingDataArc
131    #[pyo3(name = "filter_by_epoch")]
132    fn py_filter_by_epoch(&self, start: Option<Epoch>, end: Option<Epoch>) -> Self {
133        let start_bound = start.map(Included).unwrap_or(Unbounded);
134        let end_bound = end.map(Excluded).unwrap_or(Unbounded);
135        self.clone().filter_by_epoch((start_bound, end_bound))
136    }
137
138    /// Filter measurements by duration offset.
139    ///
140    /// :type start: Duration | None
141    /// :type end: Duration | None
142    /// :rtype: TrackingDataArc
143    #[pyo3(name = "filter_by_offset")]
144    fn py_filter_by_offset(&self, start: Option<Duration>, end: Option<Duration>) -> Self {
145        let start_bound = match start {
146            Some(s) => Included(s),
147            None => Unbounded,
148        };
149        let end_bound = match end {
150            Some(e) => Excluded(e),
151            None => Unbounded,
152        };
153        self.clone().filter_by_offset((start_bound, end_bound))
154    }
155
156    /// Filter measurements by tracker alias.
157    ///
158    /// :type tracker: str
159    /// :rtype: TrackingDataArc
160    #[pyo3(name = "filter_by_tracker")]
161    fn py_filter_by_tracker(&self, tracker: String) -> Self {
162        self.clone().filter_by_tracker(tracker)
163    }
164
165    /// Filter measurements by measurement type.
166    ///
167    /// :type msr_type: MeasurementType
168    /// :rtype: TrackingDataArc
169    #[pyo3(name = "filter_by_measurement_type")]
170    fn py_filter_by_measurement_type(&self, msr_type: MeasurementType) -> Self {
171        self.clone().filter_by_measurement_type(msr_type)
172    }
173
174    /// Exclude measurements by tracker alias.
175    ///
176    /// :type tracker: str
177    /// :rtype: TrackingDataArc
178    #[pyo3(name = "exclude_tracker")]
179    fn py_exclude_tracker(&self, tracker: String) -> Self {
180        self.clone().exclude_tracker(tracker)
181    }
182
183    /// Exclude measurements by epoch range.
184    ///
185    /// :type start: Epoch | None
186    /// :type end: Epoch | None
187    /// :rtype: TrackingDataArc
188    #[pyo3(name = "exclude_by_epoch")]
189    fn py_exclude_by_epoch(&self, start: Option<Epoch>, end: Option<Epoch>) -> Self {
190        let start_bound = match start {
191            Some(s) => Included(s),
192            None => Unbounded,
193        };
194        let end_bound = match end {
195            Some(e) => Excluded(e),
196            None => Unbounded,
197        };
198        self.clone().exclude_by_epoch((start_bound, end_bound))
199    }
200
201    /// Exclude measurements by measurement type.
202    ///
203    /// :type msr_type: MeasurementType
204    /// :rtype: TrackingDataArc
205    #[pyo3(name = "exclude_measurement_type")]
206    fn py_exclude_measurement_type(&self, msr_type: MeasurementType) -> Self {
207        self.clone().exclude_measurement_type(msr_type)
208    }
209
210    /// :rtype: TrackingDataArc
211    #[pyo3(name = "resid_vs_ref_check")]
212    fn py_resid_vs_ref_check(&self) -> Self {
213        self.clone().resid_vs_ref_check()
214    }
215
216    /// Write tracking data arc to a parquet file.
217    ///
218    /// :type path: str
219    /// :type cfg: ExportCfg, optional
220    /// :rtype: str
221    #[pyo3(signature = (path, cfg=None))]
222    #[pyo3(name = "to_parquet")]
223    fn py_to_parquet(
224        &self,
225        path: String,
226        cfg: Option<ExportCfg>,
227    ) -> Result<String, InputOutputError> {
228        self.to_parquet(path, cfg.unwrap_or_default())
229            .map(|pathbuf| pathbuf.to_string_lossy().into_owned())
230    }
231}
nyx_space/od/msr/trackingdata/python.rs

nyx_space/od/msr/trackingdata/
python.rs
