read implement in progress

Author: Quba1

src/atomic_message/mod.rs

@@ -0,0 +1,24 @@
+mod read;
+
+use std::sync::Arc;
+
+use eccodes_sys::codes_handle;
+
+use crate::{CodesHandle, codes_handle::ThreadSafeHandle};
+
+pub use read::{KeyRead, ArrayKeyRead, ScalarKeyRead};
+
+/// Because standard `KeyedMessage` is not Copy or Clone it can provide access methods without
+/// requiring `&mut self`. As `AtomicMessage` implements `Send + Sync` this exclusive method access is not
+/// guaranteed with just `&self`. `AtomicMessage` also implements a minimal subset of functionalities
+/// to limit the risk of some internal ecCodes functions not being thread-safe.
+///
+/// Right now `AtomicMessage` is also not clonable
+#[derive(Debug)]
+pub struct AtomicMessage<S: ThreadSafeHandle> {
+    pub(crate) _parent: Arc<CodesHandle<S>>,
+    pub(crate) message_handle: *mut codes_handle,
+}
+
+unsafe impl<S: ThreadSafeHandle> Send for AtomicMessage<S> {}
+unsafe impl<S: ThreadSafeHandle> Sync for AtomicMessage<S> {}

src/atomic_message/read.rs

@@ -0,0 +1,110 @@
+use std::cmp::Ordering;
+
+use crate::{
+    CodesError,
+    atomic_message::AtomicMessage,
+    codes_handle::ThreadSafeHandle,
+    intermediate_bindings::{
+        NativeKeyType, codes_get_bytes, codes_get_double, codes_get_double_array, codes_get_long,
+        codes_get_long_array, codes_get_native_type, codes_get_size, codes_get_string,
+    },
+};
+
+#[doc(hidden)]
+pub trait KeyReadHelpers {
+    fn get_key_size(&mut self, key_name: &str) -> Result<usize, CodesError>;
+    fn get_key_native_type(&mut self, key_name: &str) -> Result<NativeKeyType, CodesError>;
+}
+pub trait KeyRead<T>: KeyReadHelpers {
+    fn read_key_unchecked(&mut self, name: &str) -> Result<T, CodesError>;
+}
+
+pub trait ArrayKeyRead<T>: KeyRead<T> {
+    fn read_key(&mut self, key_name: &str) -> Result<T, CodesError> {
+        match self.get_key_native_type(key_name)? {
+            NativeKeyType::Bytes => (),
+            _ => return Err(CodesError::WrongRequestedKeyType),
+        }
+
+        let key_size = self.get_key_size(key_name)?;
+
+        if key_size < 1 {
+            return Err(CodesError::IncorrectKeySize);
+        }
+
+        self.read_key_unchecked(key_name)
+    }
+}
+
+pub trait ScalarKeyRead<T>: KeyRead<T> {
+    fn read_key(&mut self, key_name: &str) -> Result<T, CodesError> {
+        match self.get_key_native_type(key_name)? {
+            NativeKeyType::Long => (),
+            _ => return Err(CodesError::WrongRequestedKeyType),
+        }
+
+        let key_size = self.get_key_size(key_name)?;
+
+        match key_size.cmp(&1) {
+            Ordering::Greater => return Err(CodesError::WrongRequestedKeySize),
+            Ordering::Less => return Err(CodesError::IncorrectKeySize),
+            Ordering::Equal => (),
+        }
+
+        self.read_key_unchecked(key_name)
+    }
+}
+
+impl<S: ThreadSafeHandle> KeyReadHelpers for AtomicMessage<S> {
+    fn get_key_size(&mut self, key_name: &str) -> Result<usize, CodesError> {
+        unsafe { codes_get_size(self.message_handle, key_name) }
+    }
+
+    fn get_key_native_type(&mut self, key_name: &str) -> Result<NativeKeyType, CodesError> {
+        unsafe { codes_get_native_type(self.message_handle, key_name) }
+    }
+}
+
+impl<S: ThreadSafeHandle> KeyRead<i64> for AtomicMessage<S> {
+    fn read_key_unchecked(&mut self, key_name: &str) -> Result<i64, CodesError> {
+        unsafe { codes_get_long(self.message_handle, key_name) }
+    }
+}
+
+impl<S: ThreadSafeHandle> KeyRead<f64> for AtomicMessage<S> {
+    fn read_key_unchecked(&mut self, key_name: &str) -> Result<f64, CodesError> {
+        unsafe { codes_get_double(self.message_handle, key_name) }
+    }
+}
+
+impl<S: ThreadSafeHandle> KeyRead<String> for AtomicMessage<S> {
+    fn read_key_unchecked(&mut self, key_name: &str) -> Result<String, CodesError> {
+        unsafe { codes_get_string(self.message_handle, key_name) }
+    }
+}
+
+impl<S: ThreadSafeHandle> KeyRead<Vec<i64>> for AtomicMessage<S> {
+    fn read_key_unchecked(&mut self, key_name: &str) -> Result<Vec<i64>, CodesError> {
+        unsafe { codes_get_long_array(self.message_handle, key_name) }
+    }
+}
+
+impl<S: ThreadSafeHandle> KeyRead<Vec<f64>> for AtomicMessage<S> {
+    fn read_key_unchecked(&mut self, key_name: &str) -> Result<Vec<f64>, CodesError> {
+        unsafe { codes_get_double_array(self.message_handle, key_name) }
+    }
+}
+
+impl<S: ThreadSafeHandle> KeyRead<Vec<u8>> for AtomicMessage<S> {
+    fn read_key_unchecked(&mut self, key_name: &str) -> Result<Vec<u8>, CodesError> {
+        unsafe { codes_get_bytes(self.message_handle, key_name) }
+    }
+}
+
+impl<S: ThreadSafeHandle> ScalarKeyRead<i64> for AtomicMessage<S> {}
+impl<S: ThreadSafeHandle> ScalarKeyRead<f64> for AtomicMessage<S> {}
+
+impl<S: ThreadSafeHandle> ArrayKeyRead<String> for AtomicMessage<S> {}
+impl<S: ThreadSafeHandle> ArrayKeyRead<Vec<f64>> for AtomicMessage<S> {}
+impl<S: ThreadSafeHandle> ArrayKeyRead<Vec<i64>> for AtomicMessage<S> {}
+impl<S: ThreadSafeHandle> ArrayKeyRead<Vec<u8>> for AtomicMessage<S> {}

