add empty chunk handling, importing python crate dependencies, tx filtering by address, support for new datatypes in python crate

crates/cli/src/args.rs

@@ -195,6 +195,10 @@ pub struct Args {
     /// Avoid saving a summary report
     #[arg(long, help_heading = "Output Options")]
     pub no_report: bool,
+
+    /// Skip if writing if dataframe is empty
+    #[arg(long, help_heading = "Acquisition Options")]
+    pub write_empty: Option<bool>,
 
     /// Address(es)
     #[arg(long, help_heading = "Dataset-specific Options", num_args(1..))]

crates/cli/src/parse/execution.rs

@@ -17,6 +17,7 @@ pub(crate) fn parse_execution_env(args: &Args, n_tasks: u64) -> Result<Execution
         .verbose(verbose)
         .report(!args.no_report)
         .report_dir(args.report_dir.clone())
+        .write_empty(args.write_empty)
         .args(args_str);
 
     let builder = if !args.no_verbose {

crates/freeze/src/datasets/address_appearances.rs

@@ -331,4 +331,4 @@ fn process_appearances(
     }
 
     Ok(())
-}
+}

crates/freeze/src/datasets/contracts.rs

@@ -98,4 +98,4 @@ pub(crate) fn process_contracts(
         }
     }
     Ok(())
-}
+}

crates/freeze/src/datasets/erc20_transfers.rs

@@ -8,7 +8,6 @@ use polars::prelude::*;
 pub struct Erc20Transfers {
     n_rows: u64,
     block_number: Vec<u32>,
-    block_hash: Vec<Option<Vec<u8>>>,
     transaction_index: Vec<u32>,
     log_index: Vec<u32>,
     transaction_hash: Vec<Vec<u8>>,
@@ -21,21 +20,6 @@ pub struct Erc20Transfers {
 
 #[async_trait::async_trait]
 impl Dataset for Erc20Transfers {
-    fn default_columns() -> Option<Vec<&'static str>> {
-        Some(vec![
-            "block_number",
-            // "block_hash",
-            "transaction_index",
-            "log_index",
-            "transaction_hash",
-            "erc20",
-            "from_address",
-            "to_address",
-            "value",
-            "chain_id",
-        ])
-    }
-
     fn optional_parameters() -> Vec<Dim> {
         vec![Dim::Address, Dim::Topic0, Dim::Topic1, Dim::Topic2, Dim::FromAddress, Dim::ToAddress]
     }
@@ -104,7 +88,6 @@ fn process_erc20_transfers(logs: Vec<Log>, columns: &mut Erc20Transfers, schema:
         {
             columns.n_rows += 1;
             store!(schema, columns, block_number, bn.as_u32());
-            store!(schema, columns, block_hash, log.block_hash.map(|bh| bh.as_bytes().to_vec()));
             store!(schema, columns, transaction_index, ti.as_u32());
             store!(schema, columns, log_index, li.as_u32());
             store!(schema, columns, transaction_hash, tx.as_bytes().to_vec());

crates/freeze/src/datasets/erc721_transfers.rs

@@ -120,4 +120,4 @@ fn process_erc721_transfers(
         }
     }
     Ok(())
-}
+}

crates/freeze/src/datasets/logs.rs

@@ -149,4 +149,4 @@ fn process_logs(logs: Vec<Log>, columns: &mut Logs, schema: &Table) -> R<()> {
     }
 
     Ok(())
-}
+}

crates/freeze/src/datasets/native_transfers.rs

@@ -106,4 +106,4 @@ pub(crate) fn process_native_transfers(
         }
     }
     Ok(())
-}
+}

crates/freeze/src/datasets/transactions.rs

@@ -1,6 +1,14 @@
 use crate::*;
 use ethers::prelude::*;
 use polars::prelude::*;
+use std::collections::HashSet;
+use std::sync::Mutex;
+use lazy_static::lazy_static;
+
+lazy_static! {
+    static ref TRANSACTIONS_SET: Mutex<HashSet<Vec<u8>>> = Mutex::new(HashSet::new());
+}
+
 
 /// columns for transactions
 #[cryo_to_df::to_df(Datatype::Transactions)]
