Merge branch 'feature/type-hints'

pylas/extradims.py

@@ -1,6 +1,5 @@
-from enum import Enum
-
 from . import errors
+from .point.dims import DimensionKind
 
 _extra_dims_base_style_1 = (
     "",
@@ -58,13 +57,7 @@
 }
 
 
-class DimensionSignedness(Enum):
-    FLOATING = 0
-    SIGNED = 1
-    UNSIGNED = 2
-
-
-def get_signedness_for_extra_dim(type_index):
+def get_kind_of_extra_dim(type_index: int) -> DimensionKind:
     """Returns the signedness foe the given type index
 
     Parameters
@@ -80,16 +73,16 @@ def get_signedness_for_extra_dim(type_index):
     try:
         t = _extra_dims_style_2[type_index]
         if "uint" in t:
-            return DimensionSignedness.UNSIGNED
+            return DimensionKind.UnsignedInteger
         elif "int" in t:
-            return DimensionSignedness.SIGNED
+            return DimensionKind.SignedInteger
         else:
-            return DimensionSignedness.FLOATING
+            return DimensionKind.FloatingPoint
     except IndexError:
         raise errors.UnknownExtraType(type_index)
 
 
-def get_type_for_extra_dim(type_index):
+def get_type_for_extra_dim(type_index: int) -> str:
     """Returns the type str ('u1" or "u2", etc) for the given type index
     Parameters
     ----------
@@ -108,7 +101,7 @@ def get_type_for_extra_dim(type_index):
         raise errors.UnknownExtraType(type_index)
 
 
-def get_id_for_extra_dim_type(type_str):
+def get_id_for_extra_dim_type(type_str: str) -> int:
     """Returns the index of the type as defined in the LAS Specification
 
     Parameters

pylas/headers/rawheader.py

@@ -3,6 +3,7 @@
 import enum
 import logging
 import uuid
+from typing import Union, BinaryIO
 
 import numpy as np
 
@@ -258,22 +259,28 @@ def set_compressed(self, compressed: bool):
 
     def update(self, points: PointRecord) -> None:
         self.x_max = max(
-            self.x_max, (points["X"].max() * self.x_scale) + self.x_offset,
+            self.x_max,
+            (points["X"].max() * self.x_scale) + self.x_offset,
         )
         self.y_max = max(
-            self.y_max, (points["Y"].max() * self.y_scale) + self.y_offset,
+            self.y_max,
+            (points["Y"].max() * self.y_scale) + self.y_offset,
         )
         self.z_max = max(
-            self.z_max, (points["Z"].max() * self.z_scale) + self.z_offset,
+            self.z_max,
+            (points["Z"].max() * self.z_scale) + self.z_offset,
         )
         self.x_min = min(
-            self.x_min, (points["X"].min() * self.x_scale) + self.x_offset,
+            self.x_min,
+            (points["X"].min() * self.x_scale) + self.x_offset,
         )
         self.y_min = min(
-            self.y_min, (points["Y"].min() * self.y_scale) + self.y_offset,
+            self.y_min,
+            (points["Y"].min() * self.y_scale) + self.y_offset,
         )
         self.z_min = min(
-            self.z_min, (points["Z"].min() * self.z_scale) + self.z_offset,
+            self.z_min,
+            (points["Z"].min() * self.z_scale) + self.z_offset,
         )
 
         for i, count in zip(*np.unique(points.return_number, return_counts=True)):
@@ -357,6 +364,14 @@ def partial_reset(self):
         self.number_of_points_by_return = [0] * 15
 
 
+Header = Union[
+    RawHeader1_1,
+    RawHeader1_2,
+    RawHeader1_3,
+    RawHeader1_4,
+]
+
+
 class HeaderFactory:
     """Factory to create a new header by specifying the version.
     This Factory also handles converting headers between different
@@ -372,7 +387,7 @@ class HeaderFactory:
     _offset_to_major_version = RawHeader1_1.version_major.offset
 
     @classmethod
-    def header_class_for_version(cls, version):
+    def header_class_for_version(cls, version: Union[str, float]):
         """
         >>> HeaderFactory.header_class_for_version(2.0)
         Traceback (most recent call last):
@@ -393,7 +408,7 @@ def header_class_for_version(cls, version):
             raise errors.FileVersionNotSupported(version)
 
     @classmethod
-    def new(cls, version):
+    def new(cls, version: Union[str, float]) -> Header:
         """Returns a new instance of a header.
 
         Parameters
@@ -412,7 +427,7 @@ def new(cls, version):
         return cls.header_class_for_version(version)()
 
     @classmethod
-    def read_from_stream(cls, stream):
+    def read_from_stream(cls, stream: BinaryIO) -> Header:
         sizeof_u8 = ctypes.sizeof(ctypes.c_uint8)
         header_bytes = bytearray(
             stream.read(cls._offset_to_major_version + (sizeof_u8 * 2))
@@ -443,7 +458,7 @@ def from_mmap(cls, mmap):
         return cls.header_class_for_version(version).from_buffer(mmap)
 
     @classmethod
-    def peek_file_version(cls, stream):
+    def peek_file_version(cls, stream: BinaryIO) -> str:
         """seeks to the position of the las version header fields
         in the stream and returns it as a str
 
@@ -462,10 +477,12 @@ def peek_file_version(cls, stream):
         major = int.from_bytes(stream.read(ctypes.sizeof(ctypes.c_uint8)), "little")
         minor = int.from_bytes(stream.read(ctypes.sizeof(ctypes.c_uint8)), "little")
         stream.seek(old_pos)
-        return "{}.{}".format(major, minor)
+        return f"{major}.{minor}"
 
     @classmethod
-    def convert_header(cls, old_header, new_version):
+    def convert_header(
+        cls, old_header: Header, new_version: Union[str, float]
+    ) -> Header:
         """Converts a header to a another version
 
         Parameters

pylas/lasappender.py

@@ -1,7 +1,9 @@
 import io
 import math
-from typing import Union, Iterable
+from typing import Union, Iterable, BinaryIO
 
+from .headers.rawheader import Header
+from .vlrs.vlrlist import VLRList
 from .compression import LazBackend
 from .errors import PylasError
 from .evlrs import EVLRList, RawEVLRList
@@ -17,15 +19,18 @@
 
 
 class LazrsAppender:
-    """ Appending in LAZ file works by seeking to start of the last chunk
+    """Appending in LAZ file works by seeking to start of the last chunk
     of compressed points, decompress it while keeping the points in
     memory.
 
     Then seek back to the start of the last chunk, and recompress
     the points we just read, so that we have a compressor in the proper state
     ready to compress new points.
     """
-    def __init__(self, dest, header, vlrs, parallel):
+
+    def __init__(
+        self, dest: BinaryIO, header: Header, vlrs: VLRList, parallel: bool
+    ) -> None:
         self.dest = dest
         self.offset_to_point_data = header.offset_to_point_data
         laszip_vlr = vlrs.pop(vlrs.index("LasZipVlr"))
@@ -76,10 +81,10 @@ def __init__(self, dest, header, vlrs, parallel):
         self.dest.seek(sum(self.chunk_table), io.SEEK_CUR)
         self.compressor.compress_many(points_of_last_chunk)
 
-    def write_points(self, points):
+    def write_points(self, points: PointRecord) -> None:
         self.compressor.compress_many(points.memoryview())
 
-    def done(self):
+    def done(self) -> None:
         # The chunk table written is at the good position
         # but it is incomplete (it's missing the chunk_table of
         # chunks before the one we appended)
@@ -97,7 +102,7 @@ def done(self):
 class LasAppender:
     def __init__(
         self,
-        dest,
+        dest: BinaryIO,
         laz_backend: Union[LazBackend, Iterable[LazBackend]] = (
             LazBackend.LazrsParallel,
             LazBackend.Lazrs,
@@ -155,14 +160,14 @@ def close(self) -> None:
         if self.closefd:
             self.dest.close()
 
-    def _write_evlrs(self):
+    def _write_evlrs(self) -> None:
         if self.header.version >= "1.4" and len(self.evlrs) > 0:
             self.header.number_of_evlr = len(self.evlrs)
             self.header.start_of_first_evlr = self.dest.tell()
             raw_evlrs = RawEVLRList.from_list(self.evlrs)
             raw_evlrs.write_to(self.dest)
 
-    def _write_updated_header(self):
+    def _write_updated_header(self) -> None:
         pos = self.dest.tell()
         self.dest.seek(0, io.SEEK_SET)
         self.header.write_to(self.dest)
@@ -204,8 +209,8 @@ def _create_laz_backend(
         else:
             raise PylasError(f"No valid laz backend selected")
 
-    def __enter__(self):
+    def __enter__(self) -> "LasAppender":
         return self
 
-    def __exit__(self, exc_type, exc_val, exc_tb):
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         self.close()

pylas/lasdatas/base.py

@@ -108,7 +108,10 @@ def __getattr__(self, item):
         The requested dimension if it exists
 
         """
-        return self.points[item]
+        try:
+            return self.points[item]
+        except ValueError:
+            raise AttributeError(f"{self.__class__.__name__} object has no attribute '{item}'") from None
 
     def __setattr__(self, key, value):
         """This is called on every access to an attribute of the instance.
@@ -124,7 +127,9 @@ def __setattr__(self, key, value):
         if key in self.point_format.dimension_names:
             self.points[key] = value
         elif key in dims.DIMENSIONS_TO_TYPE:
-            raise ValueError(f"Point format {self.point_format} does not support {key} dimension")
+            raise ValueError(
+                f"Point format {self.point_format} does not support {key} dimension"
+            )
         else:
             super().__setattr__(key, value)
 
@@ -187,7 +192,7 @@ def update_header(self):
             self.header.number_of_points_by_return = counts
 
     def write_to(
-            self, out_stream, do_compress=False, laz_backend=LazBackend.detect_available()
+        self, out_stream, do_compress=False, laz_backend=LazBackend.detect_available()
     ):
         """writes the data to a stream
 
@@ -201,12 +206,12 @@ def write_to(
             By default, pylas detect available backends
         """
         with LasWriter(
-                out_stream,
-                self.header,
-                self.vlrs,
-                do_compress=do_compress,
-                closefd=False,
-                laz_backend=laz_backend,
+            out_stream,
+            self.header,
+            self.vlrs,
+            do_compress=do_compress,
+            closefd=False,
+            laz_backend=laz_backend,
         ) as writer:
             writer.write(self.points)
 
@@ -239,7 +244,7 @@ def write_to_file(self, filename, do_compress=None):
             self.write_to(out, do_compress=do_compress)
 
     def write(
-            self, destination, do_compress=None, laz_backend=LazBackend.detect_available()
+        self, destination, do_compress=None, laz_backend=LazBackend.detect_available()
     ):
         """Writes to a stream or file
 
@@ -277,9 +282,7 @@ def write(
         else:
             if do_compress is None:
                 do_compress = False
-            self.write_to(
-                destination, do_compress=do_compress, laz_backend=laz_backend
-            )
+            self.write_to(destination, do_compress=do_compress, laz_backend=laz_backend)
 
     def __repr__(self):
         return "<LasData({}.{}, point fmt: {}, {} points, {} vlrs)>".format(

pylas/lasmmap.py

@@ -4,6 +4,7 @@
 from .lasdatas import base
 from .point import PointFormat, record
 from .vlrs import vlrlist
+from .typehints import PathLike
 
 WHOLE_FILE = 0
 
@@ -23,7 +24,7 @@ class LasMMAP(base.LasBase):
         A LAZ (compressed LAS) cannot be mmapped
     """
 
-    def __init__(self, filename):
+    def __init__(self, filename: PathLike) -> None:
         fileref = open(filename, mode="r+b")
 
         m = mmap.mmap(fileref.fileno(), length=WHOLE_FILE, access=mmap.ACCESS_WRITE)
@@ -48,7 +49,7 @@ def __init__(self, filename):
         self.fileref, self.mmap = fileref, m
         self.mmap.seek(self.header.size)
 
-    def close(self):
+    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
@@ -57,8 +58,8 @@ def close(self):
         self.mmap.close()
         self.fileref.close()
 
-    def __enter__(self):
+    def __enter__(self) -> "LasMMAP":
         return self
 
-    def __exit__(self, exc_type, exc_val, exc_tb):
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
         self.close()

pylas/lasreader.py

@@ -13,6 +13,7 @@
 from .lasdatas import las14, las12
 from .point import record, PointFormat
 from .point.dims import size_of_point_format_id
+from .typehints import LasData
 from .utils import ConveyorThread
 from .vlrs.known import LasZipVlr
 from .vlrs.vlrlist import VLRList
@@ -95,7 +96,7 @@ def read_n_points(self, n: int) -> Optional[record.ScaleAwarePointRecord]:
         self.points_read += n
         return points
 
-    def read(self) -> Union[las12.LasData, las14.LasData]:
+    def read(self) -> LasData:
         """Reads all the points not read and returns a LasData object"""
         points = self.read_n_points(-1)
         if points is None:
@@ -135,7 +136,7 @@ def close(self) -> None:
         if self.closefd:
             self.point_source.close()
 
-    def _create_laz_backend(self, source):
+    def _create_laz_backend(self, source) -> Optional["IPointReader"]:
         try:
             backends = iter(self.laz_backend)
         except TypeError: