use crate::error::{to_py_result, PyResult};
use pyo3::exceptions::PyStopIteration;
use pyo3::prelude::*;
use pyo3::types::PyModule;
use std::sync::{Arc, Mutex};
use torsh_core::device::DeviceType;
use torsh_core::error::Result as TorshResult;
use torsh_data::dataloader::{simple_dataloader, simple_random_dataloader};
use torsh_data::dataloader::{SimpleDataLoader, SimpleRandomDataLoader};
use torsh_data::dataset::Dataset;
use torsh_tensor::Tensor;
#[derive(Clone)]
struct InnerDataset {
samples: Vec<Vec<f32>>,
}
impl Dataset for InnerDataset {
type Item = Vec<Tensor<f32>>;
fn len(&self) -> usize {
self.samples.len()
}
fn get(&self, index: usize) -> TorshResult<Self::Item> {
if index >= self.samples.len() {
return Err(torsh_core::error::TorshError::IndexOutOfBounds {
index,
size: self.samples.len(),
});
}
let row = &self.samples[index];
let n = row.len();
let tensor = Tensor::from_data(row.clone(), vec![n], DeviceType::Cpu)?;
Ok(vec![tensor])
}
}
#[pyclass(name = "Dataset")]
pub struct PyDataset {
inner: Arc<InnerDataset>,
}
#[pymethods]
impl PyDataset {
#[new]
pub fn new(samples: Vec<Vec<f32>>) -> Self {
Self {
inner: Arc::new(InnerDataset { samples }),
}
}
fn __len__(&self) -> usize {
self.inner.samples.len()
}
fn __getitem__(&self, index: usize) -> PyResult<Vec<f32>> {
if index >= self.inner.samples.len() {
return Err(pyo3::exceptions::PyIndexError::new_err(format!(
"Index {} out of range for dataset of length {}",
index,
self.inner.samples.len()
)));
}
Ok(self.inner.samples[index].clone())
}
#[getter]
fn len(&self) -> usize {
self.inner.samples.len()
}
#[getter]
fn is_empty(&self) -> bool {
self.inner.samples.is_empty()
}
fn __repr__(&self) -> String {
format!("Dataset(len={})", self.inner.samples.len())
}
}
type Batch = Vec<Vec<f32>>;
fn materialise_batches_sequential(loader: &SimpleDataLoader<InnerDataset>) -> Vec<Batch> {
let mut batches = Vec::new();
for result in loader.iter() {
if let Ok(tensors) = result {
let rows = tensors_to_rows(&tensors);
batches.push(rows);
}
}
batches
}
fn materialise_batches_random(loader: &SimpleRandomDataLoader<InnerDataset>) -> Vec<Batch> {
let mut batches = Vec::new();
for result in loader.iter() {
if let Ok(tensors) = result {
let rows = tensors_to_rows(&tensors);
batches.push(rows);
}
}
batches
}
fn tensors_to_rows(tensors: &[Tensor<f32>]) -> Vec<Vec<f32>> {
if tensors.is_empty() {
return Vec::new();
}
let t = &tensors[0];
let shape = t.shape().dims().to_vec();
if shape.is_empty() {
return Vec::new();
}
let batch_size = shape[0];
let feature_size: usize = if shape.len() > 1 {
shape[1..].iter().product()
} else {
1
};
let flat: Vec<f32> = t.data().unwrap_or_default().to_vec();
let mut rows = Vec::with_capacity(batch_size);
for b in 0..batch_size {
let start = b * feature_size;
let end = start + feature_size;
if end <= flat.len() {
rows.push(flat[start..end].to_vec());
}
}
rows
}
#[pyclass(name = "DataLoaderIter")]
pub struct PyDataLoaderIter {
batches: Arc<Mutex<Vec<Batch>>>,
cursor: usize,
total: usize,
}
#[pymethods]
impl PyDataLoaderIter {
fn __iter__(slf: PyRef<'_, Self>) -> PyRef<'_, Self> {
slf
}
fn __next__(&mut self) -> PyResult<Vec<Vec<f32>>> {
if self.cursor >= self.total {
return Err(PyStopIteration::new_err("exhausted"));
}
let guard = self.batches.lock().expect("lock should not be poisoned");
let batch = guard[self.cursor].clone();
drop(guard);
self.cursor += 1;
Ok(batch)
}
fn __len__(&self) -> usize {
self.total - self.cursor
}
}
#[pyclass(name = "DataLoader")]
pub struct PyDataLoader {
batches: Arc<Mutex<Vec<Batch>>>,
num_batches: usize,
batch_size: usize,
dataset_len: usize,
shuffle: bool,
}
#[pymethods]
impl PyDataLoader {
#[new]
#[pyo3(signature = (dataset, batch_size=1, shuffle=false, drop_last=false, generator=None))]
pub fn new(
dataset: &PyDataset,
batch_size: usize,
shuffle: bool,
drop_last: bool,
generator: Option<u64>,
) -> PyResult<Self> {
let _ = drop_last; let inner = (*dataset.inner).clone();
let dataset_len = inner.samples.len();
let batches: Vec<Batch> = if shuffle {
let loader = to_py_result(simple_random_dataloader(inner, batch_size, generator))?;
materialise_batches_random(&loader)
} else {
let loader = to_py_result(simple_dataloader(inner, batch_size, false))?;
materialise_batches_sequential(&loader)
};
let num_batches = batches.len();
Ok(Self {
batches: Arc::new(Mutex::new(batches)),
num_batches,
batch_size,
dataset_len,
shuffle,
})
}
fn __len__(&self) -> usize {
self.num_batches
}
fn __iter__(&self) -> PyDataLoaderIter {
PyDataLoaderIter {
batches: Arc::clone(&self.batches),
cursor: 0,
total: self.num_batches,
}
}
#[getter]
fn len(&self) -> usize {
self.num_batches
}
#[getter]
fn is_empty(&self) -> bool {
self.num_batches == 0
}
#[getter]
fn batch_size(&self) -> usize {
self.batch_size
}
#[getter]
fn shuffle(&self) -> bool {
self.shuffle
}
#[getter]
fn dataset_len(&self) -> usize {
self.dataset_len
}
fn __repr__(&self) -> String {
format!(
"DataLoader(dataset_len={}, batch_size={}, shuffle={}, num_batches={})",
self.dataset_len, self.batch_size, self.shuffle, self.num_batches
)
}
}
pub fn register_data_module(_py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_class::<PyDataset>()?;
m.add_class::<PyDataLoader>()?;
m.add_class::<PyDataLoaderIter>()?;
Ok(())
}