@@ -11,6 +19,7 @@ pub struct Transactions {
     transaction_index: Vec<Option<u64>>,
     transaction_hash: Vec<Vec<u8>>,
     nonce: Vec<u64>,
+    address: Vec<Vec<u8>>,
     from_address: Vec<Vec<u8>>,
     to_address: Vec<Option<Vec<u8>>>,
     value: Vec<U256>,
@@ -55,7 +64,8 @@ impl Dataset for Transactions {
     }
 
     fn optional_parameters() -> Vec<Dim> {
-        vec![Dim::FromAddress, Dim::ToAddress]
+        vec![Dim::Address,Dim::FromAddress, Dim::ToAddress]
+        //vec![Dim::FromAddress, Dim::ToAddress]
     }
 }
 
@@ -65,8 +75,10 @@ pub type TransactionAndReceipt = (Transaction, Option<TransactionReceipt>);
 #[async_trait::async_trait]
 impl CollectByBlock for Transactions {
     type Response = (Block<Transaction>, Vec<TransactionAndReceipt>, bool);
+    //println!("impl");
 
     async fn extract(request: Params, source: Arc<Source>, query: Arc<Query>) -> R<Self::Response> {
+
         let block = source
             .fetcher
             .get_block_with_txs(request.block_number()?)
@@ -77,8 +89,10 @@ impl CollectByBlock for Transactions {
         // 1. collect transactions and filter them if optional parameters are supplied
         // filter by from_address
         let from_filter: Box<dyn Fn(&Transaction) -> bool + Send> =
-            if let Some(from_address) = &request.from_address {
-                Box::new(move |tx| tx.from.as_bytes() == from_address)
+            if let Some(from_address) = &request.from_address {            
+                Box::new(move |tx| {
+                    from_address == tx.from.as_bytes()
+                })
             } else {
                 Box::new(|_| true)
             };
@@ -89,16 +103,37 @@ impl CollectByBlock for Transactions {
             } else {
                 Box::new(|_| true)
             };
+
+        // filter by address
+        let addr_filter: Box<dyn Fn(&Transaction) -> bool + Send> =
+        if let Some(address) = &request.address {
+            Box::new(move |tx| {
+                let mut transactions_set = TRANSACTIONS_SET.lock().unwrap();
+                if transactions_set.contains(&tx.hash.as_bytes().to_vec()) {
+                    false
+                } else {
+                    let condition = tx.to.as_ref().map_or(false, |x| x.as_bytes() == address) ||
+                                    tx.from.as_bytes() == address;
+                    if condition {
+                        transactions_set.insert(tx.hash.as_bytes().to_vec());
+                    }
+                    condition
+                }
+            })
+        } else {
+            Box::new(|_| true)
+        };
+
         let transactions =
-            block.transactions.clone().into_iter().filter(from_filter).filter(to_filter).collect();
+            block.transactions.clone().into_iter().filter(from_filter).filter(to_filter).filter(addr_filter).collect();
 
         // 2. collect receipts if necessary
         // if transactions are filtered fetch by set of transaction hashes, else fetch all receipts
         // in block
         let receipts: Vec<Option<_>> =
             if schema.has_column("gas_used") | schema.has_column("success") {
                 // receipts required
-                let receipts = if request.from_address.is_some() || request.to_address.is_some() {
+                let receipts = if request.from_address.is_some() || request.to_address.is_some() || request.address.is_some() {
                     source.get_tx_receipts(&transactions).await?
                 } else {
                     source.get_tx_receipts_in_block(&block).await?
@@ -108,8 +143,8 @@ impl CollectByBlock for Transactions {
                 vec![None; block.transactions.len()]
             };
 
-        let transactions_with_receips = transactions.into_iter().zip(receipts).collect();
-        Ok((block, transactions_with_receips, query.exclude_failed))
+        let transactions_with_receipts = transactions.into_iter().zip(receipts).collect();
+        Ok((block, transactions_with_receipts, query.exclude_failed))
     }
 
     fn transform(response: Self::Response, columns: &mut Self, query: &Arc<Query>) -> R<()> {

crates/freeze/src/freeze.rs

@@ -178,6 +178,11 @@ async fn freeze_partition(payload: PartitionPayload) -> Result<(), CollectError>
 
     // write dataframes to disk
     for (datatype, mut df) in dfs {
+        if Some(env.write_empty)==Some(Some(false)){
+            if df.shape().0 == 0 {
+                continue;
+            };
+        };
         let path = paths.get(&datatype).ok_or_else(|| {
             CollectError::CollectError("could not get path for datatype".to_string())
         })?;

crates/freeze/src/types/execution.rs

@@ -25,6 +25,8 @@ pub struct ExecutionEnv {
     pub t_end: Option<SystemTime>,
     /// report directory
     pub report_dir: Option<PathBuf>,
+    /// skips writing empty dfs
+    pub write_empty: Option<bool>,
 }
 
 impl ExecutionEnv {
@@ -67,6 +69,7 @@ pub struct ExecutionEnvBuilder {
     t_start: SystemTime,
     t_end: Option<SystemTime>,
     report_dir: Option<PathBuf>,
+    write_empty: Option<bool>,
 }
 
 impl Default for ExecutionEnvBuilder {
@@ -82,6 +85,7 @@ impl Default for ExecutionEnvBuilder {
             t_start: SystemTime::now(),
             t_end: None,
             report_dir: None,
+            write_empty: Some(true)
         }
     }
 }
@@ -134,6 +138,11 @@ impl ExecutionEnvBuilder {
         self
     }
 
+    /// write_empty
+    pub fn write_empty(mut self, write_empty: Option<bool>) -> Self {
+        self.write_empty = write_empty;
+        self
+    }
     /// build final output
     pub fn build(self) -> ExecutionEnv {
         ExecutionEnv {
@@ -147,6 +156,7 @@ impl ExecutionEnvBuilder {
             t_start: self.t_start,
             t_end: self.t_end,
             report_dir: self.report_dir,
+            write_empty: self.write_empty,
         }
     }
 }

crates/python/pyproject.toml

@@ -5,6 +5,7 @@ build-backend = "maturin"
 [project]
 name = "cryo_python"
 requires-python = ">=3.7"
+dependencies = ["maturin>=1.1,<2.0","pandas==2.1.3","polars==0.19.19","pyarrow==14.0.1"]
 classifiers = [
     "Programming Language :: Rust",
     "Programming Language :: Python :: Implementation :: CPython",

crates/python/python/cryo/_args.py

@@ -37,7 +37,8 @@ def parse_cli_args(
     else:
         raise Exception('unknown file_format')
 
-    kwargs['no_verbose'] = not verbose
-
+    if 'no_verbose' not in kwargs.keys():
+        kwargs['no_verbose'] = not verbose
+
     return kwargs
 

crates/python/python/cryo/_spec.py

@@ -54,4 +54,14 @@ class CryoCliArgs(TypedDict, total=False):
         topic3: str | bytes | None
         inner_request_size: int | None
         no_verbose: bool
+        address: typing.Sequence[str] | None
+        to_address: typing.Sequence[str] | None
+        from_address: typing.Sequence[str] | None
+        call_data: typing.Sequence[str] | None
+        function: typing.Sequence[str] | None
+        inputs: typing.Sequence[str] | None
+        slot: typing.Sequence[str] | None
+        event_signature: str | bytes | None
+        inner_request_size: int | None
+        js_tracer: str | bytes | None
 

crates/python/rust/collect_adapter.rs

@@ -64,6 +64,7 @@ use cryo_freeze::collect;
         verbose = false,
         no_verbose = false,
         event_signature = None,
+        write_empty = false
     )
 )]
 #[allow(clippy::too_many_arguments)]
@@ -125,6 +126,7 @@ pub fn _collect(
     verbose: bool,
     no_verbose: bool,
     event_signature: Option<String>,
+    write_empty: Option<bool>
 ) -> PyResult<&PyAny> {
     if let Some(command) = command {
         pyo3_asyncio::tokio::future_into_py(py, async move {
@@ -190,6 +192,7 @@ pub fn _collect(
             verbose,
             no_verbose,
             event_signature,
+            write_empty,
         };
         pyo3_asyncio::tokio::future_into_py(py, async move {
             match run_collect(args).await {

crates/python/rust/freeze_adapter.rs

@@ -61,6 +61,7 @@ use cryo_cli::{run, Args};
         verbose = false,
         no_verbose = false,
         event_signature = None,
+        write_empty = None,
     )
 )]
 #[allow(clippy::too_many_arguments)]
@@ -122,6 +123,7 @@ pub fn _freeze(
     verbose: bool,
     no_verbose: bool,
     event_signature: Option<String>,
+    write_empty: Option<bool>
 ) -> PyResult<&PyAny> {
     if let Some(command) = command {
         freeze_command(py, command)
@@ -182,6 +184,7 @@ pub fn _freeze(
             verbose,
             no_verbose,
             event_signature,
+            write_empty,
         };
 
         pyo3_asyncio::tokio::future_into_py(py, async move {