| RDB 2

rdb-example-12-clone-file.py
1 #!/usr/bin/env python3
2 # -*- coding: utf-8 -*-
3 #
4 #*******************************************************************************
5 #
6 # Copyright 2025 RIEGL Laser Measurement Systems
7 #
8 # Licensed under the Apache License, Version 2.0 (the "License");
9 # you may not use this file except in compliance with the License.
10 # You may obtain a copy of the License at
11 #
12 # http://www.apache.org/licenses/LICENSE-2.0
13 #
14 # Unless required by applicable law or agreed to in writing, software
15 # distributed under the License is distributed on an "AS IS" BASIS,
16 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 # See the License for the specific language governing permissions and
18 # limitations under the License.
19 #
20 # SPDX-License-Identifier: Apache-2.0
21 #
22 #*******************************************************************************
23 #
24 """
25 rdb-example-12-clone-file.py
26
27 This example shows how to clone (copy) an existing database.
28 """
29
30 from typing import Iterable, Optional
31
32 from riegl.rdb import *
33
34
35 def rdb_clone(
36 source: str, # input file path (must exist)
37 target: str, # output file path (is overwritten)
38 selection: Optional[str] = None, # point filter string (None = all)
39 attributes: Optional[Iterable[str]] = None, # point attribute names (None = all)
40 chunk_size: int = 100*1000 # number of points to copy in one step
41 ) -> None:
42 ctx = Context()
43 with rdb_open(source, context=ctx) as reader:
44 # get source file creation settings:
45 settings = CreateSettings(ctx)
46 settings.load(reader.transactions[1].settings)
47
48 # create target file with the same settings:
49 with rdb_create(target, settings, context=ctx) as writer:
50 with Transaction(writer, "Clone", "Duplicator 1.0") as transaction:
51 # clone all meta data entries:
52 for name, value in reader.meta_data.items():
53 if name == RDB_RIEGL_POINT_ATTRIBUTE_GROUPS:
54 continue # always defined -> skip
55 writer.meta_data[name] = value
56
57 # clone all point attribute definitions:
58 attributes = set(attributes or reader.point_attributes)
59 attributes.discard(RDB_RIEGL_ID) # is auto-generated
60 for name in attributes:
61 if name == settings.primary_attribute.name:
62 continue # always defined -> skip
63 writer.point_attributes.add(reader.point_attributes[name])
64
65 # clone all point data:
66 buffer = PointBuffer(reader, attributes, chunk_size)
67 with reader.select(selection) as select, writer.insert() as insert:
68 select.bind(buffer) # tell select query where to write data to
69 insert.bind(buffer) # tell insert query where to read data from
70 while count := select.next(chunk_size):
71 insert.next(count)
72
73 # finally commit transaction:
74 transaction.commit()