nyx_space/md/trajectory/
traj.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::traj_it::TrajIterator;
20use super::{ExportCfg, InterpolationSnafu, INTERPOLATION_SAMPLES};
21use super::{Interpolatable, TrajError};
22use crate::errors::NyxError;
23use crate::io::watermark::pq_writer;
24use crate::io::InputOutputError;
25use crate::linalg::allocator::Allocator;
26use crate::linalg::DefaultAllocator;
27use crate::md::prelude::{GuidanceMode, StateParameter};
28use crate::md::trajectory::smooth_state_diff_in_place;
29use crate::time::{Duration, Epoch, TimeSeries, TimeUnits};
30use anise::analysis::specs::StateSpecTrait;
31use anise::analysis::AnalysisError;
32use anise::astro::orbit::Orbit;
33use anise::prelude::{Aberration, Almanac};
34use arrow::array::{Array, Float64Builder, StringBuilder};
35use arrow::datatypes::{DataType, Field, Schema};
36use arrow::record_batch::RecordBatch;
37use hifitime::TimeScale;
38use log::{info, warn};
39use parquet::arrow::ArrowWriter;
40use snafu::ResultExt;
41use std::collections::HashMap;
42use std::error::Error;
43use std::fmt;
44use std::fs::File;
45use std::iter::Iterator;
46use std::ops;
47use std::ops::Bound::{Excluded, Included, Unbounded};
48use std::path::{Path, PathBuf};
49use std::sync::Arc;
50
51/// Store a trajectory of any State.
52#[derive(Clone, PartialEq)]
53pub struct Traj<S: Interpolatable>
54where
55    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
56{
57    /// Optionally name this trajectory
58    pub name: Option<String>,
59    /// We use a vector because we know that the states are produced in a chronological manner (the direction does not matter).
60    pub states: Vec<S>,
61}
62
63impl<S: Interpolatable> Traj<S>
64where
65    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
66{
67    pub fn new() -> Self {
68        Self {
69            name: None,
70            states: Vec::new(),
71        }
72    }
73    /// Orders the states, can be used to store the states out of order
74    pub fn finalize(&mut self) {
75        // Remove duplicate epochs
76        self.states.dedup_by(|a, b| a.epoch().eq(&b.epoch()));
77        // And sort
78        self.states.sort_by_key(|a| a.epoch());
79    }
80
81    /// Evaluate the trajectory at this specific epoch.
82    pub fn at(&self, epoch: Epoch) -> Result<S, TrajError> {
83        if self.states.is_empty() || self.first().epoch() > epoch || self.last().epoch() < epoch {
84            return Err(TrajError::NoInterpolationData { epoch });
85        }
86        match self
87            .states
88            .binary_search_by(|state| state.epoch().cmp(&epoch))
89        {
90            Ok(idx) => {
91                // Oh wow, we actually had this exact state!
92                Ok(self.states[idx])
93            }
94            Err(idx) => {
95                if idx == 0 || idx >= self.states.len() {
96                    // The binary search returns where we should insert the data, so if it's at either end of the list, then we're out of bounds.
97                    // This condition should have been handled by the check at the start of this function.
98                    return Err(TrajError::NoInterpolationData { epoch });
99                }
100                // This is the closest index, so let's grab the items around it.
101                // NOTE: This is essentially the same code as in ANISE for the Hermite SPK type 13
102
103                // We didn't find it, so let's build an interpolation here.
104                let num_left = INTERPOLATION_SAMPLES / 2;
105
106                // Ensure that we aren't fetching out of the window
107                let mut first_idx = idx.saturating_sub(num_left);
108                let last_idx = self.states.len().min(first_idx + INTERPOLATION_SAMPLES);
109
110                // Check that we have enough samples
111                if last_idx == self.states.len() {
112                    first_idx = last_idx.saturating_sub(2 * num_left);
113                }
114
115                let mut states = Vec::with_capacity(last_idx - first_idx);
116                for idx in first_idx..last_idx {
117                    states.push(self.states[idx]);
118                }
119
120                self.states[idx]
121                    .interpolate(epoch, &states)
122                    .context(InterpolationSnafu)
123            }
124        }
125    }
126
127    /// Returns the first state in this ephemeris
128    pub fn first(&self) -> &S {
129        // This is done after we've ordered the states we received, so we can just return the first state.
130        self.states.first().unwrap()
131    }
132
133    /// Returns the last state in this ephemeris
134    pub fn last(&self) -> &S {
135        self.states.last().unwrap()
136    }
137
138    pub fn start_epoch(&self) -> Epoch {
139        self.first().epoch()
140    }
141
142    pub fn end_epoch(&self) -> Epoch {
143        self.last().epoch()
144    }
145
146    /// Creates an iterator through the trajectory by the provided step size
147    pub fn every(&self, step: Duration) -> TrajIterator<'_, S> {
148        self.every_between(step, self.first().epoch(), self.last().epoch())
149    }
150
151    /// Creates an iterator through the trajectory by the provided step size between the provided bounds
152    pub fn every_between(&self, step: Duration, start: Epoch, end: Epoch) -> TrajIterator<'_, S> {
153        TrajIterator {
154            time_series: TimeSeries::inclusive(
155                start.max(self.first().epoch()),
156                end.min(self.last().epoch()),
157                step,
158            ),
159            traj: self,
160        }
161    }
162
163    /// Returns a new trajectory that only contains states that fall within the given epoch range.
164    pub fn filter_by_epoch<R: ops::RangeBounds<Epoch>>(mut self, bound: R) -> Self {
165        self.states = self
166            .states
167            .iter()
168            .copied()
169            .filter(|s| bound.contains(&s.epoch()))
170            .collect::<Vec<_>>();
171        self
172    }
173
174    /// Returns a new trajectory that only contains states that fall within the given offset from the first epoch.
175    /// For example, a bound of 30.minutes()..90.minutes() will only return states from the start of the trajectory + 30 minutes until start + 90 minutes.
176    pub fn filter_by_offset<R: ops::RangeBounds<Duration>>(self, bound: R) -> Self {
177        if self.states.is_empty() {
178            return self;
179        }
180        // Rebuild an epoch bound.
181        let start = match bound.start_bound() {
182            Unbounded => self.states.first().unwrap().epoch(),
183            Included(offset) | Excluded(offset) => self.states.first().unwrap().epoch() + *offset,
184        };
185
186        let end = match bound.end_bound() {
187            Unbounded => self.states.last().unwrap().epoch(),
188            Included(offset) | Excluded(offset) => self.states.first().unwrap().epoch() + *offset,
189        };
190
191        self.filter_by_epoch(start..=end)
192    }
193    /// Store this trajectory arc to a parquet file with the default configuration (depends on the state type, search for `export_params` in the documentation for details).
194    pub fn to_parquet_simple<P: AsRef<Path>>(&self, path: P) -> Result<PathBuf, Box<dyn Error>> {
195        self.to_parquet(path, ExportCfg::default())
196    }
197
198    /// Store this trajectory arc to a parquet file with the provided configuration
199    pub fn to_parquet_with_cfg<P: AsRef<Path>>(
200        &self,
201        path: P,
202        cfg: ExportCfg,
203    ) -> Result<PathBuf, Box<dyn Error>> {
204        self.to_parquet(path, cfg)
205    }
206
207    /// A shortcut to `to_parquet_with_cfg`
208    pub fn to_parquet_with_step<P: AsRef<Path>>(
209        &self,
210        path: P,
211        step: Duration,
212    ) -> Result<(), Box<dyn Error>> {
213        self.to_parquet_with_cfg(
214            path,
215            ExportCfg {
216                step: Some(step),
217                ..Default::default()
218            },
219        )?;
220
221        Ok(())
222    }
223
224    /// Store this trajectory arc to a parquet file with the provided configuration
225    pub fn to_parquet<P: AsRef<Path>>(
226        &self,
227        path: P,
228        cfg: ExportCfg,
229    ) -> Result<PathBuf, Box<dyn Error>> {
230        let tick = Epoch::now().unwrap();
231        info!("Exporting trajectory to parquet file...");
232
233        // Grab the path here before we move stuff.
234        let path_buf = cfg.actual_path(path);
235
236        // Build the schema
237        let mut hdrs = vec![Field::new("Epoch (UTC)", DataType::Utf8, false)];
238
239        let frame = self.states[0].frame();
240        let more_meta = Some(vec![(
241            "Frame".to_string(),
242            serde_dhall::serialize(&frame)
243                .static_type_annotation()
244                .to_string()
245                .map_err(|e| {
246                    Box::new(InputOutputError::SerializeDhall {
247                        what: format!("frame `{frame}`"),
248                        err: e.to_string(),
249                    })
250                })?,
251        )]);
252
253        let mut fields = match cfg.fields {
254            Some(fields) => fields,
255            None => S::export_params(),
256        };
257
258        // Check that we can retrieve this information
259        fields.retain(|param| self.first().value(*param).is_ok());
260
261        for field in &fields {
262            hdrs.push(field.to_field(more_meta.clone()));
263        }
264
265        // Build the schema
266        let schema = Arc::new(Schema::new(hdrs));
267        let mut record: Vec<Arc<dyn Array>> = Vec::new();
268
269        // Build the states iterator -- this does require copying the current states but I can't either get a reference or a copy of all the states.
270        let states = if cfg.start_epoch.is_some() || cfg.end_epoch.is_some() || cfg.step.is_some() {
271            // Must interpolate the data!
272            let start = cfg.start_epoch.unwrap_or_else(|| self.first().epoch());
273            let end = cfg.end_epoch.unwrap_or_else(|| self.last().epoch());
274            let step = cfg.step.unwrap_or_else(|| 1.minutes());
275            self.every_between(step, start, end).collect::<Vec<S>>()
276        } else {
277            self.states.to_vec()
278        };
279
280        // Build all of the records
281
282        // Epochs
283        let mut utc_epoch = StringBuilder::new();
284        for s in &states {
285            utc_epoch.append_value(s.epoch().to_time_scale(TimeScale::UTC).to_isoformat());
286        }
287        record.push(Arc::new(utc_epoch.finish()));
288
289        // Add all of the fields
290        for field in fields {
291            if field == StateParameter::GuidanceMode {
292                let mut guid_mode = StringBuilder::new();
293                for s in &states {
294                    guid_mode
295                        .append_value(format!("{:?}", GuidanceMode::from(s.value(field).unwrap())));
296                }
297                record.push(Arc::new(guid_mode.finish()));
298            } else {
299                let mut data = Float64Builder::new();
300                for s in &states {
301                    data.append_value(s.value(field).unwrap());
302                }
303                record.push(Arc::new(data.finish()));
304            }
305        }
306
307        info!(
308            "Serialized {} states from {} to {}",
309            states.len(),
310            states.first().unwrap().epoch(),
311            states.last().unwrap().epoch()
312        );
313
314        // Serialize all of the devices and add that to the parquet file too.
315        let mut metadata = HashMap::new();
316        metadata.insert("Purpose".to_string(), "Trajectory data".to_string());
317        if let Some(add_meta) = cfg.metadata {
318            for (k, v) in add_meta {
319                metadata.insert(k, v);
320            }
321        }
322
323        let props = pq_writer(Some(metadata));
324
325        let file = File::create(&path_buf)?;
326        let mut writer = ArrowWriter::try_new(file, schema.clone(), props).unwrap();
327
328        let batch = RecordBatch::try_new(schema, record)?;
329        writer.write(&batch)?;
330        writer.close()?;
331
332        // Return the path this was written to
333        let tock_time = Epoch::now().unwrap() - tick;
334        info!(
335            "Trajectory written to {} in {tock_time}",
336            path_buf.display()
337        );
338        Ok(path_buf)
339    }
340
341    /// Allows resampling this trajectory at a fixed interval instead of using the propagator step size.
342    /// This may lead to aliasing due to the Nyquist–Shannon sampling theorem.
343    pub fn resample(&self, step: Duration) -> Result<Self, NyxError> {
344        if self.states.is_empty() {
345            return Err(NyxError::Trajectory {
346                source: TrajError::CreationError {
347                    msg: "No trajectory to convert".to_string(),
348                },
349            });
350        }
351
352        let mut traj = Self::new();
353        for state in self.every(step) {
354            traj.states.push(state);
355        }
356
357        traj.finalize();
358
359        Ok(traj)
360    }
361
362    /// Rebuilds this trajectory with the provided epochs.
363    /// This may lead to aliasing due to the Nyquist–Shannon sampling theorem.
364    pub fn rebuild(&self, epochs: &[Epoch]) -> Result<Self, NyxError> {
365        if self.states.is_empty() {
366            return Err(NyxError::Trajectory {
367                source: TrajError::CreationError {
368                    msg: "No trajectory to convert".to_string(),
369                },
370            });
371        }
372
373        let mut traj = Self::new();
374        for epoch in epochs {
375            traj.states.push(self.at(*epoch)?);
376        }
377
378        traj.finalize();
379
380        Ok(traj)
381    }
382
383    /// Export the difference in RIC from of this trajectory compare to the "other" trajectory in parquet format.
384    ///
385    /// # Notes
386    /// + The RIC frame accounts for the transport theorem by performing a finite differencing of the RIC frame.
387    pub fn ric_diff_to_parquet<P: AsRef<Path>>(
388        &self,
389        other: &Self,
390        path: P,
391        cfg: ExportCfg,
392    ) -> Result<PathBuf, Box<dyn Error>> {
393        let tick = Epoch::now().unwrap();
394        info!("Exporting trajectory to parquet file...");
395
396        // Grab the path here before we move stuff.
397        let path_buf = cfg.actual_path(path);
398
399        // Build the schema
400        let mut hdrs = vec![Field::new("Epoch (UTC)", DataType::Utf8, false)];
401
402        // Add the RIC headers
403        for coord in ["X", "Y", "Z"] {
404            let mut meta = HashMap::new();
405            meta.insert("unit".to_string(), "km".to_string());
406
407            let field = Field::new(
408                format!("Delta {coord} (RIC) (km)"),
409                DataType::Float64,
410                false,
411            )
412            .with_metadata(meta);
413
414            hdrs.push(field);
415        }
416
417        for coord in ["x", "y", "z"] {
418            let mut meta = HashMap::new();
419            meta.insert("unit".to_string(), "km/s".to_string());
420
421            let field = Field::new(
422                format!("Delta V{coord} (RIC) (km/s)"),
423                DataType::Float64,
424                false,
425            )
426            .with_metadata(meta);
427
428            hdrs.push(field);
429        }
430
431        let frame = self.states[0].frame();
432        let more_meta = Some(vec![(
433            "Frame".to_string(),
434            serde_dhall::serialize(&frame)
435                .static_type_annotation()
436                .to_string()
437                .unwrap_or(frame.to_string()),
438        )]);
439
440        let mut cfg = cfg;
441
442        let mut fields = match cfg.fields {
443            Some(fields) => fields,
444            None => S::export_params(),
445        };
446
447        // Remove disallowed field and check that we can retrieve this information
448        fields.retain(|param| {
449            param != &StateParameter::GuidanceMode && self.first().value(*param).is_ok()
450        });
451
452        for field in &fields {
453            hdrs.push(field.to_field(more_meta.clone()));
454        }
455
456        // Build the schema
457        let schema = Arc::new(Schema::new(hdrs));
458        let mut record: Vec<Arc<dyn Array>> = Vec::new();
459
460        // Ensure the times match.
461        cfg.start_epoch = if self.first().epoch() > other.first().epoch() {
462            Some(self.first().epoch())
463        } else {
464            Some(other.first().epoch())
465        };
466
467        cfg.end_epoch = if self.last().epoch() > other.last().epoch() {
468            Some(other.last().epoch())
469        } else {
470            Some(self.last().epoch())
471        };
472
473        // Build the states iterator
474        let step = cfg.step.unwrap_or_else(|| 1.minutes());
475        let self_states = self
476            .every_between(step, cfg.start_epoch.unwrap(), cfg.end_epoch.unwrap())
477            .collect::<Vec<S>>();
478
479        let other_states = other
480            .every_between(step, cfg.start_epoch.unwrap(), cfg.end_epoch.unwrap())
481            .collect::<Vec<S>>();
482
483        // Build an array of all the RIC differences
484        let mut ric_diff = Vec::with_capacity(other_states.len());
485        for (other_state, self_state) in other_states.iter().zip(self_states.iter()) {
486            let self_orbit = self_state.orbit();
487            let other_orbit = other_state.orbit();
488
489            let this_ric_diff = self_orbit.ric_difference(&other_orbit).map_err(Box::new)?;
490
491            ric_diff.push(this_ric_diff);
492        }
493
494        smooth_state_diff_in_place(&mut ric_diff, if other_states.len() > 5 { 5 } else { 1 });
495
496        // Build all of the records
497
498        // Epochs (both match for self and others)
499        let mut utc_epoch = StringBuilder::new();
500        for s in &self_states {
501            utc_epoch.append_value(s.epoch().to_time_scale(TimeScale::UTC).to_isoformat());
502        }
503        record.push(Arc::new(utc_epoch.finish()));
504
505        // Add the RIC data
506        for coord_no in 0..6 {
507            let mut data = Float64Builder::new();
508            for this_ric_dff in &ric_diff {
509                data.append_value(this_ric_dff.to_cartesian_pos_vel()[coord_no]);
510            }
511            record.push(Arc::new(data.finish()));
512        }
513
514        // Add all of the fields
515        for field in fields {
516            let mut data = Float64Builder::new();
517            for (other_state, self_state) in other_states.iter().zip(self_states.iter()) {
518                let self_val = self_state.value(field).map_err(Box::new)?;
519                let other_val = other_state.value(field).map_err(Box::new)?;
520
521                data.append_value(self_val - other_val);
522            }
523
524            record.push(Arc::new(data.finish()));
525        }
526
527        info!("Serialized {} states differences", self_states.len());
528
529        // Serialize all of the devices and add that to the parquet file too.
530        let mut metadata = HashMap::new();
531        metadata.insert(
532            "Purpose".to_string(),
533            "Trajectory difference data".to_string(),
534        );
535        if let Some(add_meta) = cfg.metadata {
536            for (k, v) in add_meta {
537                metadata.insert(k, v);
538            }
539        }
540
541        let props = pq_writer(Some(metadata));
542
543        let file = File::create(&path_buf)?;
544        let mut writer = ArrowWriter::try_new(file, schema.clone(), props).unwrap();
545
546        let batch = RecordBatch::try_new(schema, record)?;
547        writer.write(&batch)?;
548        writer.close()?;
549
550        // Return the path this was written to
551        let tock_time = Epoch::now().unwrap() - tick;
552        info!(
553            "Trajectory written to {} in {tock_time}",
554            path_buf.display()
555        );
556        Ok(path_buf)
557    }
558}
559
560impl<S: Interpolatable> ops::Add for Traj<S>
561where
562    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
563{
564    type Output = Result<Traj<S>, NyxError>;
565
566    /// Add one trajectory to another. If they do not overlap to within 10ms, a warning will be printed.
567    fn add(self, other: Traj<S>) -> Self::Output {
568        &self + &other
569    }
570}
571
572impl<S: Interpolatable> ops::Add<&Traj<S>> for &Traj<S>
573where
574    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
575{
576    type Output = Result<Traj<S>, NyxError>;
577
578    /// Add one trajectory to another, returns an error if the frames don't match
579    fn add(self, other: &Traj<S>) -> Self::Output {
580        if self.first().frame() != other.first().frame() {
581            Err(NyxError::Trajectory {
582                source: TrajError::CreationError {
583                    msg: format!(
584                        "Frame mismatch in add operation: {} != {}",
585                        self.first().frame(),
586                        other.first().frame()
587                    ),
588                },
589            })
590        } else {
591            if self.last().epoch() < other.first().epoch() {
592                let gap = other.first().epoch() - self.last().epoch();
593                warn!(
594                    "Resulting merged trajectory will have a time-gap of {} starting at {}",
595                    gap,
596                    self.last().epoch()
597                );
598            }
599
600            let mut me = self.clone();
601            // Now start adding the other segments while correcting the index
602            for state in &other
603                .states
604                .iter()
605                .copied()
606                .filter(|s| s.epoch() > self.last().epoch())
607                .collect::<Vec<S>>()
608            {
609                me.states.push(*state);
610            }
611            me.finalize();
612
613            Ok(me)
614        }
615    }
616}
617
618impl<S: Interpolatable> ops::AddAssign<&Traj<S>> for Traj<S>
619where
620    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
621{
622    /// Attempt to add two trajectories together and assign it to `self`
623    ///
624    /// # Warnings
625    /// 1. This will panic if the frames mismatch!
626    /// 2. This is inefficient because both `self` and `rhs` are cloned.
627    fn add_assign(&mut self, rhs: &Self) {
628        *self = (self.clone() + rhs.clone()).unwrap();
629    }
630}
631
632impl<S: Interpolatable> fmt::Display for Traj<S>
633where
634    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
635{
636    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
637        if self.states.is_empty() {
638            write!(f, "Empty Trajectory!")
639        } else {
640            let dur = self.last().epoch() - self.first().epoch();
641            write!(
642                f,
643                "Trajectory {}in {} from {} to {} ({}, or {:.3} s) [{} states]",
644                match &self.name {
645                    Some(name) => format!("of {name} "),
646                    None => String::new(),
647                },
648                self.first().frame(),
649                self.first().epoch(),
650                self.last().epoch(),
651                dur,
652                dur.to_seconds(),
653                self.states.len()
654            )
655        }
656    }
657}
658
659impl<S: Interpolatable> fmt::Debug for Traj<S>
660where
661    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
662{
663    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
664        write!(f, "{self}",)
665    }
666}
667
668impl<S: Interpolatable> Default for Traj<S>
669where
670    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
671{
672    fn default() -> Self {
673        Self::new()
674    }
675}
676
677impl<S: Interpolatable> StateSpecTrait for Traj<S>
678where
679    DefaultAllocator: Allocator<S::VecLength> + Allocator<S::Size> + Allocator<S::Size, S::Size>,
680{
681    fn ab_corr(&self) -> Option<Aberration> {
682        None
683    }
684
685    fn evaluate(&self, epoch: Epoch, _almanac: &Almanac) -> Result<Orbit, AnalysisError> {
686        self.at(epoch)
687            .map(|state| state.orbit())
688            .map_err(|e| AnalysisError::GenericAnalysisError {
689                err: format!("{e}"),
690            })
691    }
692}
nyx_space/md/trajectory/traj.rs

nyx_space/md/trajectory/
traj.rs
