Use IO[bytes] instead of BinaryIO wherever possible
This commit is contained in:
parent
3016aec3df
commit
406550cfd6
@ -2,7 +2,7 @@
|
||||
This module contains all datatypes and parsing/writing functions for
|
||||
all abstractions below the 'record' or 'block' level.
|
||||
"""
|
||||
from typing import List, Tuple, Type, Union, Optional, Any, Sequence
|
||||
from typing import List, Tuple, Type, Union, Optional, Any, Sequence, IO
|
||||
from fractions import Fraction
|
||||
from enum import Enum
|
||||
import math
|
||||
@ -24,6 +24,7 @@ except ImportError:
|
||||
real_t = Union[int, float, Fraction]
|
||||
repetition_t = Union['ReuseRepetition', 'GridRepetition', 'ArbitraryRepetition']
|
||||
property_value_t = Union[int, bytes, 'AString', 'NString', 'PropStringReference', float, Fraction]
|
||||
bytes_t = bytes
|
||||
|
||||
|
||||
class FatamorganaError(Exception):
|
||||
@ -85,7 +86,7 @@ MAGIC_BYTES: bytes = b'%SEMI-OASIS\r\n'
|
||||
'''
|
||||
Basic IO
|
||||
'''
|
||||
def _read(stream: io.BufferedIOBase, n: int) -> bytes:
|
||||
def _read(stream: IO[bytes], n: int) -> bytes:
|
||||
"""
|
||||
Read n bytes from the stream.
|
||||
Raise an EOFError if there were not enough bytes in the stream.
|
||||
@ -106,7 +107,7 @@ def _read(stream: io.BufferedIOBase, n: int) -> bytes:
|
||||
return b
|
||||
|
||||
|
||||
def read_byte(stream: io.BufferedIOBase) -> int:
|
||||
def read_byte(stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Read a single byte and return it.
|
||||
|
||||
@ -119,7 +120,7 @@ def read_byte(stream: io.BufferedIOBase) -> int:
|
||||
return _read(stream, 1)[0]
|
||||
|
||||
|
||||
def write_byte(stream: io.BufferedIOBase, n: int) -> int:
|
||||
def write_byte(stream: IO[bytes], n: int) -> int:
|
||||
"""
|
||||
Write a single byte to the stream.
|
||||
|
||||
@ -132,7 +133,7 @@ def write_byte(stream: io.BufferedIOBase, n: int) -> int:
|
||||
return stream.write(bytes((n,)))
|
||||
|
||||
|
||||
def _py_read_bool_byte(stream: io.BufferedIOBase) -> List[bool]:
|
||||
def _py_read_bool_byte(stream: IO[bytes]) -> List[bool]:
|
||||
"""
|
||||
Read a single byte from the stream, and interpret its bits as
|
||||
a list of 8 booleans.
|
||||
@ -147,7 +148,7 @@ def _py_read_bool_byte(stream: io.BufferedIOBase) -> List[bool]:
|
||||
bits = [bool((byte >> i) & 0x01) for i in reversed(range(8))]
|
||||
return bits
|
||||
|
||||
def _py_write_bool_byte(stream: io.BufferedIOBase, bits: Tuple[Union[bool, int], ...]) -> int:
|
||||
def _py_write_bool_byte(stream: IO[bytes], bits: Tuple[Union[bool, int], ...]) -> int:
|
||||
"""
|
||||
Pack 8 booleans into a byte, and write it to the stream.
|
||||
|
||||
@ -170,7 +171,7 @@ def _py_write_bool_byte(stream: io.BufferedIOBase, bits: Tuple[Union[bool, int],
|
||||
|
||||
|
||||
if _USE_NUMPY:
|
||||
def _np_read_bool_byte(stream: io.BufferedIOBase) -> NDArray[numpy.uint8]:
|
||||
def _np_read_bool_byte(stream: IO[bytes]) -> NDArray[numpy.uint8]:
|
||||
"""
|
||||
Read a single byte from the stream, and interpret its bits as
|
||||
a list of 8 booleans.
|
||||
@ -184,7 +185,7 @@ if _USE_NUMPY:
|
||||
byte_arr = _read(stream, 1)
|
||||
return numpy.unpackbits(numpy.frombuffer(byte_arr, dtype=numpy.uint8))
|
||||
|
||||
def _np_write_bool_byte(stream: io.BufferedIOBase, bits: Tuple[Union[bool, int], ...]) -> int:
|
||||
def _np_write_bool_byte(stream: IO[bytes], bits: Tuple[Union[bool, int], ...]) -> int:
|
||||
"""
|
||||
Pack 8 booleans into a byte, and write it to the stream.
|
||||
|
||||
@ -208,7 +209,7 @@ else:
|
||||
read_bool_byte = _py_read_bool_byte # type: ignore
|
||||
write_bool_byte = _py_write_bool_byte
|
||||
|
||||
def read_uint(stream: io.BufferedIOBase) -> int:
|
||||
def read_uint(stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Read an unsigned integer from the stream.
|
||||
|
||||
@ -234,7 +235,7 @@ def read_uint(stream: io.BufferedIOBase) -> int:
|
||||
return result
|
||||
|
||||
|
||||
def write_uint(stream: io.BufferedIOBase, n: int) -> int:
|
||||
def write_uint(stream: IO[bytes], n: int) -> int:
|
||||
"""
|
||||
Write an unsigned integer to the stream.
|
||||
See format details in `read_uint()`.
|
||||
@ -297,7 +298,7 @@ def encode_sint(sint: int) -> int:
|
||||
return (abs(sint) << 1) | (sint < 0)
|
||||
|
||||
|
||||
def read_sint(stream: io.BufferedIOBase) -> int:
|
||||
def read_sint(stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Read a signed integer from the stream.
|
||||
See `decode_sint()` for format details.
|
||||
@ -311,7 +312,7 @@ def read_sint(stream: io.BufferedIOBase) -> int:
|
||||
return decode_sint(read_uint(stream))
|
||||
|
||||
|
||||
def write_sint(stream: io.BufferedIOBase, n: int) -> int:
|
||||
def write_sint(stream: IO[bytes], n: int) -> int:
|
||||
"""
|
||||
Write a signed integer to the stream.
|
||||
See `decode_sint()` for format details.
|
||||
@ -326,7 +327,7 @@ def write_sint(stream: io.BufferedIOBase, n: int) -> int:
|
||||
return write_uint(stream, encode_sint(n))
|
||||
|
||||
|
||||
def read_bstring(stream: io.BufferedIOBase) -> bytes:
|
||||
def read_bstring(stream: IO[bytes]) -> bytes:
|
||||
"""
|
||||
Read a binary string from the stream.
|
||||
The format is:
|
||||
@ -343,7 +344,7 @@ def read_bstring(stream: io.BufferedIOBase) -> bytes:
|
||||
return _read(stream, length)
|
||||
|
||||
|
||||
def write_bstring(stream: io.BufferedIOBase, bstring: bytes):
|
||||
def write_bstring(stream: IO[bytes], bstring: bytes):
|
||||
"""
|
||||
Write a binary string to the stream.
|
||||
See `read_bstring()` for format details.
|
||||
@ -359,7 +360,7 @@ def write_bstring(stream: io.BufferedIOBase, bstring: bytes):
|
||||
return stream.write(bstring)
|
||||
|
||||
|
||||
def read_ratio(stream: io.BufferedIOBase) -> Fraction:
|
||||
def read_ratio(stream: IO[bytes]) -> Fraction:
|
||||
"""
|
||||
Read a ratio (unsigned) from the stream.
|
||||
The format is:
|
||||
@ -377,7 +378,7 @@ def read_ratio(stream: io.BufferedIOBase) -> Fraction:
|
||||
return Fraction(numer, denom)
|
||||
|
||||
|
||||
def write_ratio(stream: io.BufferedIOBase, r: Fraction) -> int:
|
||||
def write_ratio(stream: IO[bytes], r: Fraction) -> int:
|
||||
"""
|
||||
Write an unsigned ratio to the stream.
|
||||
See `read_ratio()` for format details.
|
||||
@ -399,7 +400,7 @@ def write_ratio(stream: io.BufferedIOBase, r: Fraction) -> int:
|
||||
return size
|
||||
|
||||
|
||||
def read_float32(stream: io.BufferedIOBase) -> float:
|
||||
def read_float32(stream: IO[bytes]) -> float:
|
||||
"""
|
||||
Read a 32-bit float from the stream.
|
||||
|
||||
@ -413,7 +414,7 @@ def read_float32(stream: io.BufferedIOBase) -> float:
|
||||
return struct.unpack("<f", b)[0]
|
||||
|
||||
|
||||
def write_float32(stream: io.BufferedIOBase, f: float) -> int:
|
||||
def write_float32(stream: IO[bytes], f: float) -> int:
|
||||
"""
|
||||
Write a 32-bit float to the stream.
|
||||
|
||||
@ -428,7 +429,7 @@ def write_float32(stream: io.BufferedIOBase, f: float) -> int:
|
||||
return stream.write(b)
|
||||
|
||||
|
||||
def read_float64(stream: io.BufferedIOBase) -> float:
|
||||
def read_float64(stream: IO[bytes]) -> float:
|
||||
"""
|
||||
Read a 64-bit float from the stream.
|
||||
|
||||
@ -442,7 +443,7 @@ def read_float64(stream: io.BufferedIOBase) -> float:
|
||||
return struct.unpack("<d", b)[0]
|
||||
|
||||
|
||||
def write_float64(stream: io.BufferedIOBase, f: float) -> int:
|
||||
def write_float64(stream: IO[bytes], f: float) -> int:
|
||||
"""
|
||||
Write a 64-bit float to the stream.
|
||||
|
||||
@ -457,7 +458,7 @@ def write_float64(stream: io.BufferedIOBase, f: float) -> int:
|
||||
return stream.write(b)
|
||||
|
||||
|
||||
def read_real(stream: io.BufferedIOBase, real_type: Optional[int] = None) -> real_t:
|
||||
def read_real(stream: IO[bytes], real_type: Optional[int] = None) -> real_t:
|
||||
"""
|
||||
Read a real number from the stream.
|
||||
|
||||
@ -508,7 +509,7 @@ def read_real(stream: io.BufferedIOBase, real_type: Optional[int] = None) -> rea
|
||||
|
||||
|
||||
def write_real(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
r: real_t,
|
||||
force_float32: bool = False
|
||||
) -> int:
|
||||
@ -594,7 +595,7 @@ class NString:
|
||||
self._string = bstring.decode('ascii')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'NString':
|
||||
def read(stream: IO[bytes_t]) -> 'NString':
|
||||
"""
|
||||
Create an NString object by reading a bstring from the provided stream.
|
||||
|
||||
@ -609,7 +610,7 @@ class NString:
|
||||
"""
|
||||
return NString(read_bstring(stream))
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes_t]) -> int:
|
||||
"""
|
||||
Write this NString to a stream.
|
||||
|
||||
@ -631,7 +632,7 @@ class NString:
|
||||
return self._string
|
||||
|
||||
|
||||
def read_nstring(stream: io.BufferedIOBase) -> str:
|
||||
def read_nstring(stream: IO[bytes]) -> str:
|
||||
"""
|
||||
Read a name string from the provided stream.
|
||||
See `NString` for constraints on name strings.
|
||||
@ -648,7 +649,7 @@ def read_nstring(stream: io.BufferedIOBase) -> str:
|
||||
return NString.read(stream).string
|
||||
|
||||
|
||||
def write_nstring(stream: io.BufferedIOBase, string: str) -> int:
|
||||
def write_nstring(stream: IO[bytes], string: str) -> int:
|
||||
"""
|
||||
Write a name string to a stream.
|
||||
See `NString` for constraints on name strings.
|
||||
@ -708,7 +709,7 @@ class AString:
|
||||
self._string = bstring.decode('ascii')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'AString':
|
||||
def read(stream: IO[bytes_t]) -> 'AString':
|
||||
"""
|
||||
Create an `AString` object by reading a bstring from the provided stream.
|
||||
|
||||
@ -723,7 +724,7 @@ class AString:
|
||||
"""
|
||||
return AString(read_bstring(stream))
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes_t]) -> int:
|
||||
"""
|
||||
Write this `AString` to a stream.
|
||||
|
||||
@ -745,7 +746,7 @@ class AString:
|
||||
return self._string
|
||||
|
||||
|
||||
def read_astring(stream: io.BufferedIOBase) -> str:
|
||||
def read_astring(stream: IO[bytes]) -> str:
|
||||
"""
|
||||
Read an ASCII string from the provided stream.
|
||||
See `AString` for constraints on ASCII strings.
|
||||
@ -762,7 +763,7 @@ def read_astring(stream: io.BufferedIOBase) -> str:
|
||||
return AString.read(stream).string
|
||||
|
||||
|
||||
def write_astring(stream: io.BufferedIOBase, string: str) -> int:
|
||||
def write_astring(stream: IO[bytes], string: str) -> int:
|
||||
"""
|
||||
Write an ASCII string to a stream.
|
||||
See AString for constraints on ASCII strings.
|
||||
@ -853,7 +854,7 @@ class ManhattanDelta:
|
||||
return d
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'ManhattanDelta':
|
||||
def read(stream: IO[bytes]) -> 'ManhattanDelta':
|
||||
"""
|
||||
Read a `ManhattanDelta` object from the provided stream.
|
||||
|
||||
@ -868,7 +869,7 @@ class ManhattanDelta:
|
||||
n = read_uint(stream)
|
||||
return ManhattanDelta.from_uint(n)
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write a `ManhattanDelta` object to the provided stream.
|
||||
|
||||
@ -989,7 +990,7 @@ class OctangularDelta:
|
||||
return d
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'OctangularDelta':
|
||||
def read(stream: IO[bytes]) -> 'OctangularDelta':
|
||||
"""
|
||||
Read an `OctangularDelta` object from the provided stream.
|
||||
|
||||
@ -1004,7 +1005,7 @@ class OctangularDelta:
|
||||
n = read_uint(stream)
|
||||
return OctangularDelta.from_uint(n)
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write an `OctangularDelta` object to the provided stream.
|
||||
|
||||
@ -1057,7 +1058,7 @@ class Delta:
|
||||
return [self.x, self.y]
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'Delta':
|
||||
def read(stream: IO[bytes]) -> 'Delta':
|
||||
"""
|
||||
Read a `Delta` object from the provided stream.
|
||||
|
||||
@ -1083,7 +1084,7 @@ class Delta:
|
||||
y = read_sint(stream)
|
||||
return Delta(x, y)
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write a `Delta` object to the provided stream.
|
||||
|
||||
@ -1109,7 +1110,7 @@ class Delta:
|
||||
return str(self.as_list())
|
||||
|
||||
|
||||
def read_repetition(stream: io.BufferedIOBase) -> repetition_t:
|
||||
def read_repetition(stream: IO[bytes]) -> repetition_t:
|
||||
"""
|
||||
Read a repetition entry from the given stream.
|
||||
|
||||
@ -1133,7 +1134,7 @@ def read_repetition(stream: io.BufferedIOBase) -> repetition_t:
|
||||
raise InvalidDataError(f'Unexpected repetition type: {rtype}')
|
||||
|
||||
|
||||
def write_repetition(stream: io.BufferedIOBase, repetition: repetition_t) -> int:
|
||||
def write_repetition(stream: IO[bytes], repetition: repetition_t) -> int:
|
||||
"""
|
||||
Write a repetition entry to the given stream.
|
||||
|
||||
@ -1153,10 +1154,10 @@ class ReuseRepetition:
|
||||
the most recently written repetition should be reused.
|
||||
"""
|
||||
@staticmethod
|
||||
def read(_stream: io.BufferedIOBase, _repetition_type: int) -> 'ReuseRepetition':
|
||||
def read(_stream: IO[bytes], _repetition_type: int) -> 'ReuseRepetition':
|
||||
return ReuseRepetition()
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
return write_uint(stream, 0)
|
||||
|
||||
def __eq__(self, other: Any) -> bool:
|
||||
@ -1230,7 +1231,7 @@ class GridRepetition:
|
||||
self.b_count = b_count
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, repetition_type: int) -> 'GridRepetition':
|
||||
def read(stream: IO[bytes], repetition_type: int) -> 'GridRepetition':
|
||||
"""
|
||||
Read a `GridRepetition` from a stream.
|
||||
|
||||
@ -1276,7 +1277,7 @@ class GridRepetition:
|
||||
raise InvalidDataError(f'Invalid type for grid repetition {repetition_type}')
|
||||
return GridRepetition(a_vector, na, b_vector, nb)
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write the `GridRepetition` to a stream.
|
||||
|
||||
@ -1375,7 +1376,7 @@ class ArbitraryRepetition:
|
||||
self.y_displacements = list(y_displacements)
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, repetition_type: int) -> 'ArbitraryRepetition':
|
||||
def read(stream: IO[bytes], repetition_type: int) -> 'ArbitraryRepetition':
|
||||
"""
|
||||
Read an `ArbitraryRepetition` from a stream.
|
||||
|
||||
@ -1429,7 +1430,7 @@ class ArbitraryRepetition:
|
||||
raise InvalidDataError(f'Invalid ArbitraryRepetition repetition_type: {repetition_type}')
|
||||
return ArbitraryRepetition(x_displacements, y_displacements)
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write the `ArbitraryRepetition` to a stream.
|
||||
|
||||
@ -1504,7 +1505,7 @@ class ArbitraryRepetition:
|
||||
|
||||
|
||||
def read_point_list(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
implicit_closed: bool,
|
||||
) -> List[List[int]]:
|
||||
"""
|
||||
@ -1537,7 +1538,7 @@ def read_point_list(
|
||||
for i in range(list_len):
|
||||
n = read_sint(stream)
|
||||
if n == 0:
|
||||
raise Exception('Zero-sized 1-delta')
|
||||
raise InvalidDataError('Zero-sized 1-delta')
|
||||
point = [0, 0]
|
||||
point[(i + 1) % 2] = n
|
||||
points.append(point)
|
||||
@ -1593,7 +1594,7 @@ def read_point_list(
|
||||
|
||||
|
||||
def write_point_list(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
points: List[Sequence[int]],
|
||||
fast: bool = False,
|
||||
implicit_closed: bool = True
|
||||
@ -1740,7 +1741,7 @@ class PropStringReference:
|
||||
return f'[{self.ref_type} : {self.ref}]'
|
||||
|
||||
|
||||
def read_property_value(stream: io.BufferedIOBase) -> property_value_t:
|
||||
def read_property_value(stream: IO[bytes]) -> property_value_t:
|
||||
"""
|
||||
Read a property value from a stream.
|
||||
|
||||
@ -1798,7 +1799,7 @@ def read_property_value(stream: io.BufferedIOBase) -> property_value_t:
|
||||
|
||||
|
||||
def write_property_value(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
value: property_value_t,
|
||||
force_real: bool = False,
|
||||
force_signed_int: bool = False,
|
||||
@ -1850,11 +1851,11 @@ def write_property_value(
|
||||
size = write_uint(stream, 15)
|
||||
size += write_uint(stream, value.ref)
|
||||
else:
|
||||
raise Exception(f'Invalid property type: {type(value)} ({value})')
|
||||
raise InvalidDataError(f'Invalid property type: {type(value)} ({value})')
|
||||
return size
|
||||
|
||||
|
||||
def read_interval(stream: io.BufferedIOBase) -> Tuple[Optional[int], Optional[int]]:
|
||||
def read_interval(stream: IO[bytes]) -> Tuple[Optional[int], Optional[int]]:
|
||||
"""
|
||||
Read an interval from a stream.
|
||||
These are used for storing layer info.
|
||||
@ -1895,7 +1896,7 @@ def read_interval(stream: io.BufferedIOBase) -> Tuple[Optional[int], Optional[in
|
||||
|
||||
|
||||
def write_interval(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
min_bound: Optional[int] = None,
|
||||
max_bound: Optional[int] = None,
|
||||
) -> int:
|
||||
@ -1962,7 +1963,7 @@ class OffsetEntry:
|
||||
self.offset = offset
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'OffsetEntry':
|
||||
def read(stream: IO[bytes]) -> 'OffsetEntry':
|
||||
"""
|
||||
Read an offset entry from a stream.
|
||||
|
||||
@ -1977,7 +1978,7 @@ class OffsetEntry:
|
||||
entry.offset = read_uint(stream)
|
||||
return entry
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write this offset entry to a stream.
|
||||
|
||||
@ -2063,7 +2064,7 @@ class OffsetTable:
|
||||
self.xnames = xnames
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'OffsetTable':
|
||||
def read(stream: IO[bytes]) -> 'OffsetTable':
|
||||
"""
|
||||
Read an offset table from a stream.
|
||||
See class docstring for format details.
|
||||
@ -2083,7 +2084,7 @@ class OffsetTable:
|
||||
table.xnames = OffsetEntry.read(stream)
|
||||
return table
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write this offset table to a stream.
|
||||
See class docstring for format details.
|
||||
@ -2107,7 +2108,7 @@ class OffsetTable:
|
||||
self.propstrings, self.layernames, self.xnames])
|
||||
|
||||
|
||||
def read_u32(stream: io.BufferedIOBase) -> int:
|
||||
def read_u32(stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Read a 32-bit unsigned integer (little endian) from a stream.
|
||||
|
||||
@ -2121,7 +2122,7 @@ def read_u32(stream: io.BufferedIOBase) -> int:
|
||||
return struct.unpack('<I', b)[0]
|
||||
|
||||
|
||||
def write_u32(stream: io.BufferedIOBase, n: int) -> int:
|
||||
def write_u32(stream: IO[bytes], n: int) -> int:
|
||||
"""
|
||||
Write a 32-bit unsigned integer (little endian) to a stream.
|
||||
|
||||
@ -2174,7 +2175,7 @@ class Validation:
|
||||
self.checksum = checksum
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'Validation':
|
||||
def read(stream: IO[bytes]) -> 'Validation':
|
||||
"""
|
||||
Read a validation entry from a stream.
|
||||
See class docstring for format details.
|
||||
@ -2199,7 +2200,7 @@ class Validation:
|
||||
raise InvalidDataError('Invalid validation type!')
|
||||
return Validation(checksum_type, checksum)
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write this validation entry to a stream.
|
||||
See class docstring for format details.
|
||||
@ -2228,7 +2229,7 @@ class Validation:
|
||||
return f'Validation(type: {self.checksum_type} sum: {self.checksum})'
|
||||
|
||||
|
||||
def write_magic_bytes(stream: io.BufferedIOBase) -> int:
|
||||
def write_magic_bytes(stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write the magic byte sequence to a stream.
|
||||
|
||||
@ -2241,7 +2242,7 @@ def write_magic_bytes(stream: io.BufferedIOBase) -> int:
|
||||
return stream.write(MAGIC_BYTES)
|
||||
|
||||
|
||||
def read_magic_bytes(stream: io.BufferedIOBase):
|
||||
def read_magic_bytes(stream: IO[bytes]):
|
||||
"""
|
||||
Read the magic byte sequence from a stream.
|
||||
Raise an `InvalidDataError` if it was not found.
|
||||
|
@ -3,7 +3,7 @@ This module contains data structures and functions for reading from and
|
||||
writing to whole OASIS layout files, and provides a few additional
|
||||
abstractions for the data contained inside them.
|
||||
"""
|
||||
from typing import List, Dict, Union, Optional, Type
|
||||
from typing import List, Dict, Union, Optional, Type, IO
|
||||
import io
|
||||
import logging
|
||||
|
||||
@ -116,7 +116,7 @@ class OasisLayout:
|
||||
self.layers = []
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase) -> 'OasisLayout':
|
||||
def read(stream: IO[bytes]) -> 'OasisLayout':
|
||||
"""
|
||||
Read an entire .oas file into an `OasisLayout` object.
|
||||
|
||||
@ -138,7 +138,7 @@ class OasisLayout:
|
||||
|
||||
def read_record(
|
||||
self,
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
modals: Modals,
|
||||
file_state: FileModals
|
||||
) -> bool:
|
||||
@ -343,7 +343,7 @@ class OasisLayout:
|
||||
raise InvalidRecordError(f'Unknown record id: {record_id}')
|
||||
return False
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write this object in OASIS fromat to a stream.
|
||||
|
||||
@ -430,7 +430,7 @@ class Cell:
|
||||
self.placements = [] if placements is None else placements
|
||||
self.geometry = [] if geometry is None else geometry
|
||||
|
||||
def dedup_write(self, stream: io.BufferedIOBase, modals: Modals) -> int:
|
||||
def dedup_write(self, stream: IO[bytes], modals: Modals) -> int:
|
||||
"""
|
||||
Write this cell to a stream, using the provided modal variables to
|
||||
deduplicate any repeated data.
|
||||
|
@ -10,7 +10,7 @@ Higher-level code (e.g. monitoring for combinations of records with
|
||||
parse, or code for dealing with nested records in a CBlock) should live
|
||||
in main.py instead.
|
||||
"""
|
||||
from typing import List, Dict, Tuple, Union, Optional, Sequence, Any, TypeVar
|
||||
from typing import List, Dict, Tuple, Union, Optional, Sequence, Any, TypeVar, IO
|
||||
from abc import ABCMeta, abstractmethod
|
||||
import copy
|
||||
import math
|
||||
@ -159,7 +159,7 @@ class Record(metaclass=ABCMeta):
|
||||
|
||||
@staticmethod
|
||||
@abstractmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Record':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Record':
|
||||
"""
|
||||
Read a record of this type from a stream.
|
||||
This function does not merge with modal variables.
|
||||
@ -179,7 +179,7 @@ class Record(metaclass=ABCMeta):
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
"""
|
||||
Write this record to a stream as-is.
|
||||
This function does not merge or deduplicate with modal variables.
|
||||
@ -195,7 +195,7 @@ class Record(metaclass=ABCMeta):
|
||||
"""
|
||||
pass
|
||||
|
||||
def dedup_write(self, stream: io.BufferedIOBase, modals: Modals) -> int:
|
||||
def dedup_write(self, stream: IO[bytes], modals: Modals) -> int:
|
||||
"""
|
||||
Run `.deduplicate_with_modals()` and then `.write()` to the stream.
|
||||
|
||||
@ -256,7 +256,7 @@ class GeometryMixin(metaclass=ABCMeta):
|
||||
|
||||
|
||||
def read_refname(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
is_present: Union[bool, int],
|
||||
is_reference: Union[bool, int]
|
||||
) -> Union[None, int, NString]:
|
||||
@ -281,7 +281,7 @@ def read_refname(
|
||||
|
||||
|
||||
def read_refstring(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
is_present: Union[bool, int],
|
||||
is_reference: Union[bool, int],
|
||||
) -> Union[None, int, AString]:
|
||||
@ -316,14 +316,14 @@ class Pad(Record):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Pad':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Pad':
|
||||
if record_id != 0:
|
||||
raise InvalidDataError(f'Invalid record id for Pad {record_id}')
|
||||
record = Pad()
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
return write_uint(stream, 0)
|
||||
|
||||
|
||||
@ -355,14 +355,14 @@ class XYMode(Record):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'XYMode':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'XYMode':
|
||||
if record_id not in (15, 16):
|
||||
raise InvalidDataError('Invalid record id for XYMode')
|
||||
record = XYMode(record_id == 16)
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
return write_uint(stream, 15 + self.relative)
|
||||
|
||||
|
||||
@ -417,7 +417,7 @@ class Start(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Start':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Start':
|
||||
if record_id != 1:
|
||||
raise InvalidDataError(f'Invalid record id for Start: {record_id}')
|
||||
version = AString.read(stream)
|
||||
@ -432,7 +432,7 @@ class Start(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
size = write_uint(stream, 1)
|
||||
size += self.version.write(stream)
|
||||
size += write_real(stream, self.unit)
|
||||
@ -478,7 +478,7 @@ class End(Record):
|
||||
|
||||
@staticmethod
|
||||
def read(
|
||||
stream: io.BufferedIOBase,
|
||||
stream: IO[bytes],
|
||||
record_id: int,
|
||||
has_offset_table: bool
|
||||
) -> 'End':
|
||||
@ -494,7 +494,7 @@ class End(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
size = write_uint(stream, 2)
|
||||
if self.offset_table is not None:
|
||||
size += self.offset_table.write(stream)
|
||||
@ -550,7 +550,7 @@ class CBlock(Record):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'CBlock':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'CBlock':
|
||||
if record_id != 34:
|
||||
raise InvalidDataError(f'Invalid record id for CBlock: {record_id}')
|
||||
compression_type = read_uint(stream)
|
||||
@ -560,7 +560,7 @@ class CBlock(Record):
|
||||
logger.debug(f'CBlock ending at 0x{stream.tell():x} was read successfully')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
size = write_uint(stream, 34)
|
||||
size += write_uint(stream, self.compression_type)
|
||||
size += write_uint(stream, self.decompressed_byte_count)
|
||||
@ -662,7 +662,7 @@ class CellName(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'CellName':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'CellName':
|
||||
if record_id not in (3, 4):
|
||||
raise InvalidDataError(f'Invalid record id for CellName {record_id}')
|
||||
nstring = NString.read(stream)
|
||||
@ -674,7 +674,7 @@ class CellName(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
record_id = 3 + (self.reference_number is not None)
|
||||
size = write_uint(stream, record_id)
|
||||
size += self.nstring.write(stream)
|
||||
@ -717,7 +717,7 @@ class PropName(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'PropName':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'PropName':
|
||||
if record_id not in (7, 8):
|
||||
raise InvalidDataError(f'Invalid record id for PropName {record_id}')
|
||||
nstring = NString.read(stream)
|
||||
@ -729,7 +729,7 @@ class PropName(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
record_id = 7 + (self.reference_number is not None)
|
||||
size = write_uint(stream, record_id)
|
||||
size += self.nstring.write(stream)
|
||||
@ -773,7 +773,7 @@ class TextString(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'TextString':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'TextString':
|
||||
if record_id not in (5, 6):
|
||||
raise InvalidDataError(f'Invalid record id for TextString: {record_id}')
|
||||
astring = AString.read(stream)
|
||||
@ -785,7 +785,7 @@ class TextString(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
record_id = 5 + (self.reference_number is not None)
|
||||
size = write_uint(stream, record_id)
|
||||
size += self.astring.write(stream)
|
||||
@ -829,7 +829,7 @@ class PropString(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'PropString':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'PropString':
|
||||
if record_id not in (9, 10):
|
||||
raise InvalidDataError(f'Invalid record id for PropString: {record_id}')
|
||||
astring = AString.read(stream)
|
||||
@ -841,7 +841,7 @@ class PropString(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
record_id = 9 + (self.reference_number is not None)
|
||||
size = write_uint(stream, record_id)
|
||||
size += self.astring.write(stream)
|
||||
@ -896,7 +896,7 @@ class LayerName(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'LayerName':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'LayerName':
|
||||
if record_id not in (11, 12):
|
||||
raise InvalidDataError(f'Invalid record id for LayerName: {record_id}')
|
||||
is_textlayer = (record_id == 12)
|
||||
@ -907,7 +907,7 @@ class LayerName(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
record_id = 11 + self.is_textlayer
|
||||
size = write_uint(stream, record_id)
|
||||
size += self.nstring.write(stream)
|
||||
@ -973,7 +973,7 @@ class Property(Record):
|
||||
dedup_field(self, 'is_standard', modals, 'property_is_standard')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Property':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Property':
|
||||
if record_id not in (28, 29):
|
||||
raise InvalidDataError(f'Invalid record id for PropertyValue: {record_id}')
|
||||
if record_id == 29:
|
||||
@ -1003,7 +1003,7 @@ class Property(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
if self.is_standard is None and self.values is None and self.name is None:
|
||||
return write_uint(stream, 29)
|
||||
else:
|
||||
@ -1075,7 +1075,7 @@ class XName(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'XName':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'XName':
|
||||
if record_id not in (30, 31):
|
||||
raise InvalidDataError(f'Invalid record id for XName: {record_id}')
|
||||
attribute = read_uint(stream)
|
||||
@ -1088,7 +1088,7 @@ class XName(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
record_id = 30 + (self.reference_number is not None)
|
||||
size = write_uint(stream, record_id)
|
||||
size += write_uint(stream, self.attribute)
|
||||
@ -1133,7 +1133,7 @@ class XElement(Record):
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'XElement':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'XElement':
|
||||
if record_id != 32:
|
||||
raise InvalidDataError(f'Invalid record id for XElement: {record_id}')
|
||||
attribute = read_uint(stream)
|
||||
@ -1142,7 +1142,7 @@ class XElement(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
size = write_uint(stream, 32)
|
||||
size += write_uint(stream, self.attribute)
|
||||
size += write_bstring(stream, self.bstring)
|
||||
@ -1216,7 +1216,7 @@ class XGeometry(Record, GeometryMixin):
|
||||
dedup_field(self, 'datatype', modals, 'datatype')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'XGeometry':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'XGeometry':
|
||||
if record_id != 33:
|
||||
raise InvalidDataError(f'Invalid record id for XGeometry: {record_id}')
|
||||
|
||||
@ -1241,7 +1241,7 @@ class XGeometry(Record, GeometryMixin):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
x = self.x is not None
|
||||
y = self.y is not None
|
||||
r = self.repetition is not None
|
||||
@ -1288,7 +1288,7 @@ class Cell(Record):
|
||||
modals.reset()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Cell':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Cell':
|
||||
name: Union[int, NString]
|
||||
if record_id == 13:
|
||||
name = read_uint(stream)
|
||||
@ -1300,7 +1300,7 @@ class Cell(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
size = 0
|
||||
if isinstance(self.name, int):
|
||||
size += write_uint(stream, 13)
|
||||
@ -1391,7 +1391,7 @@ class Placement(Record):
|
||||
dedup_field(self, 'name', modals, 'placement_cell')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Placement':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Placement':
|
||||
if record_id not in (17, 18):
|
||||
raise InvalidDataError(f'Invalid record id for Placement: {record_id}')
|
||||
|
||||
@ -1421,7 +1421,7 @@ class Placement(Record):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
c = self.name is not None
|
||||
n = c and isinstance(self.name, int)
|
||||
x = self.x is not None
|
||||
@ -1534,7 +1534,7 @@ class Text(Record, GeometryMixin):
|
||||
dedup_field(self, 'datatype', modals, 'text_datatype')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Text':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Text':
|
||||
if record_id != 19:
|
||||
raise InvalidDataError(f'Invalid record id for Text: {record_id}')
|
||||
|
||||
@ -1559,7 +1559,7 @@ class Text(Record, GeometryMixin):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
c = self.string is not None
|
||||
n = c and isinstance(self.string, int)
|
||||
x = self.x is not None
|
||||
@ -1609,14 +1609,14 @@ class Rectangle(Record, GeometryMixin):
|
||||
repetition (Optional[repetition_t]): Repetition, if any.
|
||||
properties (List[Property]): List of property records associate with this record.
|
||||
"""
|
||||
layer: Optional[int] = None
|
||||
datatype: Optional[int] = None
|
||||
width: Optional[int] = None
|
||||
height: Optional[int] = None
|
||||
x: Optional[int] = None
|
||||
y: Optional[int] = None
|
||||
repetition: Optional[repetition_t] = None
|
||||
is_square: bool = False
|
||||
layer: Optional[int]
|
||||
datatype: Optional[int]
|
||||
width: Optional[int]
|
||||
height: Optional[int]
|
||||
x: Optional[int]
|
||||
y: Optional[int]
|
||||
repetition: Optional[repetition_t]
|
||||
is_square: bool
|
||||
properties: List['Property']
|
||||
|
||||
def __init__(
|
||||
@ -1674,7 +1674,7 @@ class Rectangle(Record, GeometryMixin):
|
||||
dedup_field(self, 'height', modals, 'geometry_h')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Rectangle':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Rectangle':
|
||||
if record_id != 20:
|
||||
raise InvalidDataError(f'Invalid record id for Rectangle: {record_id}')
|
||||
|
||||
@ -1698,7 +1698,7 @@ class Rectangle(Record, GeometryMixin):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
s = self.is_square
|
||||
w = self.width is not None
|
||||
h = self.height is not None
|
||||
@ -1749,12 +1749,12 @@ class Polygon(Record, GeometryMixin):
|
||||
Default no repetition.
|
||||
properties (List[Property]): List of property records associate with this record.
|
||||
"""
|
||||
layer: Optional[int] = None
|
||||
datatype: Optional[int] = None
|
||||
x: Optional[int] = None
|
||||
y: Optional[int] = None
|
||||
repetition: Optional[repetition_t] = None
|
||||
point_list: Optional[point_list_t] = None
|
||||
layer: Optional[int]
|
||||
datatype: Optional[int]
|
||||
x: Optional[int]
|
||||
y: Optional[int]
|
||||
repetition: Optional[repetition_t]
|
||||
point_list: Optional[point_list_t]
|
||||
properties: List['Property']
|
||||
|
||||
def __init__(
|
||||
@ -1797,7 +1797,7 @@ class Polygon(Record, GeometryMixin):
|
||||
dedup_field(self, 'point_list', modals, 'polygon_point_list')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Polygon':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Polygon':
|
||||
if record_id != 21:
|
||||
raise InvalidDataError(f'Invalid record id for Polygon: {record_id}')
|
||||
|
||||
@ -1822,7 +1822,7 @@ class Polygon(Record, GeometryMixin):
|
||||
logger.debug('Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase, fast: bool = False) -> int:
|
||||
def write(self, stream: IO[bytes], fast: bool = False) -> int:
|
||||
p = self.point_list is not None
|
||||
x = self.x is not None
|
||||
y = self.y is not None
|
||||
@ -1941,7 +1941,7 @@ class Path(Record, GeometryMixin):
|
||||
dedup_field(self, 'extension_end', modals, 'path_extension_end')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Path':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Path':
|
||||
if record_id != 22:
|
||||
raise InvalidDataError(f'Invalid record id for Path: {record_id}')
|
||||
|
||||
@ -1984,7 +1984,7 @@ class Path(Record, GeometryMixin):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase, fast: bool = False) -> int:
|
||||
def write(self, stream: IO[bytes], fast: bool = False) -> int:
|
||||
e = self.extension_start is not None or self.extension_end is not None
|
||||
w = self.half_width is not None
|
||||
p = self.point_list is not None
|
||||
@ -2136,7 +2136,7 @@ class Trapezoid(Record, GeometryMixin):
|
||||
dedup_field(self, 'height', modals, 'geometry_h')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Trapezoid':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Trapezoid':
|
||||
if record_id not in (23, 24, 25):
|
||||
raise InvalidDataError(f'Invalid record id for Trapezoid: {record_id}')
|
||||
|
||||
@ -2164,7 +2164,7 @@ class Trapezoid(Record, GeometryMixin):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
v = self.is_vertical
|
||||
w = self.width is not None
|
||||
h = self.height is not None
|
||||
@ -2358,7 +2358,7 @@ class CTrapezoid(Record, GeometryMixin):
|
||||
self.check_valid()
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'CTrapezoid':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'CTrapezoid':
|
||||
if record_id != 26:
|
||||
raise InvalidDataError(f'Invalid record id for CTrapezoid: {record_id}')
|
||||
|
||||
@ -2384,7 +2384,7 @@ class CTrapezoid(Record, GeometryMixin):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
t = self.ctrapezoid_type is not None
|
||||
w = self.width is not None
|
||||
h = self.height is not None
|
||||
@ -2508,7 +2508,7 @@ class Circle(Record, GeometryMixin):
|
||||
dedup_field(self, 'radius', modals, 'circle_radius')
|
||||
|
||||
@staticmethod
|
||||
def read(stream: io.BufferedIOBase, record_id: int) -> 'Circle':
|
||||
def read(stream: IO[bytes], record_id: int) -> 'Circle':
|
||||
if record_id != 27:
|
||||
raise InvalidDataError(f'Invalid record id for Circle: {record_id}')
|
||||
|
||||
@ -2533,7 +2533,7 @@ class Circle(Record, GeometryMixin):
|
||||
logger.debug(f'Record ending at 0x{stream.tell():x}:\n {record}')
|
||||
return record
|
||||
|
||||
def write(self, stream: io.BufferedIOBase) -> int:
|
||||
def write(self, stream: IO[bytes]) -> int:
|
||||
s = self.radius is not None
|
||||
x = self.x is not None
|
||||
y = self.y is not None
|
||||