Use IO[bytes] instead of BinaryIO wherever possible

This commit is contained in:
Jan Petykiewicz 2023-01-26 12:47:09 -08:00
parent f9d4cfec33
commit b2928c8b1c
3 changed files with 134 additions and 133 deletions

View File

@ -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.

View File

@ -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.

View File

@ -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