liboxen 0.53.0

Oxen is a fast data version control system, built with machine learning training data in mind. Designed to handle terabytes of data with ease, using a workflow similar to git. Version both structured and unstructured data of any modality: text, images, video, audio, CSV, Parquet, JSONL, model checkpoints, and more. liboxen is the embeddable core library behind the oxen CLI and server, which power fine tuning and inference pipelines for multimodal LLMs, image models, and video models on Oxen.ai.
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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150

use crate::config::RepositoryConfig;
use crate::core::progress::pull_progress::PullProgress;
use crate::error::OxenError;
use crate::model::CommitEntry;
use crate::model::LocalRepository;
use crate::model::MetadataEntry;
use crate::model::RemoteRepository;
use crate::model::merkle_tree::node::EMerkleTreeNode;
use crate::model::merkle_tree::node::MerkleTreeNode;
use crate::{api, repositories};
use std::path::Path;
use std::sync::Arc;

use crate::core;

pub async fn download_dir(
    remote_repo: &RemoteRepository,
    entry: &MetadataEntry,
    remote_path: impl AsRef<Path>,
    local_path: impl AsRef<Path>,
) -> Result<(), OxenError> {
    let remote_path = remote_path.as_ref();
    let local_path = local_path.as_ref();
    log::debug!("downloading dir {remote_path:?}");
    // Initialize temp repo to download node into
    // TODO: Where should this repo be?
    let tmp_repo = LocalRepository::new(local_path, RepositoryConfig::default())?;

    // Find and download dir node and its children from remote repo
    let commit_id = &entry.latest_commit.as_ref().unwrap().id;
    let dir_node = api::client::tree::download_tree_from_path(
        &tmp_repo,
        remote_repo,
        commit_id,
        remote_path.to_string_lossy(),
        true,
    )
    .await?;

    // Track Progress
    let pull_progress = Arc::new(PullProgress::new());

    // Recursively pull entries
    r_download_entries(
        remote_repo,
        &tmp_repo.path,
        &dir_node,
        remote_path,
        &pull_progress,
    )
    .await?;

    pull_progress.finish();

    Ok(())
}

pub async fn download_dir_entries(
    local_repo: &LocalRepository,
    remote_repo: &RemoteRepository,
    entry: &MetadataEntry,
    remote_path: impl AsRef<Path>,
    local_path: impl AsRef<Path>,
) -> Result<(), OxenError> {
    let remote_path = remote_path.as_ref();
    let local_path = local_path.as_ref();
    log::debug!("downloading dir {remote_path:?}");

    // Get tree from local repos
    let commit = &entry.latest_commit.as_ref().unwrap();
    let Some(dir_node) =
        repositories::tree::get_dir_with_children_recursive(local_repo, commit, remote_path, None)?
    else {
        log::warn!("Dir node not found for path {local_path:?}");
        return Ok(());
    };

    // Recursively pull entries
    let pull_progress = Arc::new(PullProgress::new());
    r_download_entries(
        remote_repo,
        &local_repo.path,
        &dir_node,
        remote_path,
        &pull_progress,
    )
    .await?;

    pull_progress.finish();

    Ok(())
}

async fn r_download_entries(
    remote_repo: &RemoteRepository,
    local_repo_path: &Path,
    node: &MerkleTreeNode,
    directory: &Path,
    pull_progress: &Arc<PullProgress>,
) -> Result<(), OxenError> {
    for child in &node.children {
        let mut new_directory = directory.to_path_buf();
        if let EMerkleTreeNode::Directory(dir_node) = &child.node {
            new_directory.push(dir_node.name());
        }

        let has_children = child.has_children();
        if has_children {
            Box::pin(r_download_entries(
                remote_repo,
                local_repo_path,
                child,
                &new_directory,
                pull_progress,
            ))
            .await?;
        }
    }

    if let EMerkleTreeNode::VNode(_) = &node.node {
        let mut entries: Vec<CommitEntry> = vec![];

        for child in &node.children {
            if let EMerkleTreeNode::File(file_node) = &child.node {
                entries.push(CommitEntry {
                    commit_id: file_node.last_commit_id().to_string(),
                    path: directory.join(file_node.name()),
                    hash: child.hash.to_string(),
                    num_bytes: file_node.num_bytes(),
                    last_modified_seconds: file_node.last_modified_seconds(),
                    last_modified_nanoseconds: file_node.last_modified_nanoseconds(),
                });
            }
        }

        log::debug!(
            "r_download_entries downloading {} entries to working dir",
            entries.len()
        );
        core::v_latest::fetch::download_entries_to_working_dir(
            remote_repo,
            &entries,
            local_repo_path,
            pull_progress,
        )
        .await?;
    }

    Ok(())
}