src/codes_handle/atomic_iterator.rs

@@ -1,11 +1,10 @@
-#![allow(unused)]
-
 use std::sync::Arc;
 
-use eccodes_sys::codes_handle;
 use fallible_iterator::FallibleIterator;
 
-use crate::{CodesError, CodesHandle, codes_handle::ThreadSafeHandle};
+use crate::{
+    CodesError, CodesHandle, atomic_message::AtomicMessage, codes_handle::ThreadSafeHandle,
+};
 
 #[derive(Debug)]
 pub struct AtomicMessageGenerator<S: ThreadSafeHandle> {
@@ -32,20 +31,12 @@ impl<S: ThreadSafeHandle> FallibleIterator for AtomicMessageGenerator<S> {
         } else {
             Ok(Some(AtomicMessage {
                 _parent: self.codes_handle.clone(),
-                pointer: new_eccodes_handle,
+                message_handle: new_eccodes_handle,
             }))
         }
     }
 }
 
-#[derive(Debug)]
-pub struct AtomicMessage<S: ThreadSafeHandle> {
-    _parent: Arc<CodesHandle<S>>,
-    pointer: *mut codes_handle,
-}
-unsafe impl<S: ThreadSafeHandle> Send for AtomicMessage<S> {}
-unsafe impl<S: ThreadSafeHandle> Sync for AtomicMessage<S> {}
-
 #[cfg(test)]
 mod tests {
     use std::{
@@ -79,7 +70,7 @@ mod tests {
                 for _ in 0..1000 {
                     b.wait();
                     let _ = unsafe {
-                        crate::intermediate_bindings::codes_get_size(msg.pointer, "shortName")
+                        crate::intermediate_bindings::codes_get_size(msg.message_handle, "shortName")
                             .unwrap()
                     };
                 }

src/codes_handle/iterator.rs

@@ -39,7 +39,7 @@ impl<'ch, S: HandleGenerator> FallibleIterator for KeyedMessageGenerator<'ch, S>
 #[cfg(test)]
 mod tests {
     use crate::{
-        DynamicKeyType, FallibleIterator,
+        keyed_message::DynamicKeyType, FallibleIterator,
         codes_handle::{CodesHandle, ProductKind},
     };
     use anyhow::{Context, Ok, Result};

src/keyed_message/read.rs

@@ -1,7 +1,7 @@
 use std::cmp::Ordering;
 
 use crate::{
-    DynamicKeyType, KeyRead, KeyedMessage,
+    keyed_message::DynamicKeyType, keyed_message::KeyRead, KeyedMessage,
     errors::CodesError,
     intermediate_bindings::{
         NativeKeyType, codes_get_bytes, codes_get_double, codes_get_double_array, codes_get_long,
@@ -300,7 +300,7 @@ mod tests {
     use anyhow::{Context, Result};
 
     use crate::codes_handle::{CodesHandle, ProductKind};
-    use crate::{DynamicKeyType, FallibleIterator};
+    use crate::{keyed_message::DynamicKeyType, FallibleIterator};
     use std::path::Path;
 
     #[test]

src/keyed_message/write.rs

@@ -135,7 +135,7 @@ mod tests {
     use fallible_iterator::FallibleIterator;
 
     use crate::{
-        DynamicKeyType, KeyWrite,
+        keyed_message::DynamicKeyType, keyed_message::KeyWrite,
         codes_handle::{CodesHandle, ProductKind},
     };
     use std::{fs::remove_file, path::Path};

src/lib.rs

@@ -222,6 +222,7 @@ pub mod keys_iterator;
 #[cfg_attr(docsrs, doc(cfg(feature = "message_ndarray")))]
 pub mod message_ndarray;
 mod pointer_guard;
+pub mod atomic_message;
 
 pub use codes_handle::{CodesHandle, ProductKind};
 #[cfg(feature = "experimental_index")]
@@ -230,5 +231,6 @@ pub use codes_index::CodesIndex;
 pub use codes_nearest::{CodesNearest, NearestGridpoint};
 pub use errors::CodesError;
 pub use fallible_iterator::{FallibleIterator, IntoFallibleIterator};
-pub use keyed_message::{DynamicKeyType, KeyRead, KeyWrite, KeyedMessage};
+pub use keyed_message::{KeyedMessage};
 pub use keys_iterator::{KeysIterator, KeysIteratorFlags};
+pub use atomic_message::{AtomicMessage};

src/message_ndarray.rs

@@ -3,7 +3,13 @@
 
 use ndarray::{Array2, Array3, s};
 
-use crate::{CodesError, KeyRead, KeyedMessage, errors::MessageNdarrayError};
+use crate::{
+    AtomicMessage, CodesError, KeyedMessage,
+    atomic_message::{ArrayKeyRead, ScalarKeyRead},
+    codes_handle::ThreadSafeHandle,
+    errors::MessageNdarrayError,
+    keyed_message::KeyRead,
+};
 
 /// Struct returned by [`KeyedMessage::to_lons_lats_values()`] method.
 /// The arrays are collocated, meaning that `longitudes[i, j]` and `latitudes[i, j]` are the coordinates of `values[i, j]`.
@@ -138,15 +144,135 @@ impl KeyedMessage<'_> {
     }
 }
 
+impl<S: ThreadSafeHandle> AtomicMessage<S> {
+    /// Converts the message to a 2D ndarray.
+    ///
+    /// Returns ndarray where first dimension represents y coordinates and second dimension represents x coordinates,
+    /// ie. `[lat, lon]`.
+    ///
+    /// Common convention for grib files on regular lon-lat grid assumes that:
+    /// index `[0, 0]` is the top-left corner of the grid:
+    /// x coordinates are increasing with the i index,
+    /// y coordinates are decreasing with the j index.
+    ///
+    /// This convention can be checked with `iScansNegatively` and `jScansPositively` keys -
+    /// if both are false, the above convention is used.
+    ///
+    /// Requires the keys `Ni`, `Nj` and `values` to be present in the message.
+    ///
+    /// Tested only with simple lat-lon grids.
+    ///
+    /// # Errors
+    ///
+    /// - When the required keys are not present or if their values are not of the expected type
+    /// - When the number of values mismatch with the `Ni` and `Nj` keys
+    #[cfg_attr(docsrs, doc(cfg(feature = "message_ndarray")))]
+    pub fn to_ndarray(&self) -> Result<Array2<f64>, CodesError> {
+        let ni: i64 = self.read_key("Ni")?;
+        let ni = usize::try_from(ni).map_err(MessageNdarrayError::from)?;
+
+        let nj: i64 = self.read_key("Nj")?;
+        let nj = usize::try_from(nj).map_err(MessageNdarrayError::from)?;
+
+        let vals: Vec<f64> = self.read_key("values")?;
+        if vals.len() != (ni * nj) {
+            return Err(MessageNdarrayError::UnexpectedValuesLength(vals.len(), ni * nj).into());
+        }
+
+        let j_scanning: i64 = self.read_key("jPointsAreConsecutive")?;
+
+        if ![0, 1].contains(&j_scanning) {
+            return Err(MessageNdarrayError::UnexpectedKeyValue(
+                "jPointsAreConsecutive".to_owned(),
+            )
+            .into());
+        }
+
+        let j_scanning = j_scanning != 0;
+
+        let shape = if j_scanning { (ni, nj) } else { (nj, ni) };
+        let vals = Array2::from_shape_vec(shape, vals).map_err(MessageNdarrayError::from)?;
+
+        if j_scanning {
+            Ok(vals.reversed_axes())
+        } else {
+            Ok(vals)
+        }
+    }
+
+    /// Same as [`KeyedMessage::to_ndarray()`] but returns the longitudes and latitudes alongside values.
+    /// Fields are returned as separate arrays in [`RustyCodesMessage`].
+    ///
+    /// Compared to `to_ndarray` this method has performance overhead as returned arrays may be cloned.
+    ///
+    /// This method requires the `latLonValues`, `Ni` and `Nj` keys to be present in the message.
+    ///
+    /// # Errors
+    ///
+    /// - When the required keys are not present or if their values are not of the expected type
+    /// - When the number of values mismatch with the `Ni` and `Nj` keys
+    #[cfg_attr(docsrs, doc(cfg(feature = "message_ndarray")))]
+    pub fn to_lons_lats_values(&self) -> Result<RustyCodesMessage, CodesError> {
+        let ni: i64 = self.read_key("Ni")?;
+        let ni = usize::try_from(ni).map_err(MessageNdarrayError::from)?;
+
+        let nj: i64 = self.read_key("Nj")?;
+        let nj = usize::try_from(nj).map_err(MessageNdarrayError::from)?;
+
+        let latlonvals: Vec<f64> = self.read_key("latLonValues")?;
+
+        if latlonvals.len() != (ni * nj * 3) {
+            return Err(
+                MessageNdarrayError::UnexpectedValuesLength(latlonvals.len(), ni * nj * 3).into(),
+            );
+        }
+
+        let j_scanning: i64 = self.read_key("jPointsAreConsecutive")?;
+
+        if ![0, 1].contains(&j_scanning) {
+            return Err(MessageNdarrayError::UnexpectedKeyValue(
+                "jPointsAreConsecutive".to_owned(),
+            )
+            .into());
+        }
+
+        let j_scanning = j_scanning != 0;
+
+        let shape = if j_scanning {
+            (ni, nj, 3_usize)
+        } else {
+            (nj, ni, 3_usize)
+        };
+
+        let mut latlonvals =
+            Array3::from_shape_vec(shape, latlonvals).map_err(MessageNdarrayError::from)?;
+
+        if j_scanning {
+            latlonvals.swap_axes(0, 1);
+        }
+
+        let (lats, lons, vals) =
+            latlonvals
+                .view_mut()
+                .multi_slice_move((s![.., .., 0], s![.., .., 1], s![.., .., 2]));
+
+        Ok(RustyCodesMessage {
+            longitudes: lons.into_owned(),
+            latitudes: lats.into_owned(),
+            values: vals.into_owned(),
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use fallible_iterator::FallibleIterator;
     use float_cmp::assert_approx_eq;
 
     use super::*;
-    use crate::DynamicKeyType;
     use crate::ProductKind;
     use crate::codes_handle::CodesHandle;
+    use crate::keyed_message::DynamicKeyType;
     use std::path::Path;
 
     #[test]

tests/handle.rs

@@ -1,7 +1,7 @@
 use std::{path::Path, thread};
 
 use anyhow::{Context, Result};
-use eccodes::{CodesHandle, DynamicKeyType, FallibleIterator, ProductKind};
+use eccodes::{CodesHandle, keyed_message::DynamicKeyType, FallibleIterator, ProductKind};
 
 #[test]
 fn thread_safety() {