Refactor: Move DataServer to its own module
Moves the `DataServer` struct and its implementation into a new `dataserver` module. This improves code organization and separates data loading logic from the main application logic.
This commit is contained in:
Michael Schimmel
@@ -0,0 +1,166 @@
|
||||
use bytemuck::cast_slice;
|
||||
use std::collections::HashMap;
|
||||
use std::io::Read;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::Instant;
|
||||
use zip::ZipArchive;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct DataFile {
|
||||
pub path: PathBuf,
|
||||
pub symbol: String,
|
||||
pub year: i32,
|
||||
pub month: i32,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct CacheEntry<T> {
|
||||
pub data: Arc<Vec<T>>,
|
||||
pub last_accessed: Instant,
|
||||
}
|
||||
|
||||
pub struct DataServer<T: bytemuck::Pod + Send + Sync + 'static> {
|
||||
base_path: PathBuf,
|
||||
pub symbols: RwLock<HashMap<String, Vec<DataFile>>>,
|
||||
pub cache: RwLock<HashMap<PathBuf, CacheEntry<T>>>,
|
||||
}
|
||||
|
||||
impl<T: bytemuck::Pod + Send + Sync + 'static> DataServer<T> {
|
||||
pub fn new(path: impl AsRef<Path>) -> Self {
|
||||
Self {
|
||||
base_path: path.as_ref().to_path_buf(),
|
||||
symbols: RwLock::new(HashMap::new()),
|
||||
cache: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn update_symbols(&self) -> tokio::io::Result<()> {
|
||||
let mut entries = tokio::fs::read_dir(&self.base_path).await?;
|
||||
let mut temp_map: HashMap<String, Vec<DataFile>> = HashMap::new();
|
||||
|
||||
while let Some(entry) = entries.next_entry().await? {
|
||||
let path = entry.path();
|
||||
if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) {
|
||||
if file_name.ends_with(".m1") || file_name.ends_with(".tick") {
|
||||
if let Some(data_file) = self.parse_file_name(&path, file_name) {
|
||||
temp_map
|
||||
.entry(data_file.symbol.clone())
|
||||
.or_default()
|
||||
.push(data_file);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for files in temp_map.values_mut() {
|
||||
files.sort_by(|a, b| a.year.cmp(&b.year).then(a.month.cmp(&b.month)));
|
||||
}
|
||||
|
||||
let mut symbols = self.symbols.write().unwrap();
|
||||
*symbols = temp_map;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn enumerate_symbols(&self) -> Vec<String> {
|
||||
let symbols = self.symbols.read().unwrap();
|
||||
let mut keys: Vec<String> = symbols.keys().cloned().collect();
|
||||
keys.sort();
|
||||
keys
|
||||
}
|
||||
|
||||
fn parse_file_name(&self, path: &Path, file_name: &str) -> Option<DataFile> {
|
||||
let stem = Path::new(file_name).file_stem()?.to_str()?;
|
||||
let ext = Path::new(file_name).extension()?.to_str()?;
|
||||
let parts: Vec<&str> = stem.split('_').collect();
|
||||
if parts.len() < 3 {
|
||||
return None;
|
||||
}
|
||||
let month = parts.last()?.parse::<i32>().ok()?;
|
||||
let year = parts[parts.len() - 2].parse::<i32>().ok()?;
|
||||
let symbol = format!("{}.{}", parts[..parts.len() - 2].join("_"), ext);
|
||||
Some(DataFile {
|
||||
path: path.to_path_buf(),
|
||||
symbol,
|
||||
year,
|
||||
month,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn load_data<R>(&self, data_file: &DataFile) -> anyhow::Result<Arc<Vec<T>>>
|
||||
where
|
||||
R: crate::records::SourceRecord,
|
||||
crate::records::DataPoint<R::Mapped>: Into<T>,
|
||||
{
|
||||
{
|
||||
let mut cache = self.cache.write().unwrap();
|
||||
if let Some(entry) = cache.get_mut(&data_file.path) {
|
||||
entry.last_accessed = Instant::now();
|
||||
return Ok(Arc::clone(&entry.data));
|
||||
}
|
||||
}
|
||||
|
||||
let path = data_file.path.clone();
|
||||
let shared_data = tokio::task::spawn_blocking(move || {
|
||||
let file = std::fs::File::open(&path)?;
|
||||
let mut archive = ZipArchive::new(file)?;
|
||||
for i in 0..archive.len() {
|
||||
let mut file = archive.by_index(i)?;
|
||||
if file.name().ends_with(".bin") {
|
||||
let mut bin_content = vec![0u8; file.size() as usize];
|
||||
file.read_exact(&mut bin_content)?;
|
||||
let raw_data: &[R] = cast_slice(&bin_content);
|
||||
|
||||
// Perform mapping inside blocking task
|
||||
let mapped_data: Vec<T> =
|
||||
raw_data.iter().map(|&r| r.to_data_point().into()).collect();
|
||||
|
||||
return Ok(Arc::new(mapped_data));
|
||||
}
|
||||
}
|
||||
Err(anyhow::anyhow!("No .bin file found"))
|
||||
})
|
||||
.await??;
|
||||
|
||||
let mut cache = self.cache.write().unwrap();
|
||||
cache.insert(
|
||||
data_file.path.clone(),
|
||||
CacheEntry {
|
||||
data: Arc::clone(&shared_data),
|
||||
last_accessed: Instant::now(),
|
||||
},
|
||||
);
|
||||
Ok(shared_data)
|
||||
}
|
||||
|
||||
// async fn read_from_zip - removed as it was unused in main.rs context or seemingly redundant/internal helper not used in extracted public interface?
|
||||
// Checking main.rs: It was NOT used in main.rs, it was a private helper method inside impl DataServer.
|
||||
// However, it wasn't used by load_data either. It seems dead code or future code.
|
||||
// I will include it to be safe, but make it private or allow dead code if it warns.
|
||||
// Actually, looking at main.rs again, read_from_zip was defined but NOT used in the provided snippet of DataServer.
|
||||
// Wait, load_data implements similar logic inline.
|
||||
// I will keep it to strictly follow "never change code unless..." (preserving the code structure).
|
||||
|
||||
#[allow(dead_code)]
|
||||
async fn read_from_zip<R: bytemuck::Pod + Send + 'static>(
|
||||
&self,
|
||||
path: &Path,
|
||||
) -> anyhow::Result<Vec<R>> {
|
||||
let path = path.to_path_buf();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let file = std::fs::File::open(&path)?;
|
||||
let mut archive = ZipArchive::new(file)?;
|
||||
for i in 0..archive.len() {
|
||||
let mut file = archive.by_index(i)?;
|
||||
if file.name().ends_with(".bin") {
|
||||
let mut bin_content = vec![0u8; file.size() as usize];
|
||||
file.read_exact(&mut bin_content)?;
|
||||
let records: &[R] = cast_slice(&bin_content);
|
||||
return Ok(records.to_vec());
|
||||
}
|
||||
}
|
||||
Err(anyhow::anyhow!("No .bin file found"))
|
||||
})
|
||||
.await?
|
||||
}
|
||||
}
|
||||
+6
-161
@@ -1,7 +1,12 @@
|
||||
mod dataserver;
|
||||
mod records;
|
||||
|
||||
use crate::dataserver::DataServer;
|
||||
use crate::records::{DataPoint, M1Record, OhlcItem, TickRecord};
|
||||
use chrono::{DateTime, Utc};
|
||||
use chrono::DateTime;
|
||||
use eframe::egui;
|
||||
use std::sync::Arc;
|
||||
use tokio::runtime::Runtime;
|
||||
|
||||
fn format_ts(ts: f64) -> String {
|
||||
// Delphi TDateTime: Days since 1899-12-30.
|
||||
@@ -16,166 +21,6 @@ fn format_ts(ts: f64) -> String {
|
||||
format!("{:.6}", ts)
|
||||
}
|
||||
}
|
||||
use bytemuck::cast_slice;
|
||||
use eframe::egui;
|
||||
use std::collections::HashMap;
|
||||
use std::io::Read;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::Instant;
|
||||
use tokio::runtime::Runtime;
|
||||
use zip::ZipArchive;
|
||||
|
||||
// --- Data Structures ---
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct DataFile {
|
||||
pub path: PathBuf,
|
||||
pub symbol: String,
|
||||
pub year: i32,
|
||||
pub month: i32,
|
||||
}
|
||||
|
||||
struct CacheEntry<T> {
|
||||
data: Arc<Vec<T>>,
|
||||
last_accessed: Instant,
|
||||
}
|
||||
|
||||
pub struct DataServer<T: bytemuck::Pod + Send + Sync + 'static> {
|
||||
base_path: PathBuf,
|
||||
symbols: RwLock<HashMap<String, Vec<DataFile>>>,
|
||||
cache: RwLock<HashMap<PathBuf, CacheEntry<T>>>,
|
||||
}
|
||||
|
||||
impl<T: bytemuck::Pod + Send + Sync + 'static> DataServer<T> {
|
||||
pub fn new(path: impl AsRef<Path>) -> Self {
|
||||
Self {
|
||||
base_path: path.as_ref().to_path_buf(),
|
||||
symbols: RwLock::new(HashMap::new()),
|
||||
cache: RwLock::new(HashMap::new()),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn update_symbols(&self) -> tokio::io::Result<()> {
|
||||
let mut entries = tokio::fs::read_dir(&self.base_path).await?;
|
||||
let mut temp_map: HashMap<String, Vec<DataFile>> = HashMap::new();
|
||||
|
||||
while let Some(entry) = entries.next_entry().await? {
|
||||
let path = entry.path();
|
||||
if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) {
|
||||
if file_name.ends_with(".m1") || file_name.ends_with(".tick") {
|
||||
if let Some(data_file) = self.parse_file_name(&path, file_name) {
|
||||
temp_map
|
||||
.entry(data_file.symbol.clone())
|
||||
.or_default()
|
||||
.push(data_file);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for files in temp_map.values_mut() {
|
||||
files.sort_by(|a, b| a.year.cmp(&b.year).then(a.month.cmp(&b.month)));
|
||||
}
|
||||
|
||||
let mut symbols = self.symbols.write().unwrap();
|
||||
*symbols = temp_map;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn enumerate_symbols(&self) -> Vec<String> {
|
||||
let symbols = self.symbols.read().unwrap();
|
||||
let mut keys: Vec<String> = symbols.keys().cloned().collect();
|
||||
keys.sort();
|
||||
keys
|
||||
}
|
||||
|
||||
fn parse_file_name(&self, path: &Path, file_name: &str) -> Option<DataFile> {
|
||||
let stem = Path::new(file_name).file_stem()?.to_str()?;
|
||||
let ext = Path::new(file_name).extension()?.to_str()?;
|
||||
let parts: Vec<&str> = stem.split('_').collect();
|
||||
if parts.len() < 3 {
|
||||
return None;
|
||||
}
|
||||
let month = parts.last()?.parse::<i32>().ok()?;
|
||||
let year = parts[parts.len() - 2].parse::<i32>().ok()?;
|
||||
let symbol = format!("{}.{}", parts[..parts.len() - 2].join("_"), ext);
|
||||
Some(DataFile {
|
||||
path: path.to_path_buf(),
|
||||
symbol,
|
||||
year,
|
||||
month,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn load_data<R>(&self, data_file: &DataFile) -> anyhow::Result<Arc<Vec<T>>>
|
||||
where
|
||||
R: crate::records::SourceRecord,
|
||||
crate::records::DataPoint<R::Mapped>: Into<T>,
|
||||
{
|
||||
{
|
||||
let mut cache = self.cache.write().unwrap();
|
||||
if let Some(entry) = cache.get_mut(&data_file.path) {
|
||||
entry.last_accessed = Instant::now();
|
||||
return Ok(Arc::clone(&entry.data));
|
||||
}
|
||||
}
|
||||
|
||||
let path = data_file.path.clone();
|
||||
let shared_data = tokio::task::spawn_blocking(move || {
|
||||
let file = std::fs::File::open(&path)?;
|
||||
let mut archive = ZipArchive::new(file)?;
|
||||
for i in 0..archive.len() {
|
||||
let mut file = archive.by_index(i)?;
|
||||
if file.name().ends_with(".bin") {
|
||||
let mut bin_content = vec![0u8; file.size() as usize];
|
||||
file.read_exact(&mut bin_content)?;
|
||||
let raw_data: &[R] = cast_slice(&bin_content);
|
||||
|
||||
// Perform mapping inside blocking task
|
||||
let mapped_data: Vec<T> =
|
||||
raw_data.iter().map(|&r| r.to_data_point().into()).collect();
|
||||
|
||||
return Ok(Arc::new(mapped_data));
|
||||
}
|
||||
}
|
||||
Err(anyhow::anyhow!("No .bin file found"))
|
||||
})
|
||||
.await??;
|
||||
|
||||
let mut cache = self.cache.write().unwrap();
|
||||
cache.insert(
|
||||
data_file.path.clone(),
|
||||
CacheEntry {
|
||||
data: Arc::clone(&shared_data),
|
||||
last_accessed: Instant::now(),
|
||||
},
|
||||
);
|
||||
Ok(shared_data)
|
||||
}
|
||||
|
||||
async fn read_from_zip<R: bytemuck::Pod + Send + 'static>(
|
||||
&self,
|
||||
path: &Path,
|
||||
) -> anyhow::Result<Vec<R>> {
|
||||
let path = path.to_path_buf();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let file = std::fs::File::open(&path)?;
|
||||
let mut archive = ZipArchive::new(file)?;
|
||||
for i in 0..archive.len() {
|
||||
let mut file = archive.by_index(i)?;
|
||||
if file.name().ends_with(".bin") {
|
||||
let mut bin_content = vec![0u8; file.size() as usize];
|
||||
file.read_exact(&mut bin_content)?;
|
||||
let records: &[R] = cast_slice(&bin_content);
|
||||
return Ok(records.to_vec());
|
||||
}
|
||||
}
|
||||
Err(anyhow::anyhow!("No .bin file found"))
|
||||
})
|
||||
.await?
|
||||
}
|
||||
}
|
||||
|
||||
// --- EGui App ---
|
||||
|
||||
|
||||
Reference in New Issue
Block a user