splimes 0.1.0

Spline interpolation over irregularly sampled time series, on SIMD/parallel CPU or GPU (wgpu), with BigDecimal values.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101

use std::io::Write;

use anyhow::{Context, Result, bail};
use bigdecimal::FromPrimitive;
use chrono::{DateTime, Utc};
use rayon::prelude::*;

use crate::{
	Error, POINT_SIZE, Point, Resolution, Spline, generate_target_times, helpers::{InterpolationState, batch}, splines::{cubic, cubic_simd, linear, linear_simd, polynomial, polynomial_simd, quadratic, quadratic_simd}
};

mod fast_path;
pub use fast_path::apply_fast_path; // Re-export apply_fast_path

/// # Errors
/// todo
pub async fn cpu_interpolate(points: &mut [Point], start: DateTime<Utc>, end: DateTime<Utc>, resolution: Resolution, spline: Spline) -> Result<Vec<Point>> {
	let _input_count = points.len();
	let _target_times = generate_target_times(start, end, resolution);
	let _output_count = _target_times.len();

	match spline {
		Spline::Linear => linear(points, &start, &end, &resolution).await,
		Spline::Quadratic => quadratic(points, &start, &end, &resolution).await,
		Spline::Cubic => cubic(points, &start, &end, &resolution).await,
		Spline::Polynomial(_, _) => polynomial(points, &start, &end, &resolution, &spline).await,
	}
}

/// # Errors
/// todo
pub async fn parallel_interpolate(points: &mut [Point], start: &DateTime<Utc>, end: &DateTime<Utc>, spline: Spline, resolution: Resolution) -> Result<Vec<Point>> {
	spline.pre_check(points, start, end)?;
	batch(points, start, end, &spline, &resolution, |state| Box::pin(p_interpolate(state))).await
}

// Fixed p_interpolate function in mod.rs
// The key fix is to use state.result instead of always creating temp files

pub async fn p_interpolate(state: &mut InterpolationState) -> Result<()> {
	let Some(input_points) = &state.input_points else {
		bail!("No input points provided for interpolation");
	};
	let Some(batch_times) = &state.batch_times else {
		bail!("No batch times provided for interpolation");
	};

	state.system.refresh_memory();
	let available_memory = usize::from_u64(state.system.available_memory()).context("Failed to get available memory")?;

	let batch_size = if available_memory < state.memory_threshold {
		let max_points = (available_memory / POINT_SIZE).max(50);
		batch_times.len().min(max_points).max(1)
	} else {
		batch_times.len()
	};

	// Extract values before parallel processing to avoid borrowing issues
	let input_points_ref = input_points;
	let batch_times_ref = batch_times;
	let spline = state.spline;
	let resolution = state.resolution;

	// Collect chunks first, then process in parallel
	let chunks: Vec<Vec<DateTime<Utc>>> = batch_times_ref.chunks(batch_size).map(<[chrono::DateTime<chrono::Utc>]>::to_vec).collect();

	let chunk_results: Result<Vec<Vec<Point>>> = chunks
		.par_iter()
		.map(|times| match spline {
			Spline::Linear => linear_simd(input_points_ref, times, resolution),
			Spline::Quadratic => quadratic_simd(input_points_ref, times, resolution),
			Spline::Cubic => cubic_simd(input_points_ref, times, resolution),
			Spline::Polynomial(_, _) => polynomial_simd(input_points_ref, times, resolution, &spline),
		})
		.collect();

	let chunk_results = chunk_results?;

	// Flatten results in the correct order
	let all_points: Vec<Point> = chunk_results.into_iter().flatten().collect();

	// Verify we have the expected number of points
	assert_eq!(all_points.len(), batch_times_ref.len(), "Parallel processing lost points: expected {}, got {}", batch_times_ref.len(), all_points.len());

	// Check if we should use temp file or in-memory storage
	if state.temp_file.is_some() {
		// Use temp file (for large datasets)
		if let Some(writer) = state.temp_file.as_mut() {
			let mut writer = writer.lock().await;
			for point in all_points {
				writeln!(writer, "{},{}", point.timestamp.to_rfc3339(), point.value)?;
			}
			writer.flush().map_err(|e| Error::IOError(e.to_string()))?;
		}
	} else {
		// Use in-memory storage (for smaller datasets)
		state.result = Some(all_points);
	}

	Ok(())
}