File: lasmmap.py

package info (click to toggle)
python-laspy 2.5.4-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 3,928 kB
  • sloc: python: 9,065; makefile: 20
file content (58 lines) | stat: -rw-r--r-- 1,691 bytes parent folder | download
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
import io
import mmap

from . import lasdata
from .header import LasHeader
from .point import record
from .typehints import PathLike

WHOLE_FILE = 0


class LasMMAP(lasdata.LasData):
    """Memory map a LAS file.
    It works like a regular LasData however the data is not actually read in memory,

    Access to dimensions are made directly from the file itself, changes made to the points
    are directly reflected in the mmap file.

    Vlrs cannot be modified.

    This can be useful if you want to be able to process a big LAS file

    .. note::
        A LAZ (compressed LAS) cannot be mmapped
    """

    def __init__(self, filename: PathLike) -> None:
        fileref = open(filename, mode="r+b")

        m = mmap.mmap(fileref.fileno(), length=WHOLE_FILE, access=mmap.ACCESS_WRITE)
        header = LasHeader.read_from(m)
        if header.are_points_compressed:
            raise ValueError("Cannot mmap a compressed LAZ file")

        points_data = record.PackedPointRecord.from_buffer(
            m,
            header.point_format,
            count=header.point_count,
            offset=header.offset_to_point_data,
        )
        super().__init__(header=header, points=points_data)

        self.fileref, self.mmap = fileref, m
        self.mmap.seek(0, io.SEEK_SET)

    def close(self) -> None:
        # These need to be set to None, so that
        # mmap.close() does not give an error because
        # there are still exported pointers
        self._points = None
        self.mmap.close()
        self.fileref.close()

    def __enter__(self) -> "LasMMAP":
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        self.close()