hdp_cli/
common.rs

1use alloy::{primitives::ChainId, transports::http::reqwest::Url};
2use anyhow::Result;
3use hdp::preprocessor::{
4    compile::config::CompilerConfig, module_registry::ModuleRegistry, PreProcessor,
5};
6use hdp::primitives::{
7    aggregate_fn::{AggregationFunction, FunctionContext},
8    processed_types::cairo_format::AsCairoFormat,
9    task::{
10        datalake::{
11            block_sampled::BlockSampledDatalake, compute::Computation, envelope::DatalakeEnvelope,
12            transactions::TransactionsInBlockDatalake, DatalakeCompute,
13        },
14        TaskEnvelope,
15    },
16};
17use std::{env, fs, path::PathBuf};
18use tracing_subscriber::{EnvFilter, FmtSubscriber};
19
20use clap::Parser;
21use hdp::processor::Processor;
22
23use tracing::{debug, info};
24
25use crate::{
26    commands::{DataLakeCommands, HDPCli, HDPCliCommands},
27    config::Config,
28    interactive,
29    query::{SubmitBatchQuery, Task},
30};
31
32pub async fn run() -> anyhow::Result<()> {
33    let start_run = std::time::Instant::now();
34    let cli = init_cli()?;
35    match cli.command {
36        HDPCliCommands::Start => {
37            interactive::run_interactive().await?;
38        }
39        HDPCliCommands::RunDatalake {
40            rpc_url,
41            chain_id,
42            preprocessor_output_file,
43            sound_run_cairo_file,
44            output_file,
45            cairo_pie_file,
46            aggregate_fn_id,
47            aggregate_fn_ctx,
48            datalake,
49        } => {
50            datalake_entry_run(
51                aggregate_fn_id,
52                aggregate_fn_ctx,
53                datalake,
54                rpc_url,
55                chain_id,
56                preprocessor_output_file,
57                sound_run_cairo_file,
58                output_file,
59                cairo_pie_file,
60            )
61            .await?
62        }
63
64        HDPCliCommands::RunModule {
65            program_hash,
66            local_class_path,
67            save_fetch_keys_file,
68            module_inputs,
69            rpc_url,
70            chain_id,
71            dry_run_cairo_file,
72            preprocessor_output_file,
73            sound_run_cairo_file,
74            output_file,
75            cairo_pie_file,
76        } => {
77            module_entry_run(
78                program_hash,
79                local_class_path,
80                save_fetch_keys_file,
81                module_inputs,
82                rpc_url,
83                chain_id,
84                dry_run_cairo_file,
85                preprocessor_output_file,
86                sound_run_cairo_file,
87                output_file,
88                cairo_pie_file,
89            )
90            .await?;
91        }
92        HDPCliCommands::Run {
93            request_file,
94            rpc_url,
95            dry_run_cairo_file,
96            preprocessor_output_file,
97            sound_run_cairo_file,
98            output_file,
99            cairo_pie_file,
100        } => {
101            entry_run(
102                request_file,
103                rpc_url,
104                dry_run_cairo_file,
105                preprocessor_output_file,
106                sound_run_cairo_file,
107                output_file,
108                cairo_pie_file,
109            )
110            .await?;
111        }
112    }
113    let duration_run = start_run.elapsed();
114    info!("HDP Cli Finished in: {:?}", duration_run);
115    Ok(())
116}
117
118/// Initialize the CLI
119fn init_cli() -> Result<HDPCli> {
120    let rust_log = env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string());
121    let subscriber = FmtSubscriber::builder()
122        .with_env_filter(EnvFilter::new(&rust_log))
123        .finish();
124    tracing::subscriber::set_global_default(subscriber).expect("Failed to set subscriber");
125    debug!("running on log level: {}", rust_log);
126    let cli = HDPCli::parse();
127    dotenv::dotenv().ok();
128    Ok(cli)
129}
130
131#[allow(clippy::too_many_arguments)]
132pub async fn module_entry_run(
133    class_hash: Option<String>,
134    local_class_path: Option<PathBuf>,
135    save_fetch_keys_file: Option<PathBuf>,
136    module_inputs: Vec<String>,
137    rpc_url: Option<Url>,
138    chain_id: Option<ChainId>,
139    dry_run_cairo_file: Option<PathBuf>,
140    preprocessor_output_file: Option<PathBuf>,
141    sound_run_cairo_file: Option<PathBuf>,
142    output_file: Option<PathBuf>,
143    cairo_pie_file: Option<PathBuf>,
144) -> Result<()> {
145    let config = Config::init(
146        rpc_url,
147        chain_id,
148        dry_run_cairo_file,
149        sound_run_cairo_file,
150        save_fetch_keys_file,
151    )
152    .await;
153    let module_registry = ModuleRegistry::new();
154    let module = module_registry
155        .get_extended_module_from_class_source_string(class_hash, local_class_path, module_inputs)
156        .await?;
157    // TODO: for now, we only support one task if its a module
158    let tasks = vec![TaskEnvelope::Module(module)];
159
160    handle_running_tasks(
161        config,
162        tasks,
163        preprocessor_output_file,
164        output_file,
165        cairo_pie_file,
166    )
167    .await?;
168    Ok(())
169}
170
171#[allow(clippy::too_many_arguments)]
172pub async fn datalake_entry_run(
173    aggregate_fn_id: AggregationFunction,
174    aggregate_fn_ctx: Option<FunctionContext>,
175    datalake: DataLakeCommands,
176    rpc_url: Option<Url>,
177    chain_id: Option<ChainId>,
178    pre_processor_output: Option<PathBuf>,
179    sound_run_cairo_file: Option<PathBuf>,
180    output_file: Option<PathBuf>,
181    cairo_pie_file: Option<PathBuf>,
182) -> Result<()> {
183    let config = Config::init(rpc_url, chain_id, None, sound_run_cairo_file, None).await;
184    let parsed_datalake = match datalake {
185        DataLakeCommands::BlockSampled {
186            block_range_start,
187            block_range_end,
188            sampled_property,
189            increment,
190        } => DatalakeEnvelope::BlockSampled(BlockSampledDatalake::new(
191            11155111,
192            block_range_start,
193            block_range_end,
194            increment,
195            sampled_property,
196        )),
197        DataLakeCommands::TransactionsInBlock {
198            target_block,
199            sampled_property,
200            start_index,
201            end_index,
202            increment,
203            included_types,
204        } => DatalakeEnvelope::TransactionsInBlock(TransactionsInBlockDatalake::new(
205            11155111,
206            target_block,
207            sampled_property,
208            start_index,
209            end_index,
210            increment,
211            included_types,
212        )),
213    };
214    let tasks = vec![TaskEnvelope::DatalakeCompute(DatalakeCompute::new(
215        parsed_datalake,
216        Computation::new(aggregate_fn_id, aggregate_fn_ctx),
217    ))];
218
219    handle_running_tasks(
220        config,
221        tasks,
222        pre_processor_output,
223        output_file,
224        cairo_pie_file,
225    )
226    .await?;
227    Ok(())
228}
229
230pub async fn handle_running_tasks(
231    config: &Config,
232    tasks: Vec<TaskEnvelope>,
233    pre_processor_output_file: Option<PathBuf>,
234    output_file: Option<PathBuf>,
235    cairo_pie_file: Option<PathBuf>,
236) -> Result<()> {
237    let compiler_config = CompilerConfig {
238        dry_run_program_path: config.dry_run_program_path.clone(),
239        provider_config: config.evm_provider.clone(),
240        save_fetch_keys_file: config.save_fetch_keys_file.clone(),
241    };
242    let preprocessor = PreProcessor::new_with_config(compiler_config);
243    let preprocessor_result = preprocessor.process(tasks).await?;
244
245    if pre_processor_output_file.is_none() {
246        info!("Finished pre processing the data");
247        Ok(())
248    } else {
249        let input_string = serde_json::to_string_pretty(&preprocessor_result.as_cairo_format())
250            .map_err(|e| anyhow::anyhow!("Failed to serialize preprocessor result: {}", e))?;
251        if let Some(input_file_path) = pre_processor_output_file {
252            fs::write(&input_file_path, input_string)
253                .map_err(|e| anyhow::anyhow!("Unable to write input file: {}", e))?;
254            info!(
255                "Finished pre processing the data, saved the input file in {}",
256                input_file_path.display()
257            );
258            if output_file.is_none() && cairo_pie_file.is_none() {
259                Ok(())
260            } else {
261                info!("Starting processing the data... ");
262                let output_file_path = output_file
263                    .ok_or_else(|| anyhow::anyhow!("Output file path should be specified"))?;
264                let pie_file_path = cairo_pie_file
265                    .ok_or_else(|| anyhow::anyhow!("PIE path should be specified"))?;
266                let processor = Processor::new(config.sound_run_program_path.clone());
267                let processor_result = processor
268                    .process(preprocessor_result, &pie_file_path)
269                    .await?;
270                fs::write(
271                    &output_file_path,
272                    serde_json::to_string_pretty(&processor_result).map_err(|e| {
273                        anyhow::anyhow!("Failed to serialize processor result: {}", e)
274                    })?,
275                )
276                .map_err(|e| anyhow::anyhow!("Unable to write output file: {}", e))?;
277
278                info!(
279                    "Finished processing the data, saved the output file in {} and pie file in {}",
280                    output_file_path.display(),
281                    pie_file_path.display()
282                );
283                Ok(())
284            }
285        } else {
286            Err(anyhow::anyhow!("Cairo input path should be specified"))
287        }
288    }
289}
290
291pub async fn entry_run(
292    request_file: PathBuf,
293    rpc_url: Option<Url>,
294    dry_run_cairo_file: Option<PathBuf>,
295    pre_processor_output_file: PathBuf,
296    sound_run_cairo_file: Option<PathBuf>,
297    output_file: Option<PathBuf>,
298    cairo_pie_file: Option<PathBuf>,
299) -> Result<()> {
300    let request_context =
301        fs::read_to_string(request_file).expect("No request file exist in the path");
302    let parsed: SubmitBatchQuery = serde_json::from_str(&request_context)
303        .expect("Invalid format of request. Cannot parse it.");
304    let config = Config::init(
305        rpc_url,
306        Some(parsed.destination_chain_id),
307        dry_run_cairo_file,
308        sound_run_cairo_file,
309        None,
310    )
311    .await;
312    let module_registry = ModuleRegistry::new();
313    let mut task_envelopes = Vec::new();
314    for task in parsed.tasks {
315        match task {
316            Task::DatalakeCompute(task) => {
317                task_envelopes.push(TaskEnvelope::DatalakeCompute(task));
318            }
319            Task::Module(task) => {
320                let module = module_registry
321                    .get_extended_module_from_class_source(
322                        Some(task.program_hash),
323                        None,
324                        task.inputs,
325                    )
326                    .await?;
327                task_envelopes.push(TaskEnvelope::Module(module));
328            }
329        }
330    }
331    handle_running_tasks(
332        config,
333        task_envelopes,
334        Some(pre_processor_output_file),
335        output_file,
336        cairo_pie_file,
337    )
338    .await?;
339    Ok(())
340}
hdp_cli/common.rs

hdp_cli/
common.rs
