[Git][debian-gis-team/pyshp][upstream] 3 commits: New upstream version 3.1.0
Bas Couwenberg (@sebastic)
gitlab at salsa.debian.org
Thu Jun 25 04:52:39 BST 2026
Bas Couwenberg pushed to branch upstream at Debian GIS Project / pyshp
Commits:
9c5adba7 by Bas Couwenberg at 2026-06-25T05:39:17+02:00
New upstream version 3.1.0
- - - - -
2a72b146 by Bas Couwenberg at 2026-06-25T05:39:32+02:00
New upstream version 3.1.1
- - - - -
7ed26c44 by Bas Couwenberg at 2026-06-25T05:39:50+02:00
New upstream version 3.1.2
- - - - -
7 changed files:
- .gitignore
- README.md
- changelog.txt
- src/shapefile.py
- tests/hypothesis_tests.py
- tests/run_benchmarks.py
- tests/test_shapefile.py
Changes:
=====================================
.gitignore
=====================================
@@ -35,3 +35,4 @@ venv/
.mypy_cache/
.pytest_cache/
.ruff_cache/
+.hypothesis
=====================================
README.md
=====================================
@@ -8,8 +8,8 @@ The Python Shapefile Library (PyShp) reads and writes ESRI Shapefiles in pure Py
- **Author**: [Joel Lawhead](https://github.com/GeospatialPython)
- **Maintainers**: [James Parrott](https://github.com/JamesParrott) & [Karim Bahgat](https://github.com/karimbahgat)
-- **Version**: 3.0.13
-- **Date**: 19th June 2026
+- **Version**: 3.1.2
+- **Date**: 24th June 2026
- **License**: [MIT](https://github.com/GeospatialPython/pyshp/blob/master/LICENSE.TXT)
## Contents
@@ -93,6 +93,43 @@ part of your geospatial project.
# Version Changes
+## 3.1.2
+ - Raise error in strict mode when creating field with a name, or writing strings, that ends with whole code points
+ that whose encoding is pad bytes.
+
+## 3.1.1
+### Unicode support made even more robust and yet another encoding bug fixed!
+ - When reading, only use minimum number of pad bytes to decode text successfully (fixes issue 423).
+ - When writing, warn (or raise in strict mode) if the text's encoding ends in pad bytes.
+
+## 3.1.0
+### Unicode support made more robust and encoding bugs fixed
+ - Truncation of field names and text fields now respects unicode code point boundaries (fixes issues -
+ 416 and 148).
+ - Warnings raised if truncation, or replacing b" " with b"_" would corrupt encoded field names, -
+ both if they would either be undecodable, or would silently decode to incorrect data (warns users
+ if issue 421 applies).
+ - Correctly truncated field names are now stored in field instances, as would actually be seen in the file.
+ - New strict mode. Writer(strict=True) raises errors or refuse to create fields and text records with data that -
+ would be truncated or cannot be correctly decoded back again by PyShp, exactly as given by the user.
+ - In strict mode, ascii spaces in encoded names are no longer replaced by ascii underscores at all
+ (work around to avoid corrupting unicode field names - provides opt-in fix for issue 421).
+ - BREAKING. When reading .dbf files, Trailing ascii spaces in text fields before a null terminator char (in the -
+ decoded string) is now removed (i.e. instead of .strip().rstrip('\x00') we now do: .rstrip("\x00").rstrip(" ")).
+ - BREAKING. Enclosing Whitespace other than trailing ascii spaces (0x20) after null chars in text fields is now -
+ preserved, when reading .dbf files (fixes issue 418 - James feels this was a bug. Let him know if you think otherwise).
+ - BREAKING Trailing null chars other than null terminators & null padding bytes, followed by whitespace other than
+ ascii spaces, are now preserved.
+ - Writing dbf records is now atomic.
+
+### ShpWriter.shape API Tweak (small breaking change).
+ - Make ShpWriter.shape return shape length in bytes (the
+ same as for offset) not in 16 bit words.
+### Testing
+ - Include NullShapes in shp round trip test.
+ - En/decoding of Dbf files and Fields round trips correctly.
+
+
## 3.0.13
### Bug fix
- Fix bug when reading empty shp files.
@@ -1301,8 +1338,14 @@ All logging happens under the namespace `shapefile`. So another way to suppress
PyShp supports reading and writing shapefiles in any language or character encoding, and provides several options for decoding and encoding text.
Most shapefiles are written in UTF-8 encoding, PyShp's default encoding, so in most cases you don't have to specify the encoding.
+#### Encoding errors
If you encounter an encoding error when reading a shapefile, this means the shapefile was likely written in a non-utf8 encoding.
For instance, when working with English language shapefiles, a common reason for encoding errors is that the shapefile was written in Latin-1 encoding.
+#### Recommendation
+*We recommend avoiding writing Shapefiles with unicode encoded as UTF-16 and UTF-32, especially UTF-16-LE (if at all possible).*
+When using other encodings, especially UTF-32 or UTF-16, a whole host of other known (and unknown) unicode related issues can be encountered. See ## Specialist Unicode Handling below for more details.
+
+#### Non-utf8 encodings
For reading shapefiles in any non-utf8 encoding, such as Latin-1, just
supply the encoding option when creating the Reader class.
@@ -1329,7 +1372,10 @@ should give you the same unicode string you started with.
>>> r.close()
If you supply the wrong encoding and the string is unable to be decoded, PyShp will by default raise an
-exception. If however, on rare occasion, you are unable to find the correct encoding and want to ignore
+exception.
+
+#### Custom handling of encoding errors.
+If however, on rare occasion, you are unable to find the correct encoding and want to ignore
or replace encoding errors, you can specify the "encodingErrors" to be used by the decode method. This
applies to both reading and writing.
@@ -1339,7 +1385,100 @@ applies to both reading and writing.
True
>>> r.close()
+Valid values for encodingErrors are those supported by both `bytes.decode` and `str.encode` in the Python
+version being used, e.g. 'strict', 'ignore' or 'replace'
+in [CPython 3.9 - 3.14](https://docs.python.org/3/library/stdtypes.html#bytes.decode)
+('xmlcharrefreplace' and 'backslashreplace' are only supported by
+[`str.encode`](https://docs.python.org/3/library/stdtypes.html#str.encode)).
+
+## Specialist Unicode handling.
+### Summary
+If at all possible use UTF-8. Or at the very least avoid UTF-16 and UTF-32
+### Background
+Unicode is one of humanity's greatest achievements. GIS applications support it within Shapefiles.
+But most unicode encodings (e.g. UTF-8, UTF-16) assume
+data will be stored in pure binary form, as bytes (8-bit octets). The Shapefile spec however, defers to
+a long lost dbf spec (the URL link to a former company's website is now dead) which like most other dbf
+specifications, will require text to be stored as Ascii.
+The fields are some known width, and Extended Ascii character sets (with 256 entries, not only the first
+128 common to them all) are essentially one-to-one mappings to bytes - so this would not be a
+problem in isolation.
+### Padding bytes
+Dbf field names are post padded with null bytes, and dbf text fields ("C" and "M") are post
+padded with Ascii spaces (the byte 0x20). Example shapefiles exist where text fields also end in numerous
+null bytes, essentially also as padding.
+#### Problem 1.
+Encodings of some unicode strings may contain these exact padding bytes as part of their data (in particular
+the first null byte is not necessarily a null terminator).
+
+ >>> "囊萤映雪".encode("utf-16-be")
+ b'V\xca\x84$f \x96\xea'
+
+#### Problem 2.
+Some unicode strings encode to byte sequences under some non-UTF-8 codecs, that contain and even
+trail a null byte as part of their data.
+
+ >>> "ABC".encode("utf-16-le")
+ b'A\x00B\x00C\x00'
+
+#### Problem 3.
+Some unicode strings encode to byte sequences under some non-UTF-8 codecs, that
+are entirely dbf padding bytes, e.g.:
+
+ >>> '††††'.encode("utf-16-le")
+ b' '
+
+#### Problem 4.
+If encodings of even only moderately long unicode strings (especially the 10 byte field names) are truncated at byte
+boundaries (i.e. not truncated correctly at one of the string's code-point boundaries), their data is corrupted, and
+the encoding of the final code point whose bytes were chopped is corrupted.
+ >>> try:
+ ... "ÀÀÀÀ०".encode()[:10].decode()
+ ... except UnicodeDecodeError:
+ ... print("Failed to decode")
+ ...
+ Failed to decode
+
+#### Problem 5.
+Historically, PyShp has attempted to prettify dbf string data. E.g. stripping whitespace.
+In particular, ascii spaces (`b" "`) in a dbf name were transformed to underscores (`b"_"`) - but
+crucially, only after encoding to bytes, whether the codec was ascii or not. This can result in silent
+changes to the user's data, and since version 3.1 PyShp no longer does this.
+
+ >>> s = "囊萤映雪"
+ >>> s2 = s.encode("utf-16-be").replace(b" ",b"_").decode("utf-16-be")
+ >>> print(s2)
+ 囊萤晟雪
+ >>> s == s2
+ False
+
+### PyShp's approach
+The simplest most robust solution to support arbitrary codec encoding as ascii bytes, is probably to encode those
+bytes as Base64 (with an alphabet excluding dbf pad bytes), and then encode that Base64 string a third time as bytes.
+PyShp tries to be a little easier to use than this. The intention is to adhere to [Postel's Law](https://en.wikipedia.org/wiki/Robustness_principle) so hopefully PyShp is still fairly lenient when decoding and reading Shapefiles, but either informative or
+stricter, when the user attempts to encoding or write invalid files.
+#### Warnings
+When possible data corruption is detected when reading or writing a Shapefile (or dbf file), by default
+PyShp will raise a warning. Supported situations are listed below under Decoding and Encoding.
+#### Strict mode
+If a Writer (or dbf version) is created with `strict=True` then an exception is raised instead of any of the above warnings
+(hopefully before corrupt data can be written at all).
+#### Decoding
+ - PyShp first removes all padding bytes. If decoding fails, those padding bytes are restored one by one to the data
+ until decoding succeeds. A warning is raised if any padding bytes were required, but an exception is not raised for this
+ in strict mode (those pad bytes might just have been part of a valid encoding).
+ - PyShp warns if there is a null character in a decoded string (but silently accepts null bytes in encoded strings). This
+ warning does not become an exception in strict mode (there might simply be null characters in the data).
+#### Encoding
+ - PyShp warns when trying to write data to a text field or field name that PyShp itself would not be able to decode correctly.
+ - PyShp now truncates text fields the dbf field `size` or 10 byte limit for names, by if necessary, successively
+truncating the actual string, one code point at a time, until the number of encoded bytes is less than or equal
+to the maximum. If truncation was necessary, a warning is raised.
+ - When trying to write text data, that ends with code points that would encode
+ to pad bytes (these would be lost on decoding, otherwise all fields stored as utf-16 under the limit will end
+ in ††††††††††††...).
+ - PyShp warns if a field name is written containing a null character (before encoding). For users who wish to steer well clear of potential bugs.
## Reading Large Shapefiles
=====================================
changelog.txt
=====================================
@@ -1,3 +1,47 @@
+VERSION 3.1.2
+
+2026-06-24
+ * Raise error in strict mode when creating field with a name, or writing strings, that ends with whole code points
+ that whose encoding is pad bytes.
+ * Document handling of unicode.
+
+VERSION 3.1.1
+
+2026-06-24
+ Unicode support made even more robust and yet another encoding bug fixed!
+ * When reading, only use minimum number of pad bytes to decode text successfully (fixes issue 423).
+ * When writing, warn (or raise in strict mode) if the text's encoding ends in pad bytes.
+
+VERSION 3.1.0
+
+2026-06-23
+ Unicode support made more robust and encoding bugs fixed
+ * Truncation of field names and text fields now respects unicode code point boundaries (fixes issues
+ 416 and 148).
+ * Warnings raised if truncation, or replacing b" " with b"_" would corrupt encoded field names,
+ both if they would either be undecodable, or would silently decode to incorrect data (warns users
+ if issue 421 applies).
+ * Correctly truncated field names are now stored in field instances, as would actually be seen in the file.
+ * New strict mode. Writer(strict=True) raises errors or refuse to create fields and text records with data that
+ would be truncated or cannot be correctly decoded back again by PyShp, exactly as given by the user.
+ * In strict mode, ascii spaces in encoded names are no longer replaced by ascii underscores at all
+ (work around to avoid corrupting unicode field names - provides opt-in fix for issue 421)
+ * BREAKING. When reading .dbf files, Trailing ascii spaces in text fields before a null terminator char (in the
+ decoded string) is now removed (i.e. instead of .strip().rstrip('\x00') we now do: .rstrip("\x00").rstrip(" ")).
+ * BREAKING. Enclosing Whitespace other than trailing ascii spaces (0x20) after null chars in text fields is now
+ preserved, when reading .dbf files (fixes issue 418 - James feels this was a bug. Let him know if you think otherwise).
+ * BREAKING Trailing null chars other than null terminators & null padding bytes, followed by whitespace other than
+ ascii spaces, are now preserved
+ * Writing dbf records is now atomic.
+ Testing.
+ * En/decoding of Dbf files and Fields round trips correctly.
+
+
+2026-06-20
+ * API Tweak (small breaking change). Make ShpWriter.shape return shape length in bytes
+ (the same as for offset) not in 16 bit words.
+ * Include NullShapes in shp round trip test.
+
VERSION 3.0.13
2026-06-19
@@ -21,7 +65,7 @@ VERSION 3.0.11
2026-06-04
Edge case handling
- * Raise ShapefileException i) when creating Non-null Shapes without (or with empty) points
+ * Raise ShapefileException: i) when creating Non-null Shapes without (or with empty) points
and ii) when creating Null Shapes with non-empty points.
* Ensure Shape.z and Shape.partTypes are _Arrays.
* Make Shape stricter about its args, e.g. only points or lines, only one point for Points.
=====================================
src/shapefile.py
=====================================
@@ -8,7 +8,7 @@ Compatible with Python versions >=3.9
from __future__ import annotations
-__version__ = "3.0.13"
+__version__ = "3.1.2"
import abc
import array
@@ -24,7 +24,7 @@ import warnings
import zipfile
from collections.abc import Container, Iterable, Iterator, Reversible, Sequence
from contextlib import AbstractContextManager, ExitStack
-from datetime import date
+from datetime import date, datetime
from os import PathLike
from pathlib import Path
from struct import Struct, calcsize, error, pack, unpack
@@ -231,13 +231,260 @@ for c in FieldType.__members__:
FIELD_TYPE_ALIASES[c.encode("ascii").upper()] = c
-# Use functional syntax to have an attribute named type, a Python keyword
+class PossibleDataLoss(Warning):
+ pass
+
+
+class DbfStringDataLoss(ValueError):
+ pass
+
+
+class Decoder(Protocol):
+ __name__: str
+
+ def __call__(
+ self,
+ b: bytes,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = False,
+ ) -> str: ...
+
+
+def _check_if_string_ends_with_decoded_pad_bytes(
+ s: str,
+ pad_byte: bytes,
+ encoding: str = "utf-8",
+ encodingErrors: str = "strict",
+ strict: bool = True,
+) -> None:
+ """Warns if e.g. the encoding is utf-16-le, and the
+ decoded text ends in "†", which encodes to a pair of
+ ascii spaces (b" ", the pad byte for C and M fields).
+ """
+ # Max code unit size under UTF-8, UTF-16, and UTF-32 is 4 bytes.
+ for n in range(1, 5):
+ # TODO: test for encodings ending in a null terminator preceded
+ # by pad bytes, that are exactly the field's size (length).
+ pad_bytes = pad_byte * n
+ try:
+ decoded_pad_bytes: str = pad_bytes.decode(encoding, encodingErrors)
+ except UnicodeDecodeError:
+ continue
+ if s.endswith(decoded_pad_bytes):
+ msg = (
+ f"Under the given encoding: {encoding}, "
+ f" the text (field name or 'C' or 'M' field): {s!r} "
+ f" ends with {decoded_pad_bytes!r}, which coincidentally"
+ f"encodes to the pad bytes: {pad_bytes!r}. "
+ "The real end of the actual data may be earlier. "
+ )
+ if strict:
+ raise DbfStringDataLoss(msg)
+ warnings.warn(msg, category=PossibleDataLoss)
+ break
+
+
+def _encode_dbf_string(
+ s: str,
+ size: int,
+ decode: Decoder,
+ pad_byte: bytes | None = None,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = True,
+) -> tuple[bytes, str]:
+ """Attempts to encoded s with the codec specified,
+ progressively truncating its code points until
+ the resulting bytes are less than size
+ (e.g. the dbf field length or field name size == 10).
+ If less, these bytes are then padded to length size.
+
+ Replaces: s.encode(self.encoding, self.encodingErrors)[:size]
+ .ljust(size))
+ in the legacy string encoding implementation:
+ """
+ N = len(s)
+ trimmed: str
+ encoded: bytes
+
+ # i - num of characters to keep. Starts by trying to keep all N.
+ for i in reversed(range(0, N + 1)):
+ trimmed = s[:i]
+ encoded = trimmed.encode(encoding, encodingErrors)
+
+ if len(encoded) <= size:
+ if i <= N - 1:
+ msg = (
+ f"Dropped {N - i} code points (e.g. characters)! "
+ f"{s} was truncated to {trimmed} (discarding: {s[i:]}), "
+ f"in order to encode it under {size} bytes for the field or field name. "
+ f"Used: {encoding=} and {encodingErrors=}. "
+ )
+ if strict:
+ raise DbfStringDataLoss(f"Data loss. {strict=}.\n{msg}")
+ warnings.warn(
+ msg,
+ category=PossibleDataLoss,
+ )
+ break
+ else: # for loop did not break, len(encoded) <= size,
+ # e.g. encoding "" preppends a BOM bigger than size.
+ raise ValueError(
+ f"Maximum truncation not sufficient to encode below {size=}. "
+ f"Could not encode first code point (e.g. character): {s[0]} "
+ f"to a short enough byte string, using {encoding=}, {encodingErrors=}"
+ )
+
+ if pad_byte is not None:
+ _check_if_string_ends_with_decoded_pad_bytes(
+ s=trimmed,
+ pad_byte=pad_byte,
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ strict=strict,
+ )
+
+ if len(encoded) < size and pad_byte is not None:
+ padded = encoded.ljust(size, pad_byte)
+ else:
+ padded = encoded
+
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore")
+ decoded = decode(
+ b=padded,
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ )
+
+ if decoded != trimmed:
+ msg = f"Padded value: {padded!r} does not decode to {trimmed!r} using PyShp's decoder: {decode.__name__}"
+ if len(trimmed) < len(s):
+ msg = f"{msg} (trimmed, original string: {s}). "
+ if strict:
+ raise DbfStringDataLoss(msg)
+ warnings.warn(
+ msg,
+ category=PossibleDataLoss,
+ )
+
+ return padded, trimmed
+
+
+def _try_to_decode_dbf_name_or_text_field(
+ b: bytes,
+ pad_bytes: bytes, # Pad bytes will be trimmed (from the R of b) in their order in the byte-string
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+) -> str:
+ N = len(b)
+ decoded: str
+ trimmed = b
+ num_trailing_pad_bytes = N - len(b.rstrip(pad_bytes))
+
+ # Test if we need to restore any pad_bytes to
+ # correctly decode the remaining bytes to a string.
+ # num_to_trim starts from num_trailing_pad_bytes
+ # - initially trimming all trailing pad bytes
+ for num_to_trim in reversed(range(num_trailing_pad_bytes + 1)):
+ i = N - num_to_trim
+ trimmed = b[:i]
+ try:
+ decoded = trimmed.decode(encoding, encodingErrors)
+ except UnicodeDecodeError:
+ continue
+ if num_to_trim < num_trailing_pad_bytes:
+ warnings.warn(
+ f"Used {num_trailing_pad_bytes - num_to_trim} pad bytes ({pad_bytes!r}) "
+ f"from padding to decode raw field: {b!r} "
+ f"to: {decoded!r} ({encoding=}, {encodingErrors=}) ",
+ category=PossibleDataLoss,
+ )
+ return decoded
+
+ raise dbfFileException(
+ f"Could not decode field name or text/memo field: {b!r} using {encoding=} and {encodingErrors=}"
+ " no matter how many trailing pad bytes (if any) ({pad_bytes!r}) were used. "
+ )
+
+
+def _decode_C_or_M_field(
+ b: bytes,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = True,
+) -> str:
+ retval = _try_to_decode_dbf_name_or_text_field(
+ b=b,
+ pad_bytes=b" \x00",
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ )
+
+ if not strict:
+ return retval
+
+ if retval.rstrip("\x00") != retval:
+ msg = (
+ f"More trailing null chars in: {retval!r}"
+ " after removing one trailing null char and ascii spaces"
+ f" from {b!r}, and decoding (codec: {encoding}, errors: {encodingErrors}). "
+ )
+ warnings.warn(msg, category=PossibleDataLoss)
+
+ return retval
+
+
class Field(NamedTuple):
name: str
field_type: FieldTypeT
size: int
decimal: int
+ @classmethod
+ @functools.cache
+ def get_struct(cls) -> Struct:
+ # En/decoding the name as "<10sx" embeds the null terminator.
+ return Struct("<10sxc4xBB14x")
+
+ @staticmethod
+ def decode_name(
+ b: bytes,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = True,
+ ) -> str:
+ decoded = _try_to_decode_dbf_name_or_text_field(
+ b=b,
+ pad_bytes=b"\x00",
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ )
+ if not strict:
+ decoded = decoded.lstrip()
+ return decoded
+
+ @classmethod
+ def from_byte_stream(
+ cls,
+ b_io: ReadableBinStream,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = False,
+ ) -> Field:
+ encoded_field_tuple: tuple[bytes, bytes, int, int]
+ encoded_field_tuple = cls.get_struct().unpack(b_io.read(32))
+ encoded_name, encoded_type_char, size, decimal = encoded_field_tuple
+
+ name = cls.decode_name(encoded_name, encoding, encodingErrors)
+
+ field_type = FIELD_TYPE_ALIASES[encoded_type_char]
+
+ return cls.from_unchecked(
+ name, field_type, size, decimal, encoding, encodingErrors, strict
+ )
+
@classmethod
def from_unchecked(
cls,
@@ -245,12 +492,26 @@ class Field(NamedTuple):
field_type: str | bytes | FieldTypeT = "C",
size: int = 50,
decimal: int = 0,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = False,
) -> Field:
+
+ if "\x00" in name:
+ msg = (
+ "Field names should not contain null characters "
+ "as null bytes are used for padding in the header. "
+ f"Got: {name=} "
+ )
+ if strict:
+ raise dbfFileException(msg)
+ warnings.warn(msg, category=PossibleDataLoss)
+
try:
type_ = FIELD_TYPE_ALIASES[field_type]
except KeyError:
- raise ShapefileException(
- f"field_type must be in {{FieldType.__members__}}. Got: {field_type=}. "
+ raise dbfFileException(
+ f"field_type must be in {FieldType.__members__}. Got: {field_type=}. "
)
if type_ is FieldType.D:
@@ -260,10 +521,78 @@ class Field(NamedTuple):
size = 1
decimal = 0
+ if not strict and " " in name:
+ warnings.warn(
+ f"Replacing ascii spaces (0x20, ' 's) with underscores ('_'s) in {name!r}. "
+ "Use a Writer(file, strict=True) to preserve the field name as it is. ",
+ category=PossibleDataLoss,
+ )
+ name = name.replace(" ", "_")
+
+ # Only use the portion of the name that we are able to encode to
+ # 10 bytes or less.
+ _encoded_name, trimmed_name = cls.trim_name_until_encodable(
+ name=str(name),
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ strict=strict,
+ )
+
# A doctest in README.md previously passed in a string ('40') for size,
# so explictly convert name to str, and size and decimal to ints.
- return cls(
- name=str(name), field_type=type_, size=int(size), decimal=int(decimal)
+ inst = cls(
+ name=trimmed_name, field_type=type_, size=int(size), decimal=int(decimal)
+ )
+
+ # Raise Exception or trigger warning early, before user adds more fields
+ # (fields are only written when first record added, and on close)
+ inst.encode_field_descriptor(
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ strict=strict,
+ )
+ return inst
+
+ @classmethod
+ def trim_name_until_encodable(
+ cls,
+ name: str,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = False,
+ ) -> tuple[bytes, str]:
+ return _encode_dbf_string(
+ s=name,
+ size=10,
+ decode=cls.decode_name,
+ pad_byte=b"\x00",
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ strict=strict,
+ )
+
+ @functools.cache
+ def encode_field_descriptor(
+ self,
+ encoding: str = "utf8",
+ encodingErrors: str = "strict",
+ strict: bool = False,
+ ) -> bytes:
+ # encoded_name = self.name.encode(encoding, encodingErrors)
+ # encoded_name = encoded_name[:10].ljust(10, b"\x00")
+ encoded_name, _trimmed_name = self.trim_name_until_encodable(
+ name=self.name,
+ encoding=encoding,
+ encodingErrors=encodingErrors,
+ strict=strict,
+ )
+
+ encoded_field_type = self.field_type.encode("ascii")
+ return self.get_struct().pack(
+ encoded_name,
+ encoded_field_type,
+ self.size,
+ self.decimal,
)
def __repr__(self) -> str:
@@ -875,7 +1204,7 @@ class Shape:
self._errors: dict[str, int] = {}
# add oid
- self.__oid: int = -1 if oid is None else oid
+ self._oid: int = -1 if oid is None else oid
if self.shapeType != NULL and self.shapeType not in Point_shapeTypes:
self.bbox: BBox = bbox or self._bbox_from_points()
@@ -915,7 +1244,7 @@ class Shape:
@property
def oid(self) -> int:
"""The index position of the shape in the original shapefile"""
- return self.__oid
+ return self._oid
@property
def shapeTypeName(self) -> str:
@@ -935,8 +1264,8 @@ class Shape:
def __repr__(self) -> str:
class_name = self.__class__.__name__
if class_name == "Shape":
- return f"Shape #{self.__oid}: {self.shapeTypeName}"
- return f"{class_name} #{self.__oid}"
+ return f"Shape #{self._oid}: {self.shapeTypeName}"
+ return f"{class_name} #{self._oid}"
def _bbox_from_points(self) -> BBox:
xs: list[float] = []
@@ -1607,7 +1936,6 @@ class _HasM(_CanHaveBBox):
raise ShapefileException(
f"Failed to write measure extremes for record {i}. Expected floats"
)
-
ms_to_encode = replace_None_with_NODATA(s.m)
try:
num_bytes_written += b_io.write(pack(f"<{len(s.m)}d", *ms_to_encode))
@@ -2067,11 +2395,11 @@ class _Record(list[RecordValue]):
:param values: A sequence of values
:param oid: The object id, an int (optional)
"""
- self.__field_positions = field_positions
+ self._field_positions = field_positions
if oid is not None:
- self.__oid = oid
+ self._oid = oid
else:
- self.__oid = -1
+ self._oid = -1
list.__init__(self, values)
def __getattr__(self, item: str) -> RecordValue:
@@ -2088,7 +2416,7 @@ class _Record(list[RecordValue]):
try:
if item == "__setstate__": # Prevent infinite loop from copy.deepcopy()
raise AttributeError("_Record does not implement __setstate__")
- index = self.__field_positions[item]
+ index = self._field_positions[item]
return list.__getitem__(self, index)
except KeyError:
raise AttributeError(f"{item} is not a field name")
@@ -2108,7 +2436,7 @@ class _Record(list[RecordValue]):
if key.startswith("_"): # Prevent infinite loop when setting mangled attribute
return list.__setattr__(self, key, value)
try:
- index = self.__field_positions[key]
+ index = self._field_positions[key]
return list.__setitem__(self, index, value)
except KeyError:
raise AttributeError(f"{key} is not a field name")
@@ -2136,7 +2464,7 @@ class _Record(list[RecordValue]):
return list.__getitem__(self, cast(Union[SupportsIndex, slice], item))
except TypeError:
try:
- index = self.__field_positions[cast(str, item)]
+ index = self._field_positions[cast(str, item)]
except KeyError:
index = None
if index is not None:
@@ -2170,7 +2498,7 @@ class _Record(list[RecordValue]):
list.__setitem__(self, *cast(ValidKVTuple, (key, value)))
return
except TypeError:
- index = self.__field_positions.get(cast(str, key))
+ index = self._field_positions.get(cast(str, key))
if index is not None:
list.__setitem__(self, index, cast(RecordValue, value))
return
@@ -2180,14 +2508,14 @@ class _Record(list[RecordValue]):
@property
def oid(self) -> int:
"""The index position of the record in the original shapefile"""
- return self.__oid
+ return self._oid
def as_dict(self, date_strings: bool = False) -> dict[str, RecordValue]:
"""
Returns this Record as a dictionary using the field names as keys
:return: dict
"""
- dct = {f: self[i] for f, i in self.__field_positions.items()}
+ dct = {f: self[i] for f, i in self._field_positions.items()}
if date_strings:
for k, v in dct.items():
if isinstance(v, date):
@@ -2195,7 +2523,7 @@ class _Record(list[RecordValue]):
return dct
def __repr__(self) -> str:
- return f"Record #{self.__oid}: {list(self)}"
+ return f"Record #{self._oid}: {list(self)}"
def __dir__(self) -> list[str]:
"""
@@ -2208,13 +2536,13 @@ class _Record(list[RecordValue]):
dir(type(self))
) # default list methods and attributes of this class
fnames = list(
- self.__field_positions.keys()
+ self._field_positions.keys()
) # plus field names (random order if Python version < 3.6)
return default + fnames
def __eq__(self, other: Any) -> bool:
if isinstance(other, _Record):
- if self.__field_positions != other.__field_positions:
+ if self._field_positions != other._field_positions:
return False
return list.__eq__(self, other)
@@ -2569,14 +2897,16 @@ class DbfReader(_HasCheckedReadableFile):
*,
encoding: str = "utf-8",
encodingErrors: str = "strict",
+ strict: bool = False,
):
super().__init__(file=dbf)
self.encoding = encoding
self.encodingErrors = encodingErrors
+ self.strict = strict
self.fields: list[Field] = []
- self.__fieldLookup: dict[str, int] = {}
+ self._fieldLookup: dict[str, int] = {}
self._dbfHeader()
@@ -2591,33 +2921,26 @@ class DbfReader(_HasCheckedReadableFile):
# read relevant header parts
self.file.seek(0)
- self.numRecords, self.__dbfHdrLength, self._record_length = cast(
+ self.numRecords, self._dbfHdrLength, self._record_length = cast(
tuple[int, int, int], unpack("<xxxxLHH20x", self.file.read(32))
)
# read fields
- numFields = (self.__dbfHdrLength - 33) // 32
- for __field in range(numFields):
- encoded_field_tuple: tuple[bytes, bytes, int, int] = unpack(
- "<11sc4xBB14x", self.file.read(32)
+ numFields = (self._dbfHdrLength - 33) // 32
+ for __ in range(numFields):
+ self.fields.append(
+ Field.from_byte_stream(
+ b_io=self.file,
+ encoding=self.encoding,
+ encodingErrors=self.encodingErrors,
+ strict=self.strict,
+ )
)
- encoded_name, encoded_type_char, size, decimal = encoded_field_tuple
-
- if b"\x00" in encoded_name:
- idx = encoded_name.index(b"\x00")
- else:
- idx = len(encoded_name) - 1
- encoded_name = encoded_name[:idx]
- name = encoded_name.decode(self.encoding, self.encodingErrors)
- name = name.lstrip()
- field_type = FIELD_TYPE_ALIASES[encoded_type_char]
-
- self.fields.append(Field(name, field_type, size, decimal))
terminator = self.file.read(1)
if terminator != b"\r":
- raise ShapefileException(
- "Shapefile dbf header lacks expected terminator. (likely corrupt?)"
+ raise dbfFileException(
+ "Dbf header lacks expected terminator. (likely corrupt?)"
)
# insert deletion field at start
@@ -2625,14 +2948,19 @@ class DbfReader(_HasCheckedReadableFile):
# store all field positions for easy lookups
# note: fieldLookup gives the index position of a field inside Reader.fields
- self.__fieldLookup = {f[0]: i for i, f in enumerate(self.fields)}
+ self._fieldLookup = {f.name: i for i, f in enumerate(self.fields)}
# by default, read all fields except the deletion flag, hence "[1:]"
# note: recLookup gives the index position of a field inside a _Record list
- fieldnames = [f[0] for f in self.fields[1:]]
+ fieldnames = [f.name for f in self.fields[1:]]
__fieldTuples, recLookup, recStruct = self._record_fields(fieldnames)
- self.__fullRecStruct = recStruct
- self.__fullRecLookup = recLookup
+ self._fullRecStruct = recStruct
+ self._fullRecLookup = recLookup
+
+ @property
+ def data_fields(self) -> list[Field]:
+ """All fields except the DeletionFlag."""
+ return self.fields[1:]
def _record_fmt(self, fields: Container[str] | None = None) -> tuple[str, int]:
"""Calculates the format and size of a .dbf record. Optional 'fields' arg
@@ -2676,21 +3004,20 @@ class DbfReader(_HasCheckedReadableFile):
recStruct = Struct(fmt)
# make sure the given fieldnames exist
for name in unique_fields:
- if name not in self.__fieldLookup or name == "DeletionFlag":
+ if name not in self._fieldLookup or name == "DeletionFlag":
raise ValueError(f'"{name}" is not a valid field name')
# fetch relevant field info tuples
fieldTuples = []
for fieldinfo in self.fields[1:]:
- name = fieldinfo[0]
- if name in unique_fields:
+ if fieldinfo.name in unique_fields:
fieldTuples.append(fieldinfo)
# store the field positions
- recLookup = {f[0]: i for i, f in enumerate(fieldTuples)}
+ recLookup = {f.name: i for i, f in enumerate(fieldTuples)}
else:
# use all the dbf fields
fieldTuples = self.fields[1:] # sans deletion flag
- recStruct = self.__fullRecStruct
- recLookup = self.__fullRecLookup
+ recStruct = self._fullRecStruct
+ recLookup = self._fullRecLookup
return fieldTuples, recLookup, recStruct
def _record(
@@ -2731,8 +3058,8 @@ class DbfReader(_HasCheckedReadableFile):
for (__name, typ, __size, decimal), value in zip(fieldTuples, recordContents):
if typ is FieldType.N or typ is FieldType.F:
# numeric or float: number stored as a string, right justified, and padded with blanks to the width of the field.
- value = value.split(b"\0")[0]
- value = value.replace(b"*", b"") # QGIS NULL is all '*' chars
+ value, __, __ = value.partition(b"\x00")
+ value = value.strip(b"*") # QGIS NULL is all '*' chars
if value == b"":
value = None
elif decimal:
@@ -2766,13 +3093,13 @@ class DbfReader(_HasCheckedReadableFile):
# but can check for all hex null-chars, all spaces, or all 0s (QGIS null)
value = None
else:
+ date_str = value.decode("ascii")
try:
# return as python date object
- y, m, d = int(value[:4]), int(value[4:6]), int(value[6:8])
- value = date(y, m, d)
+ value = datetime.strptime(date_str, "%Y%m%d").date()
except (TypeError, ValueError):
- # if invalid date, just return as unicode string so user can decimalde
- value = str(value.strip())
+ # if invalid date, just return as unicode string so user can handle it.
+ value = date_str
elif typ is FieldType.L:
# logical: 1 byte - initialized to 0x20 (space) otherwise T or F.
if value == b" ":
@@ -2785,10 +3112,11 @@ class DbfReader(_HasCheckedReadableFile):
else:
value = None # unknown value is set to missing
else:
- value = value.decode(self.encoding, self.encodingErrors)
- value = value.strip().rstrip(
- "\x00"
- ) # remove null-padding at end of strings
+ value = _decode_C_or_M_field(
+ value,
+ encoding=self.encoding,
+ encodingErrors=self.encodingErrors,
+ )
record.append(value)
return _Record(recLookup, record, oid)
@@ -2804,7 +3132,7 @@ class DbfReader(_HasCheckedReadableFile):
i = ensure_within_bounds(i, self.numRecords)
recSize = self._record_length
self.file.seek(0)
- self.file.seek(self.__dbfHdrLength + (i * recSize))
+ self.file.seek(self._dbfHdrLength + (i * recSize))
fieldTuples, recLookup, recStruct = self._record_fields(fields)
return self._record(
oid=i, fieldTuples=fieldTuples, recLookup=recLookup, recStruct=recStruct
@@ -2876,7 +3204,7 @@ class DbfReader(_HasCheckedReadableFile):
elif stop < 0:
stop = range(self.numRecords)[stop]
recSize = self._record_length
- self.file.seek(self.__dbfHdrLength + (start * recSize))
+ self.file.seek(self._dbfHdrLength + (start * recSize))
fieldTuples, recLookup, recStruct = self._record_fields(fields)
for i in range(start, stop):
r = self._record(
@@ -3408,6 +3736,11 @@ class Reader(_HasExitStack):
def fields(self) -> list[Field]:
return self.dbf_reader.fields
+ @property
+ def data_fields(self) -> list[Field]:
+ """All fields except the DeletionFlag."""
+ return self.dbf_reader.data_fields
+
def record(self, i: int = 0, fields: list[str] | None = None) -> _Record | None:
return self.dbf_reader.record(i, fields)
@@ -3832,6 +4165,7 @@ class DbfWriter(_HasCheckedWriteableFile):
encoding: str = "utf-8",
encodingErrors: str = "strict",
max_num_fields: int = 2046,
+ strict: bool = False,
# Keep kwargs even though unused, to preserve PyShp 2.4 API
**kwargs: Any,
):
@@ -3839,9 +4173,10 @@ class DbfWriter(_HasCheckedWriteableFile):
self.encoding = encoding
self.encodingErrors = encodingErrors
+ self.max_num_fields = max_num_fields
+ self.strict = strict
self.fields: list[Field] = []
- self.max_num_fields = max_num_fields
self.recNum = 0
def field(
@@ -3857,8 +4192,16 @@ class DbfWriter(_HasCheckedWriteableFile):
raise dbfFileException(
f".dbf Shapefile Writer reached maximum number of fields: {self.max_num_fields}."
)
- field_ = Field.from_unchecked(name, field_type, size, decimal)
- self.fields.append(field_)
+ field = Field.from_unchecked(
+ name=name,
+ field_type=field_type,
+ size=size,
+ decimal=decimal,
+ encoding=self.encoding,
+ encodingErrors=self.encodingErrors,
+ strict=self.strict,
+ )
+ self.fields.append(field)
def _header(self) -> None:
"""Writes the dbf header and field descriptors."""
@@ -3868,17 +4211,17 @@ class DbfWriter(_HasCheckedWriteableFile):
year, month, day = time.localtime()[:3]
year -= 1900
# Get all fields, ignoring DeletionFlag if specified
- fields = [field for field in self.fields if field[0] != "DeletionFlag"]
+ fields = [field for field in self.fields if field.name != "DeletionFlag"]
# Ensure has at least one field
if not fields:
- raise ShapefileException(
+ raise dbfFileException(
"Shapefile dbf file must contain at least one field."
)
numRecs = self.recNum
numFields = len(fields)
headerLength = numFields * 32 + 33
if headerLength >= 65535:
- raise ShapefileException(
+ raise dbfFileException(
"Shapefile dbf header length exceeds maximum length."
)
recordLength = sum(field.size for field in fields) + 1
@@ -3893,21 +4236,16 @@ class DbfWriter(_HasCheckedWriteableFile):
recordLength,
)
f.write(header)
+
# Field descriptors
for field in fields:
- encoded_name = field.name.encode(self.encoding, self.encodingErrors)
- encoded_name = encoded_name.replace(b" ", b"_")
- encoded_name = encoded_name[:10].ljust(11).replace(b" ", b"\x00")
- encodedFieldType = field.field_type.encode("ascii")
- fld = pack(
- "<11sc4xBB14x",
- encoded_name,
- encodedFieldType,
- field.size,
- field.decimal,
+ f.write(
+ field.encode_field_descriptor(
+ self.encoding, self.encodingErrors, self.strict
+ )
)
- f.write(fld)
- # Terminator
+
+ # Terminator (0x0d from dbf spec https://en.wikipedia.org/wiki/.dbf#File_header)
f.write(b"\r")
def record(
@@ -3944,23 +4282,23 @@ class DbfWriter(_HasCheckedWriteableFile):
else:
# Blank fields for empty record
record = ["" for _ in range(fieldCount)]
- self.__dbfRecord(record)
+ self._record(record)
- def __dbfRecord(self, record: list[RecordValue]) -> None:
+ def _record(self, record: list[RecordValue]) -> None:
"""Writes the dbf records."""
- f = self.file
+ record_stream = io.BytesIO() # Temporary buffer to make record writing atomic.
if self.recNum == 0:
# first records, so all fields should be set
# allowing us to write the dbf header
# cannot change the fields after this point
self._header()
# first byte of the record is deletion flag, always disabled
- f.write(b" ")
+ record_stream.write(b" ")
# begin
- self.recNum += 1
fields = (
field for field in self.fields if field[0] != "DeletionFlag"
) # ignore deletionflag field in case it was specified
+
for (fieldName, fieldType, size, deci), value in zip(fields, record):
# write
# fieldName, fieldType, size and deci were already checked
@@ -3991,17 +4329,19 @@ class DbfWriter(_HasCheckedWriteableFile):
) # caps the size if exceeds the field size
elif fieldType == "D":
# date: 8 bytes - date stored as a string in the format YYYYMMDD.
+ if isinstance(value, list) and len(value) == 3:
+ value = date(*value)
if isinstance(value, date):
- str_val = f"{value.year:04d}{value.month:02d}{value.day:02d}"
- elif isinstance(value, list) and len(value) == 3:
- str_val = f"{value[0]:04d}{value[1]:02d}{value[2]:02d}"
+ str_val = value.strftime("%Y%m%d")
elif value in MISSING:
str_val = "0" * 8 # QGIS NULL for date type
elif isinstance(value, str) and len(value) == 8:
pass # value is already a date string
else:
raise ShapefileException(
- "Date values must be either a datetime.date object, a list, a YYYYMMDD string, or a missing value."
+ f"Could not read as date: {value}. "
+ "Date values must be either a datetime.date object, "
+ "a list, a YYYYMMDD string, or a missing value."
)
elif fieldType == "L":
# logical: 1 byte - initialized to 0x20 (space) otherwise T or F.
@@ -4015,20 +4355,22 @@ class DbfWriter(_HasCheckedWriteableFile):
str_val = " " # unknown is set to space
if str_val is None:
- # Types C and M, and anything else, value is forced to string,
- # encoded by the codec specified to the Writer (utf-8 by default),
- # then the resulting bytes are padded and truncated to the length
- # of the field
- encoded = (
- str(value)
- .encode(self.encoding, self.encodingErrors)[:size]
- .ljust(size)
+ # Types C and M, and anything else, value is forced to string.
+ encoded, _trimmed = _encode_dbf_string(
+ s=str(value),
+ size=size,
+ decode=_decode_C_or_M_field,
+ pad_byte=b" ",
+ encoding=self.encoding,
+ encodingErrors=self.encodingErrors,
+ strict=self.strict,
)
else:
# str_val was given a not-None string value
# under the checks for fieldTypes "N", "F", "D", or "L" above
# Numeric, logical, and date numeric types are ascii already, but
- # for Shapefile or dbf spec reasons
+ # for Shapefile or dbf spec reasons ( "All field data is ASCII"
+ # https://en.wikipedia.org/wiki/.dbf#Database_records )
# "should be default ascii encoding"
encoded = str_val.encode("ascii", self.encodingErrors)
@@ -4037,7 +4379,10 @@ class DbfWriter(_HasCheckedWriteableFile):
f"Shapefile Writer unable to pack incorrect sized {value=}"
f" (encoded as {len(encoded)}B) into field '{fieldName}' ({size}B)."
)
- f.write(encoded)
+ record_stream.write(encoded)
+
+ self.file.write(record_stream.getvalue())
+ self.recNum += 1
class _ShpShxHeaderWriter(_HasCheckedWriteableFile):
@@ -4135,7 +4480,11 @@ class ShpWriter(_ShpWriterInfo):
def _write_file_length(self) -> None:
# self.file required to be at correct position, e.g.
# if called by self._header
- self.file.write(pack(">i", self._shp_file_length_B()))
+
+ # Calculate size as 16-bit words
+ size_B = self._shp_file_length_B()
+ size_16b_words = size_B // 2
+ self.file.write(pack(">i", size_16b_words))
def _shp_file_length_B(self) -> int:
"""Calculates the file length of the shp file."""
@@ -4144,9 +4493,7 @@ class ShpWriter(_ShpWriterInfo):
# Calculate size of all shapes
self.file.seek(0, 2)
- size_16b_words = self.file.tell()
- # Calculate size as 16-bit words
- size_B = size_16b_words // 2
+ size_B = self.file.tell()
# Return to start
self.file.seek(start_B)
return size_B
@@ -4200,6 +4547,7 @@ class ShpWriter(_ShpWriterInfo):
self,
s: Shape | HasGeoInterface | GeoJSONHomogeneousGeometryObject,
) -> tuple[int, int]:
+ """Appends s to the file. Returns shape's offset and length in B"""
if not isinstance(s, Shape):
if isinstance(s, HasGeoInterface):
shape_dict = s.__geo_interface__
@@ -4216,6 +4564,7 @@ class ShpWriter(_ShpWriterInfo):
return self._shp_record(s)
def _shp_record(self, s: Shape) -> tuple[int, int]:
+ """Appends s to the file. Returns shape's offset and length in B"""
offset = self.file.tell()
self.shpNum += 1
@@ -4274,7 +4623,7 @@ class ShpWriter(_ShpWriterInfo):
# Flush to file.
b_io.seek(0)
self.file.write(b_io.read())
- return offset, length_16bw
+ return offset, n
class ShxWriter(_ShpShxHeaderWriter):
@@ -4288,7 +4637,7 @@ class ShxWriter(_ShpShxHeaderWriter):
super().__init__(file=shx)
self.shp_writer = shp_writer
- def _shx_record(self, offset_B: int, length_16bw: int) -> None:
+ def _shx_record(self, offset_B: int, length_B: int) -> None:
"""Writes the shx records."""
f = self.file
@@ -4299,7 +4648,7 @@ class ShxWriter(_ShpShxHeaderWriter):
"It's over 4GB, perhaps split the .shp or the Shapefile into smaller ones? "
)
- offset_16bw = offset_B // 2
+ offset_16bw, length_16bw = offset_B // 2, length_B // 2
f.write(pack(">2i", offset_16bw, length_16bw))
def _header(self) -> None:
@@ -4329,6 +4678,7 @@ class Writer(_HasExitStack):
*,
encoding: str = "utf-8",
encodingErrors: str = "strict",
+ strict: bool = False,
shp: WriteSeekableBinStream | None = None,
shx: WriteSeekableBinStream | None = None,
dbf: WriteSeekableBinStream | None = None,
@@ -4360,7 +4710,7 @@ class Writer(_HasExitStack):
raise TypeError(
"Unused kwargs were silently ignored by previous versions of PyShp. "
"Either specify target (first positional only arg), "
- "or shp and/or dbf, possible plus shx"
+ "or shp and/or dbf, possibly plus shx"
)
self._shp = target.with_suffix(".shp")
self._shx = target.with_suffix(".shx")
@@ -4385,6 +4735,7 @@ class Writer(_HasExitStack):
dbf=self._dbf,
encoding=encoding,
encodingErrors=encodingErrors,
+ strict=strict,
)
self.exit_stack.enter_context(self._dbf_writer)
@@ -4454,9 +4805,9 @@ class Writer(_HasExitStack):
# Balance if already not balanced
if self.autoBalance and self.dbf_writer.recNum < self.shp_writer.shpNum:
self.balance()
- offset_B, length_16bw = self.shp_writer.shape(s)
+ offset_B, length_B = self.shp_writer.shape(s)
if self._shx:
- self.shx_writer._shx_record(offset_B, length_16bw)
+ self.shx_writer._shx_record(offset_B, length_B)
def record(
self,
=====================================
tests/hypothesis_tests.py
=====================================
@@ -1,10 +1,12 @@
from __future__ import annotations
+import datetime
import io
import itertools
+import string
import pytest
-from hypothesis import HealthCheck, given, settings
+from hypothesis import HealthCheck, given, settings, reproduce_failure
from hypothesis.strategies import (
builds,
composite, # Preferably avoid. Shrinking composite strategies is slow.
@@ -16,6 +18,9 @@ from hypothesis.strategies import (
one_of,
tuples,
sampled_from,
+ text,
+ characters,
+ dates,
)
import shapefile as shp
@@ -27,6 +32,7 @@ ms = one_of(none(), float_nums)
zs = one_of(just(0.0), float_nums)
PointsLengths = integers(min_value=1, max_value=8000) # length of points
oid = one_of(none(), integers(min_value=0))
+null_shapes = builds(shp.NullShape, oid=oid)
point_2D = builds(shp.Point, x=xs, y=ys, oid=oid)
pointm = builds(
shp.PointM,
@@ -163,7 +169,6 @@ def multipointM_from_xyms(point_ms: tuple[float, float, float | None], oid_: int
multipointm = builds(multipointM_from_xyms, lists(tuples(xs, ys, ms), min_size=1), oid)
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(expected=multipointm, i=integers(min_value=1))
def test_MultiPointM_roundtrips(
expected: shp.MultiPointM,
@@ -196,7 +201,6 @@ multipointz = builds(multipointZ_from_xyzms, lists(tuples(xs, ys, zs, ms), min_s
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(expected=multipointz, i=integers(min_value=1))
def test_MultiPointZ_roundtrips(
expected: shp.MultiPointZ,
@@ -248,7 +252,6 @@ def test_Polyline_roundtrips(
assert actual.oid == expected.oid
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(expected=polylinem, i=integers(min_value=1))
def test_PolylineM_roundtrips(
expected: shp.PolylineM,
@@ -273,7 +276,6 @@ def test_PolylineM_roundtrips(
assert actual.oid == expected.oid
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(expected=polylinez, i=integers(min_value=1))
def test_PolylineZ_roundtrips(
expected: shp.PolylineZ,
@@ -327,7 +329,6 @@ def test_Polygon_roundtrips(
assert actual.oid == expected.oid
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(expected=polygonm, i=integers(min_value=1))
def test_PolygonM_roundtrips(
expected: shp.PolygonM,
@@ -352,7 +353,6 @@ def test_PolygonM_roundtrips(
assert actual.oid == expected.oid
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(expected=polygonz, i=integers(min_value=1))
def test_PolygonZ_roundtrips(
expected: shp.PolygonZ,
@@ -392,7 +392,6 @@ multipatch = builds(
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(expected=multipatch, i=integers(min_value=1))
def test_MultiPatch_roundtrips(
expected: shp.MultiPatch,
@@ -418,9 +417,15 @@ def test_MultiPatch_roundtrips(
assert actual.oid == expected.oid
assert actual.partTypes == expected.partTypes, f"{type(actual.partTypes)=}, {type(expected.partTypes)=}"
+MAX_FILE_SIZE_16bw = (1 << 31) - 1 # This bound comes from encoding the
+ # actual file size (in 16 bit words)
+ # as a 4 byte signed integer.
+MAX_NUM_SHAPES = (MAX_FILE_SIZE_16bw - 50) // 6 # Minus 100B header, 12 bytes
+ # per record (the minimum for
+ # a Null shape).
shape_codes_names_and_strategies = [
-# (0, "Null Shape"),
+(0, "Null Shape", null_shapes),
(1, "Point", point_2D),
(3, "PolyLine", polyline),
(5, "Polygon", polygon),
@@ -438,8 +443,14 @@ shape_codes_names_and_strategies = [
def code_and_shape_strat_from_triple(t):
x, _name, shapes = t
- return tuples(just(x), lists(shapes, min_size = 0)) # Empty shp files are in the esri spec.
-
+ return tuples(
+ just(x),
+ lists(
+ one_of(shapes, null_shapes),
+ min_size = 0, # Empty shp files are in the ESRI spec.
+ max_size=MAX_NUM_SHAPES,
+ ),
+ )
codes_and_shapes_strats = [
code_and_shape_strat_from_triple(t)
for t in shape_codes_names_and_strategies
@@ -448,21 +459,22 @@ codes_and_shapes_strats = [
codes_and_shapes = one_of(codes_and_shapes_strats)
@pytest.mark.hypothesis
-# @settings(suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large])
@given(codes_and_shapes=codes_and_shapes)
def test_shp_reader_writer_roundtrip(codes_and_shapes)-> None:
+
code_ex, expected_shapes = codes_and_shapes
stream = io.BytesIO()
+
with shp.ShpWriter(shp=stream, shapeType=code_ex) as w:
for shape in expected_shapes:
w.shape(shape)
- stream.seek(0)
+
with shp.ShpReader(shp=stream) as r:
assert r.shapeType == code_ex
for actual, expected in itertools.zip_longest(r.shapes(), expected_shapes):
- assert isinstance(actual, shp.SHAPE_CLASS_FROM_SHAPETYPE[code_ex])
+ assert isinstance(actual, (shp.SHAPE_CLASS_FROM_SHAPETYPE[code_ex], shp.NullShape))
assert actual.points_3D == expected.points_3D
# Don't assert actual.oid == expected.oid it's defined by
# actual.oid indicates the order actual was written in, expected.oid
@@ -483,3 +495,180 @@ def test_shp_reader_writer_roundtrip(codes_and_shapes)-> None:
assert actual.partTypes == expected.partTypes, f"{type(actual.partTypes)=}, {type(expected.partTypes)=}"
else:
assert not hasattr(expected, "partTypes")
+
+
+ at pytest.mark.hypothesis
+ at given(codes_and_shapes=codes_and_shapes)
+def test_shx_reader_writer_roundtrip(codes_and_shapes)-> None:
+ code_ex, expected_shapes = codes_and_shapes
+
+ sizes_B = []
+ offsets_B = []
+ offset_B = 100 # "Thus, the offset for the first record in the
+ # main file is 50 (16bw), given the 100-byte header. "
+ shp_stream = io.BytesIO()
+ shx_stream = io.BytesIO()
+ with shp.ShpWriter(shp=shp_stream, shapeType=code_ex) as shp_w:
+ with shp.ShxWriter(shx=shx_stream, shp_writer = shp_w) as shx_w:
+ for shape in expected_shapes:
+ offset_B, size_B = shp_w.shape(shape)
+ sizes_B.append(size_B)
+ offsets_B.append(offset_B)
+ shx_w._shx_record(offset_B, size_B)
+
+ with shp.ShxReader(shx=shx_stream) as r:
+ assert r.numShapes == len(expected_shapes)
+ assert r.offsets == offsets_B
+ assert r.shape_lengths_B == sizes_B
+
+
+
+DBF_FIELD_TYPES = {
+ "C": {},
+ "N": {"max_decimal" : 20, "max_length": 22}, # max length=23 to avoid error due to precision limit, e.g.:
+ "F": {"max_decimal" : 20, "max_length": 22}, # hypothesis.errors.InvalidArgument: max_value=100000000000000000000000
+ # cannot be exactly represented as a float of
+ # width 64 - use max_value=1e+23 instead.
+ "L": {"max_length": 1},
+ "D": {"min_length": 8, "max_length": 8},
+}
+
+ at composite
+def dbf_fields(draw):
+ field_type, bounds_dict = draw(sampled_from(list(DBF_FIELD_TYPES.items())))
+
+ name = draw(
+ text(
+ alphabet=characters(
+ codec="ascii",
+ exclude_categories=["Z", "C"] # Z - Whitespace, C - Control chars++
+ ),
+ min_size=1,
+ max_size=10,
+ )
+ )
+
+ max_length = bounds_dict.get("max_length", 254)
+ min_length = bounds_dict.get("min_length", 1)
+ max_decimal = bounds_dict.get("max_decimal", 0)
+ size = draw(integers(min_value=min_length, max_value=max_length))
+ decimal = draw(integers(min_value=0, max_value=max(0,min(size - 3, max_decimal))))
+
+
+ return {"name": name, "field_type": field_type, "size": size, "decimal": decimal}
+
+
+ at pytest.mark.hypothesis
+ at given(field_kwargs=dbf_fields())
+def test_dbf_Field_roundtrips(
+ field_kwargs: dict,
+) -> None:
+ expected = shp.Field.from_unchecked(**field_kwargs)
+ stream = io.BytesIO()
+ encoded = expected.encode_field_descriptor(strict=True)
+ stream.write(encoded)
+ stream.seek(0)
+ actual = shp.Field.from_byte_stream(stream, strict=True)
+ assert isinstance(actual, shp.Field)
+ assert actual.name == expected.name
+ assert actual[1:] == expected[1:]
+
+
+ascii_printable = string.ascii_letters + string.digits + string.punctuation + " "
+
+def record_value_for_field(name: str, field_type: str, size: int, decimal: int = 0):
+
+ if field_type == "C":
+ return text(
+ alphabet=ascii_printable,
+ min_size=0,
+ max_size=size,
+ )
+ if field_type in {"N", "F"}:
+
+ int_digits = size if decimal == 0 else size - decimal - 1
+ min_int = -(10 ** (int_digits - 1) - 1)
+ max_int = 10 ** int_digits - 1
+
+ if decimal == 0:
+ return integers(min_value=min_int, max_value=max_int)
+
+ # Max finite float: 2**1023 * (2 - 2**(-52))
+ return floats(
+ min_value=min_int - 1,
+ max_value=max_int + 1,
+ exclude_min=True,
+ exclude_max=True,
+ )
+ if field_type == "L":
+ return sampled_from([True, False, None])
+ if field_type == "D":
+ return one_of(dates(), dates().map(lambda d: d.strftime("%Y%m%d")))
+
+ raise ValueError(f"Unsupported: {field_type=}")
+
+
+def _dbf_fields_and_record_strategy(
+ draw,
+ max_fields=10, # In DbfWriter.__init__, max_num_fields: int = 2046,
+ max_records=20,
+ ):
+
+ fields = draw(lists(dbf_fields(), min_size=1, max_size=max_fields))
+
+ record_strategy = tuples(*(record_value_for_field(**field) for field in fields))
+
+ return fields, record_strategy
+
+
+ at composite
+def dbf_fields_and_records(
+ draw,
+ max_fields=10, # In DbfWriter.__init__, max_num_fields: int = 2046,
+ max_records=20,
+ ):
+
+ fields, record_strategy = _dbf_fields_and_record_strategy(draw, max_fields, max_records)
+
+ records = draw(lists(record_strategy, min_size=0, max_size=max_records))
+
+ return fields, records
+
+
+ at pytest.mark.hypothesis
+ at given(fields_and_records=dbf_fields_and_records())
+def test_dbf_reader_writer_roundtrip(fields_and_records)-> None:
+ fields, records = fields_and_records
+ stream = io.BytesIO()
+ written_records = []
+ with shp.DbfWriter(dbf=stream, strict=True) as dbf_w:
+ for field in fields:
+ dbf_w.field(**field)
+ for record in records:
+ try:
+ dbf_w.record(*record)
+ except shp.DbfStringDataLoss:
+ pass
+ else:
+ written_records.append(record)
+
+
+ with shp.DbfReader(dbf=stream) as r:
+ actual_fields = iter(r.fields)
+ next(actual_fields) # skip deletion flag
+ for f_r, f_w in itertools.zip_longest(actual_fields, fields):
+ actual_field_dict = f_r._asdict()
+ for k in ("field_type", "size", "decimal"):
+ assert actual_field_dict[k] == f_w[k], f"{k=}, {actual_field_dict[k]=}, {f_w[k]=}"
+ for exp_rec, actual_rec in itertools.zip_longest(written_records, r.records()):
+ for expected, actual, field in itertools.zip_longest(exp_rec, actual_rec, fields):
+ field_type = field["field_type"]
+ decimal = field["decimal"]
+ if field_type == "D":
+ if isinstance(expected, datetime.date):
+ expected = expected.strftime("%Y%m%d")
+ if isinstance(actual, datetime.date):
+ actual = actual.strftime("%Y%m%d")
+ elif field_type in ("N", "F") and decimal >= 1:
+ expected = float(format(expected, f".{decimal}f"))
+ assert actual == expected, f"{actual=}, {expected=}, {field_type=}, {type(actual)=}, {type(expected)=}"
=====================================
tests/run_benchmarks.py
=====================================
@@ -10,7 +10,7 @@ from collections.abc import Callable, Iterable
from os import PathLike
from pathlib import Path
from tempfile import TemporaryFile as TempF
-from typing import cast
+from typing import cast, Iterable
import shapefile
@@ -55,14 +55,14 @@ shapeRecords: dict[str | PathLike, list[shapefile.shapeRecord]] = (
def open_shapefile_with_PyShp(target: str | PathLike):
with shapefile.Reader(target) as r:
- fields[target] = r.fields
+ fields[target] = r.data_fields
for shapeRecord in r.iterShapeRecords():
shapeRecords[target].append(shapeRecord)
def write_shapefile_with_PyShp(target: str | PathLike):
with TempF("wb") as shp, TempF("wb") as dbf, TempF("wb") as shx:
- with shapefile.Writer(shp=shp, dbf=dbf, shx=shx) as w: # type: ignore [arg-type]
+ with shapefile.Writer(shp=shp, dbf=dbf, shx=shx, strict=False) as w: # type: ignore [arg-type]
for field_info_tuple in fields[target]:
w.field(*field_info_tuple)
for shapeRecord in shapeRecords[target]:
@@ -87,32 +87,32 @@ for file_path in SHAPEFILES.values():
COLS_WIDTHS = (22, 10)
-reader_benchmarks = [
- functools.partial(
+reader_benchmarks = {
+ test_name : functools.partial(
benchmark,
name=f"Read {test_name}",
func=functools.partial(open_shapefile_with_PyShp, target=target),
col_widths=COLS_WIDTHS,
)
for test_name, target in SHAPEFILES.items()
-]
+}
# Require fields and shapeRecords to first have been populated
# from data from previouly running the reader_benchmarks
-writer_benchmarks = [
- functools.partial(
+writer_benchmarks = {
+ test_name : functools.partial(
benchmark,
name=f"Write {test_name}",
func=functools.partial(write_shapefile_with_PyShp, target=target),
col_widths=COLS_WIDTHS,
)
for test_name, target in SHAPEFILES.items()
-]
+}
def run(
run_count: int,
- benchmarks: list[Callable[[], None]],
+ benchmarks: Iterable[Callable[[], None]],
col_widths: tuple[int, int] = COLS_WIDTHS,
) -> None:
col_head = ("parser", "exec time", "performance (more is better)")
@@ -125,9 +125,11 @@ def run(
run_count=run_count,
)
-
-if __name__ == "__main__":
+def main():
print("Reader tests:")
- run(1, reader_benchmarks) # type: ignore [arg-type]
+ run(1, reader_benchmarks.values()) # type: ignore [arg-type]
print("\n\nWriter tests:")
- run(1, writer_benchmarks) # type: ignore [arg-type]
+ run(1, writer_benchmarks.values()) # type: ignore [arg-type]
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
=====================================
tests/test_shapefile.py
=====================================
@@ -2,7 +2,9 @@
This module tests the functionality of shapefile.py.
"""
+import contextlib
import datetime
+import io
import json
import os.path
from pathlib import Path
@@ -1561,25 +1563,39 @@ def test_reader_zip_polyylinez_no_m_itershaperecords():
pass
-def test_write_field_name_limit(tmpdir):
- """
- Abc...
- """
+def test_write_field_name_below_limit(tmpdir):
filename = tmpdir.join("test.shp").strpath
- with shapefile.Writer(filename) as writer:
+ with shapefile.Writer(filename, strict=True) as writer:
writer.field("a" * 5, "C") # many under length limit
writer.field("a" * 9, "C") # 1 under length limit
- writer.field("a" * 10, "C") # at length limit
- writer.field("a" * 11, "C") # 1 over length limit
- writer.field("a" * 20, "C") # many over limit
with shapefile.Reader(filename) as reader:
fields = reader.fields[1:]
assert len(fields[0][0]) == 5
assert len(fields[1][0]) == 9
+
+def test_write_field_names_above_limit_non_strict(tmpdir):
+ filename = tmpdir.join("test.shp").strpath
+ with shapefile.Writer(filename, strict=False) as w:
+ w.field("a" * 10, "C") # at length limit
+ for l in [11, 20]: # 1 over, and twice the limit
+ with pytest.warns(shapefile.PossibleDataLoss):
+ w.field("a" * l, "C")
+
+ with shapefile.Reader(filename) as reader:
+ fields = reader.fields[1:]
+ assert len(fields[0][0]) == 10
+ assert len(fields[1][0]) == 10
assert len(fields[2][0]) == 10
- assert len(fields[3][0]) == 10
- assert len(fields[4][0]) == 10
+
+def test_write_field_names_above_limit_strict(tmpdir):
+ filename = tmpdir.join("test.shp").strpath
+ with shapefile.Writer(filename, strict=True) as writer:
+ writer.field("a" * 10, "C") # at length limit
+ for l in [11, 20]: # at 1 over length limitand twice the limit
+ with pytest.raises(ValueError):
+ writer.field("a" * l, "C")
+
def test_write_shp_only(tmpdir):
@@ -2021,3 +2037,119 @@ def test_write_multipatch(tmpdir):
w.record("house1")
w.close()
+
+DATES = [datetime.date(*triple) for triple in [
+ (2000,1,1),
+]]
+
+ at pytest.mark.parametrize("expected_date", DATES)
+def test_round_trip_dbf_date_record(expected_date):
+ stream = io.BytesIO()
+ dbf_w = shapefile.DbfWriter(dbf=stream)
+ dbf_w.field("Date","D")
+ dbf_w.record(expected_date)
+ dbf_w.close()
+
+ dbf_r = shapefile.DbfReader(dbf=stream)
+ dbf_r.record(0)[0] == expected_date
+ dbf_r.close()
+
+
+LONG_FIELD_NAMES = [
+ ("ÀÀÀÀ०", 8, "utf-8", "strict"), # Encoded bytes are corrupted if truncated to 10 bytes
+]
+
+ at pytest.mark.parametrize("name,encoded_len,codec,errors", LONG_FIELD_NAMES)
+def test_encode_dbf_field_name_truncation(name,encoded_len,codec,errors):
+ stream = io.BytesIO()
+ w = shapefile.DbfWriter(
+ stream,
+ encoding=codec,
+ encodingErrors=errors,
+ strict = False,
+ )
+ with pytest.warns(shapefile.PossibleDataLoss):
+ w.field(name=name)
+ field = w.fields[0]
+ assert name.startswith(field.name)
+ assert len(w.fields[0].name.encode(codec, errors)) == encoded_len
+ w.close()
+
+ r = shapefile.DbfReader(stream, encoding=codec, encodingErrors=errors, strict=False)
+ assert r.fields[1].name == field.name
+ r.close()
+
+
+TEST_ENCODING_WARNINGS_FIELD_NAMES = [
+ ("A", 2, "utf-16-le", "strict"), # Encoded bytes end in null byte (second byte in low end UTF16 code unit)
+ ("ABC", 6, "utf-16-le", "strict"), # Encoded bytes end in null byte (second byte in low end UTF16 code unit)
+ ("ABCDE", 10, "utf-16-le", "strict"), # Encoded bytes end in null byte (second byte in low end UTF16 code unit)
+]
+
+ at pytest.mark.parametrize("name,encoded_len,codec,errors", TEST_ENCODING_WARNINGS_FIELD_NAMES)
+def test_encode_dbf_field_name_padding(name,encoded_len,codec,errors):
+ stream = io.BytesIO()
+ w = shapefile.DbfWriter(
+ stream,
+ encoding=codec,
+ encodingErrors=errors,
+ strict = True,
+ )
+ w.field(name=name)
+ field = w.fields[0]
+ assert name.startswith(field.name)
+ assert len(w.fields[0].name.encode(codec, errors)) == encoded_len
+ w.close()
+
+ with pytest.warns(shapefile.PossibleDataLoss):
+ r = shapefile.DbfReader(stream, encoding=codec, encodingErrors=errors, strict=False)
+ assert r.fields[1].name == field.name
+ r.close()
+
+NON_ASCII_FIELD_NAMES = [
+ ("囊萤映雪", 8, 'utf-16-be', "strict"), # Issue 421. Encoded bytes contain an ascii space (0x20) so by applying
+ # encoded.replace(b" ",b"_") the text is corrupted from
+ # "囊萤映雪" ("Studying by the light of fireflies and snow")
+ # to: "囊萤晟雪" ("Gathering Fireflies and Flourishing Snow")
+ # (English translation from Google Translate).
+]
+
+ at pytest.mark.parametrize("name,encoded_len,codec,errors", NON_ASCII_FIELD_NAMES)
+def test_encode_dbf_field_name_corruption(name,encoded_len,codec,errors):
+ stream = io.BytesIO()
+ w = shapefile.DbfWriter(
+ stream,
+ encoding=codec,
+ encodingErrors=errors,
+ strict = True,
+ )
+ w.field(name=name)
+ field = w.fields[0]
+ assert name.startswith(field.name)
+ assert len(w.fields[0].name.encode(codec, errors)) == encoded_len
+ w.close()
+
+ r = shapefile.DbfReader(stream, encoding=codec, encodingErrors=errors, strict=False)
+ assert r.fields[1].name == field.name
+ r.close()
+
+TEST_STR_VALUES = LONG_FIELD_NAMES + TEST_ENCODING_WARNINGS_FIELD_NAMES + NON_ASCII_FIELD_NAMES
+
+ at pytest.mark.parametrize("value,encoded_len,codec,errors", TEST_STR_VALUES)
+def test_encode_dbf_field_values(value,encoded_len,codec,errors):
+ stream = io.BytesIO()
+ w = shapefile.DbfWriter(
+ stream,
+ encoding=codec,
+ encodingErrors=errors,
+ strict = False,
+ )
+ w.field("name", "C")
+ w.record(value)
+ w.close()
+ WARNS = codec.lower() == "utf-16-le" and value.isascii()
+ context = pytest.warns(shapefile.PossibleDataLoss) if WARNS else contextlib.nullcontext()
+ with context:
+ r = shapefile.DbfReader(stream, encoding=codec, encodingErrors=errors, strict=False)
+ assert r.record(0)[0] == value
+ r.close()
\ No newline at end of file
View it on GitLab: https://salsa.debian.org/debian-gis-team/pyshp/-/compare/2bbf00cf8216ed6802bc3db3b128c52719950575...7ed26c4481474cf984d17e185176acccd6f088f1
--
View it on GitLab: https://salsa.debian.org/debian-gis-team/pyshp/-/compare/2bbf00cf8216ed6802bc3db3b128c52719950575...7ed26c4481474cf984d17e185176acccd6f088f1
You're receiving this email because of your account on salsa.debian.org. Manage all notifications: https://salsa.debian.org/-/profile/notifications | Help: https://salsa.debian.org/help
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://alioth-lists.debian.net/pipermail/pkg-grass-devel/attachments/20260625/edc3d846/attachment-0001.htm>
More information about the Pkg-grass-devel
mailing list