Merge pull request #565 from kitsudaiki/feat/upload-csv-files

related issue: #540

Author: kitsudaiki

.github/workflows/build_test.yml

@@ -619,6 +619,7 @@ jobs:
           mkdir -p test_dir
           cp sdk_api_test.py test_dir/
           cp test_values.py test_dir/
+          cp csv_test.csv test_dir/
           cd test_dir
           /usr/bin/python3 sdk_api_test.py
 

Cargo.lock

@@ -56,29 +56,23 @@ var createMnistDatasetCmd = &cobra.Command{
 	},
 }
 
-// var createCsvDatasetCmd = &cobra.Command{
-// 	Use:   "csv -i INPUT_FILE_PATH DATASET_NAME",
-// 	Short: "Upload new csv dataset.",
-// 	Args:  cobra.ExactArgs(1),
-// 	Run: func(cmd *cobra.Command, args []string) {
-// 		token := Login()
-// 		address := os.Getenv("HANAMI_ADDRESS")
-// 		datasetName := args[0]
-// 		uuid, err := hanami_sdk.UploadCsvFiles(address, token, datasetName, inputFilePath, hanamictl_common.DisableTlsVerification)
-// 		if err != nil {
-// 			fmt.Println(err)
-// 			os.Exit(1)
-// 		}
-
-// 		content, err := hanami_sdk.GetDataset(address, token, uuid, hanamictl_common.DisableTlsVerification)
-// 		if err == nil {
-// 			hanamictl_common.PrintSingle(content)
-// 		} else {
-// 			fmt.Println(err)
-// 			os.Exit(1)
-// 		}
-// 	},
-// }
+var createCsvDatasetCmd = &cobra.Command{
+	Use:   "csv -i INPUT_FILE_PATH DATASET_NAME",
+	Short: "Upload new csv dataset.",
+	Args:  cobra.ExactArgs(1),
+	Run: func(cmd *cobra.Command, args []string) {
+		token := Login()
+		address := os.Getenv("HANAMI_ADDRESS")
+		datasetName := args[0]
+		content, err := hanami_sdk.CreateCsvDataset(address, token, datasetName, inputFilePath, hanamictl_common.DisableTlsVerification)
+		if err == nil {
+			hanamictl_common.PrintSingle(content)
+		} else {
+			fmt.Println(err)
+			os.Exit(1)
+		}
+	},
+}
 
 var checkDatasetCmd = &cobra.Command{
 	Use:   "check -r REFERENCE_DATASET_UUID DATASET_UUID",
@@ -189,9 +183,9 @@ func Init_Dataset_Commands(rootCmd *cobra.Command) {
 	createMnistDatasetCmd.MarkFlagRequired("input")
 	createMnistDatasetCmd.MarkFlagRequired("label")
 
-	// createDatasetCmd.AddCommand(createCsvDatasetCmd)
-	// createCsvDatasetCmd.Flags().StringVarP(&inputFilePath, "input", "i", "", "Path to file with input-data (mandatory)")
-	// createCsvDatasetCmd.MarkFlagRequired("input")
+	createDatasetCmd.AddCommand(createCsvDatasetCmd)
+	createCsvDatasetCmd.Flags().StringVarP(&inputFilePath, "input", "i", "", "Path to file with input-data (mandatory)")
+	createCsvDatasetCmd.MarkFlagRequired("input")
 
 	datasetCmd.AddCommand(checkDatasetCmd)
 	checkDatasetCmd.Flags().StringVarP(&referenceDatasetUuid, "reference", "r", "", "UUID of the dataset, which works as reference (mandatory)")

src/cli/hanamictl/resources/dataset.go

@@ -29,7 +29,8 @@ use crate::api::errors::ErrorResponse;
 use crate::database::dataset_table;
 use crate::config;
 
-use hanami_dataset::converter::load_mnist_images;
+use hanami_dataset::converter::{load_mnist_images, load_csv_file};
+use hanami_dataset::dataset_io::read_data_set_file;
 use hanami_common::error::HanamiError;
 
 use super::dataset_structs::DatasetResp;
@@ -175,6 +176,30 @@ pub async fn upload_binary(mut payload: Multipart, path: Path<(String, String)>,
                 }
             },
         };
+    } else if dataset_type == "csv" {
+        let path_len = temp_file_paths.len();
+        if temp_file_paths.len() != 1 {
+            let msg = format!("CSV-dataset expect 1 uploaded files, but there were {path_len} files found.");
+            return Err(ErrorResponse::BadRequest(msg));
+        }
+        match load_csv_file(
+            &temp_file_paths[0], 
+            &target_filepath,
+            dataset_uuid.clone(),
+            name.clone()) 
+        {
+            Ok(()) => {},
+            Err(e) => match e.downcast_ref::<HanamiError>() {
+                Some(HanamiError::InputError(e)) => {
+                    let msg = format!("{}", e);
+                    return Err(ErrorResponse::BadRequest(msg));
+                },
+                _ => {
+                    error!("{}", e);
+                    return Err(ErrorResponse::InternalError("".to_string()));
+                }
+            },
+        };
     }
 
     // add new dataset to datbase
@@ -207,10 +232,19 @@ pub async fn upload_binary(mut payload: Multipart, path: Path<(String, String)>,
         }
     }
 
+    let file_handle = match read_data_set_file(&target_filepath) {
+        Ok(file_handle) => file_handle,
+        Err(_) => {
+            return Err(ErrorResponse::InternalError("".to_string()));
+        }
+    };
+
     // create response
     let resp = DatasetResp {
         uuid: dataset_uuid.clone(),
         name: dataset.name.clone(),
+        number_of_rows: file_handle.get_number_of_rows(),
+        number_of_columns: file_handle.header.columns.len() as u64,
         created_by: dataset.created_by.clone(),
         created_at: dataset.created_at.clone(),
         updated_by: dataset.updated_by.clone(),

src/hanami/src/api/http_endpoints/dataset/create_dataset_v1_0.rs

@@ -27,6 +27,8 @@ pub struct DatasetCreateReq {
 pub struct DatasetResp {
     pub uuid: Uuid,
     pub name: String,
+    pub number_of_rows: u64,
+    pub number_of_columns: u64,
     pub created_at: String,
     pub created_by: String,
     pub updated_at: String,

src/hanami/src/api/http_endpoints/dataset/dataset_structs.rs

@@ -16,11 +16,14 @@ use actix_web::web::Json;
 use actix_web::web::Path;
 use apistos::api_operation;
 use uuid::Uuid;
+use std::path::PathBuf;
 
 use crate::api::errors::ErrorResponse;
 use crate::api::user_context::UserContext;
 use crate::database::dataset_table;
+
 use hanami_common::enums;
+use hanami_dataset::dataset_io::read_data_set_file;
 
 use super::dataset_structs::DatasetResp;
 
@@ -34,19 +37,8 @@ use super::dataset_structs::DatasetResp;
     error_code = 500
 )]
 pub async fn get_dataset(dataset_uuid: Path<Uuid>, context: UserContext) -> Result<Json<DatasetResp>, ErrorResponse> {
-    match dataset_table::get_dataset(&dataset_uuid, &context) {
-        Ok(dataset) => {
-            let resp = DatasetResp {
-                uuid: dataset_uuid.clone(),
-                name: dataset.name.clone(),
-                created_by: dataset.created_by.clone(),
-                created_at: dataset.created_at.clone(),
-                updated_by: dataset.updated_by.clone(),
-                updated_at: dataset.updated_at.clone(),
-            };
-        
-            return Ok(Json(resp));
-        },
+    let dataset_data = match dataset_table::get_dataset(&dataset_uuid, &context) {
+        Ok(dataset_data) => dataset_data,
         Err(enums::DbError::InternalError) => {
             return Err(ErrorResponse::InternalError("".to_string()));
         },
@@ -55,4 +47,24 @@ pub async fn get_dataset(dataset_uuid: Path<Uuid>, context: UserContext) -> Resu
             return Err(ErrorResponse::NotFound(msg));
         }
     };
+
+    let file_handle = match read_data_set_file(&PathBuf::from(dataset_data.file_path)) {
+        Ok(file_handle) => file_handle,
+        Err(_) => {
+            return Err(ErrorResponse::InternalError("".to_string()));
+        }
+    };
+
+    let resp = DatasetResp {
+        uuid: dataset_uuid.clone(),
+        name: dataset_data.name.clone(),
+        number_of_rows: file_handle.get_number_of_rows(),
+        number_of_columns: file_handle.header.columns.len() as u64,
+        created_by: dataset_data.created_by.clone(),
+        created_at: dataset_data.created_at.clone(),
+        updated_by: dataset_data.updated_by.clone(),
+        updated_at: dataset_data.updated_at.clone(),
+    };
+
+    return Ok(Json(resp));
 }

src/hanami/src/api/http_endpoints/dataset/get_dataset_v1_0.rs

@@ -8,9 +8,10 @@ hanami_common = { path = "../hanami_common" }
 # json
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
-uuid = { version = "1.17.0", features = ["v4", "serde"] }
 # common stuff
 sanitize-filename = "0.6"
 byteorder = "1.5"
 bincode = "2.0"
 bytemuck = "1.22"
+csv = "1.3"
+uuid = { version = "1.17.0", features = ["v4", "serde"] }

src/libs/rust/hanami_dataset/Cargo.toml

@@ -15,11 +15,11 @@
 use std::error::Error;
 use std::path::PathBuf;
 use std::fs::File;
-use std::io::{BufReader, Read};
+use std::io::{BufReader, Write, Read};
 use std::collections::HashMap;
 use byteorder::{ReadBytesExt, BigEndian};
 use uuid::Uuid;
-use std::io::Write;
+use csv::ReaderBuilder;
 
 use super::dataset_io::*;
 
@@ -140,3 +140,56 @@ pub fn load_mnist_images(
     Ok(())
 }
 
+pub fn load_csv_file(
+    file_path: &PathBuf,
+    target_filepath: &PathBuf,
+    uuid: Uuid,
+    name: String,
+) -> Result<(), Box<dyn Error>> {
+    let file = File::open(file_path)?;
+    let mut rdr = ReaderBuilder::new().has_headers(true).from_reader(file);
+
+    // get number of columns from header
+    let headers = rdr.headers()?;
+    let num_columns = headers.len();
+
+    // get column-names from header
+    let mut columns: HashMap<String, Column> = HashMap::new();
+    for (i, name) in headers.iter().enumerate() {
+        let col = Column {
+            start: i as u64,
+            end: i as u64 + 1,
+        };
+        columns.insert(name.to_string(), col);
+    }
+
+    // init dataset
+    let mut dataset_handle = init_new_data_set_file(
+        &target_filepath, 
+        uuid,
+        name, 
+        "".to_string(),
+        num_columns as u64,
+        columns,
+        DataSetType::FloatType)?; // TODO: use u8-type
+
+    // read body into the dataset-file
+    for result in rdr.records() {
+        let record = result?;
+
+        let row = record.iter()
+            .map(|field| field.parse::<f32>().unwrap_or(0.0))
+            .collect::<Vec<f32>>();
+
+        let row_bytes = unsafe {
+            std::slice::from_raw_parts(
+                row.as_ptr() as *const u8,
+                row.len() * std::mem::size_of::<f32>(),
+            )
+        };
+
+        dataset_handle.target_file.write_all(&row_bytes)?;
+    }
+
+    Ok(())
+}

src/libs/rust/hanami_dataset/src/converter.rs

@@ -32,6 +32,12 @@ func CreateMnistDataset(address, token, datasetName, imageFilePath, labelFilePat
 	return UploadFiles(address, token, path, files, skipTlsVerification)
 }
 
+func CreateCsvDataset(address, token, datasetName, filePath string, skipTlsVerification bool) (map[string]interface{}, error) {
+	path := fmt.Sprintf("v1alpha/dataset/csv/%s", datasetName)
+	files := []string{filePath}
+	return UploadFiles(address, token, path, files, skipTlsVerification)
+}
+
 func GetDataset(address, token, datasetUuid string, skipTlsVerification bool) (map[string]interface{}, error) {
 	path := fmt.Sprintf("v1alpha/dataset/%s", datasetUuid)
 	vars := map[string]interface{}{}

src/sdk/go/hanami_sdk/dataset.go

@@ -104,3 +104,18 @@ def upload_mnist_files(token: str,
                                        path,
                                        files,
                                        verify=verify_connection)
+
+
+def upload_csv_files(token: str,
+                     address: str,
+                     name: str,
+                     input_file_path: str,
+                     verify_connection: bool = True) -> str:
+    path = f"/v1alpha/dataset/csv/{name}"
+    files = [input_file_path]
+
+    return hanami_request.upload_files(token,
+                                       address,
+                                       path,
+                                       files,
+                                       verify=verify_connection)