utiles 0.8.0

Web map tile utils (aka utiles)
Documentation 
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
151

use futures::{Stream, StreamExt};
use hex::ToHex;
use md5::Digest;
use noncrypto_digests::Fnv;
use serde::Serialize;
use tokio_stream::wrappers::ReceiverStream;
use tracing::debug;

use utiles_core::TileLike;

use crate::errors::UtilesResult;
use crate::hash_types::HashType;
use crate::mbt::MbtilesClientAsync;
use crate::mbt::TilesFilter;
use crate::tile_stream::TileReceiverStream;

#[derive(Debug, Serialize)]
pub struct AggHashResult {
    pub hash_type: HashType,
    pub hash: String,
    pub ntiles: usize,
    pub dt: std::time::Duration,
}
// =================================================================
// HASH FUNCTIONS ~ HASH FUNCTIONS ~ HASH FUNCTIONS ~ HASH FUNCTIONS
// =================================================================
// pub fn mbt_agg_tile_hash_query(
//     hash_type: HashType,
//     prefix: Option<&str>,
//     filter: &Option<TilesFilter>,
// ) -> UtilesResult<String> {
//     let where_clause = if let Some(filter) = filter {
//         filter.mbtiles_sql_where(prefix)?
//     } else {
//         String::new()
//     };
//     let sql = format!(
//         "
// SELECT
//     coalesce(
//         {hash_type}_concat_hex(
//             cast(zoom_level AS text),
//             cast(tile_column AS text),
//             cast(tile_row AS text),
//             tile_data
//             ORDER BY zoom_level, tile_column, tile_row
//         ),
//         {hash_type}_hex('')
//     ) AS concatenated_hash,
//     COUNT(*) AS total_count
// FROM tiles
// {where_clause}
//     ",
//     );
//     Ok(sql)
// }
//
// pub fn mbt_agg_tiles_hash_query(
//     conn: &Connection,
//     hash_type: HashType,
//     prefix: Option<&str>,
//     filter: &Option<TilesFilter>,
// ) -> UtilesResult<AggHashResult> {
//     let sql = mbt_agg_tile_hash_query(hash_type, prefix, filter)?;
//     debug!("sql: {:?}", sql);
//     let mut stmt = conn.prepare_cached(&sql)?;
//     // start time
//     let ti = std::time::Instant::now();
//     let (agg_tiles_hash_str, count): (String, i64) =
//         stmt.query_row([], |row| Ok((row.get(0)?, row.get(1)?)))?;
//     let dt = ti.elapsed();
//     Ok(AggHashResult {
//         hash_type,
//         hash: agg_tiles_hash_str,
//         ntiles: count as usize,
//         dt,
//     })
// }

pub(super) async fn hash_stream<T: Digest>(
    mut data: impl Stream<Item = Vec<u8>> + Unpin,
) -> (String, usize) {
    let mut hasher = T::new();
    let mut count = 0;
    while let Some(chunk) = data.next().await {
        hasher.update(&chunk);
        count += 1;
    }
    // hasher.update(data);
    (hasher.finalize().to_vec().encode_hex_upper(), count)
}
pub(super) fn tile_stream_to_bytes_stream(
    mut data: TileReceiverStream,
) -> ReceiverStream<Vec<u8>> {
    let (tx, rx) = tokio::sync::mpsc::channel(100);

    tokio::spawn(async move {
        while let Some((tile, tile_data)) = data.next().await {
            let bytes = [
                tile.z().to_string().as_bytes().to_vec(),
                tile.x().to_string().as_bytes().to_vec(),
                tile.yup().to_string().as_bytes().to_vec(),
                tile_data,
            ]
            .concat();
            if let Err(e) = tx.send(bytes).await {
                debug!("tile_stream_to_bytes_stream send error: {:?}", e);
            }
        }
    });
    ReceiverStream::new(rx)
}
impl HashType {
    async fn hash_stream(&self, data: ReceiverStream<Vec<u8>>) -> (String, usize) {
        match self {
            Self::Md5 => hash_stream::<md5::Md5>(data).await,
            Self::Fnv1a => hash_stream::<Fnv>(data).await,
            Self::Xxh32 => hash_stream::<noncrypto_digests::Xxh32>(data).await,
            Self::Xxh64 => hash_stream::<noncrypto_digests::Xxh64>(data).await,
            Self::Xxh3_64 => hash_stream::<noncrypto_digests::Xxh3_64>(data).await,
            Self::Xxh3_128 => hash_stream::<noncrypto_digests::Xxh3_128>(data).await,
        }
    }
}

pub async fn mbt_agg_tiles_hash_stream(
    mbt: &MbtilesClientAsync,
    hash_type: HashType,
    prefix: Option<&str>,
    filter: &Option<TilesFilter>,
) -> UtilesResult<AggHashResult> {
    let where_clause = if let Some(filter) = filter {
        filter.mbtiles_sql_where(prefix)?
    } else {
        String::new()
    };
    let query = format!(
        "SELECT zoom_level, tile_column, tile_row, tile_data FROM tiles {where_clause} ORDER BY zoom_level, tile_column, tile_row;"
    );
    let stream = mbt.tiles_stream(Some(query.as_str()))?;
    let bstream = tile_stream_to_bytes_stream(stream);
    let ti = std::time::Instant::now();
    let (agg_tiles_hash_str, ntiles) = hash_type.hash_stream(bstream).await;
    let dt = ti.elapsed();
    Ok(AggHashResult {
        hash_type,
        hash: agg_tiles_hash_str,
        ntiles,
        dt,
    })
}