diff --git a/fatamorgana/basic.py b/fatamorgana/basic.py index b745f2f..2e41b6c 100644 --- a/fatamorgana/basic.py +++ b/fatamorgana/basic.py @@ -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(" 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(" 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(' 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. diff --git a/fatamorgana/main.py b/fatamorgana/main.py index a11af4e..8748d7f 100644 --- a/fatamorgana/main.py +++ b/fatamorgana/main.py @@ -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. diff --git a/fatamorgana/records.py b/fatamorgana/records.py index 453ad03..20e9819 100644 --- a/fatamorgana/records.py +++ b/fatamorgana/records.py @@ -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