From 0bbc8f8e0874effb25d2fa7a14823c8b74ca9851 Mon Sep 17 00:00:00 2001 From: Jan Petykiewicz Date: Thu, 26 Jan 2023 12:32:19 -0800 Subject: [PATCH] Loosen constraints from BinaryIO to IO[bytes] --- klamath/basic.py | 4 ++-- klamath/elements.py | 34 +++++++++++++++++----------------- klamath/library.py | 16 ++++++++-------- klamath/record.py | 36 ++++++++++++++++++------------------ 4 files changed, 45 insertions(+), 45 deletions(-) diff --git a/klamath/basic.py b/klamath/basic.py index fd1b417..e680cb1 100644 --- a/klamath/basic.py +++ b/klamath/basic.py @@ -1,7 +1,7 @@ """ Functionality for encoding/decoding basic datatypes """ -from typing import Sequence, BinaryIO, List +from typing import Sequence, IO, List import struct from datetime import datetime @@ -172,7 +172,7 @@ def pack_datetime(data: Sequence[datetime]) -> bytes: return pack_int2(parts) -def read(stream: BinaryIO, size: int) -> bytes: +def read(stream: IO[bytes], size: int) -> bytes: """ Read and check for failure """ data = stream.read(size) if len(data) != size: diff --git a/klamath/elements.py b/klamath/elements.py index a477642..1fe3837 100644 --- a/klamath/elements.py +++ b/klamath/elements.py @@ -2,7 +2,7 @@ Functionality for reading/writing elements (geometry, text labels, structure references) and associated properties. """ -from typing import Dict, Tuple, Optional, BinaryIO, TypeVar, Type, Union +from typing import Dict, Tuple, Optional, IO, TypeVar, Type, Union from abc import ABCMeta, abstractmethod from dataclasses import dataclass @@ -29,7 +29,7 @@ X = TypeVar('X', bound='Box') -def read_properties(stream: BinaryIO) -> Dict[int, bytes]: +def read_properties(stream: IO[bytes]) -> Dict[int, bytes]: """ Read element properties. @@ -56,7 +56,7 @@ def read_properties(stream: BinaryIO) -> Dict[int, bytes]: return properties -def write_properties(stream: BinaryIO, properties: Dict[int, bytes]) -> int: +def write_properties(stream: IO[bytes], properties: Dict[int, bytes]) -> int: """ Write element properties. @@ -78,7 +78,7 @@ class Element(metaclass=ABCMeta): """ @classmethod @abstractmethod - def read(cls: Type[E], stream: BinaryIO) -> E: + def read(cls: Type[E], stream: IO[bytes]) -> E: """ Read from a stream to construct this object. Consumes up to (and including) the ENDEL record. @@ -92,7 +92,7 @@ class Element(metaclass=ABCMeta): pass @abstractmethod - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: """ Write this element to a stream. Finishes with an ENDEL record. @@ -151,7 +151,7 @@ class Reference(Element): """ Properties associated with this reference. """ @classmethod - def read(cls: Type[R], stream: BinaryIO) -> R: + def read(cls: Type[R], stream: IO[bytes]) -> R: invert_y = False mag = 1 angle_deg = 0 @@ -177,7 +177,7 @@ class Reference(Element): return cls(struct_name=struct_name, xy=xy, properties=properties, colrow=colrow, invert_y=invert_y, mag=mag, angle_deg=angle_deg) - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: b = 0 if self.colrow is None: b += SREF.write(stream, None) @@ -226,14 +226,14 @@ class Boundary(Element): """ Properties for the element. """ @classmethod - def read(cls: Type[B], stream: BinaryIO) -> B: + def read(cls: Type[B], stream: IO[bytes]) -> B: layer = LAYER.skip_and_read(stream)[0] dtype = DATATYPE.read(stream)[0] xy = XY.read(stream).reshape(-1, 2) properties = read_properties(stream) return cls(layer=(layer, dtype), xy=xy, properties=properties) - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: b = BOUNDARY.write(stream, None) b += LAYER.write(stream, self.layer[0]) b += DATATYPE.write(stream, self.layer[1]) @@ -272,7 +272,7 @@ class Path(Element): """ Properties for the element. """ @classmethod - def read(cls: Type[P], stream: BinaryIO) -> P: + def read(cls: Type[P], stream: IO[bytes]) -> P: path_type = 0 width = 0 bgn_ext = 0 @@ -299,7 +299,7 @@ class Path(Element): properties=properties, extension=(bgn_ext, end_ext), path_type=path_type, width=width) - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: b = PATH.write(stream, None) b += LAYER.write(stream, self.layer[0]) b += DATATYPE.write(stream, self.layer[1]) @@ -337,14 +337,14 @@ class Box(Element): """ Properties for the element. """ @classmethod - def read(cls: Type[X], stream: BinaryIO) -> X: + def read(cls: Type[X], stream: IO[bytes]) -> X: layer = LAYER.skip_and_read(stream)[0] dtype = BOXTYPE.read(stream)[0] xy = XY.read(stream).reshape(-1, 2) properties = read_properties(stream) return cls(layer=(layer, dtype), xy=xy, properties=properties) - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: b = BOX.write(stream, None) b += LAYER.write(stream, self.layer[0]) b += BOXTYPE.write(stream, self.layer[1]) @@ -371,14 +371,14 @@ class Node(Element): """ Properties for the element. """ @classmethod - def read(cls: Type[N], stream: BinaryIO) -> N: + def read(cls: Type[N], stream: IO[bytes]) -> N: layer = LAYER.skip_and_read(stream)[0] dtype = NODETYPE.read(stream)[0] xy = XY.read(stream).reshape(-1, 2) properties = read_properties(stream) return cls(layer=(layer, dtype), xy=xy, properties=properties) - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: b = NODE.write(stream, None) b += LAYER.write(stream, self.layer[0]) b += NODETYPE.write(stream, self.layer[1]) @@ -431,7 +431,7 @@ class Text(Element): """ Properties for the element. """ @classmethod - def read(cls: Type[T], stream: BinaryIO) -> T: + def read(cls: Type[T], stream: IO[bytes]) -> T: path_type = 0 presentation = 0 invert_y = False @@ -467,7 +467,7 @@ class Text(Element): string=string, presentation=presentation, path_type=path_type, width=width, invert_y=invert_y, mag=mag, angle_deg=angle_deg) - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: b = TEXT.write(stream, None) b += LAYER.write(stream, self.layer[0]) b += TEXTTYPE.write(stream, self.layer[1]) diff --git a/klamath/library.py b/klamath/library.py index 1318854..50c8398 100644 --- a/klamath/library.py +++ b/klamath/library.py @@ -1,7 +1,7 @@ """ File-level read/write functionality. """ -from typing import List, Dict, Tuple, Optional, BinaryIO, TypeVar, Type, MutableMapping +from typing import List, Dict, Tuple, Optional, IO, TypeVar, Type, MutableMapping import io from datetime import datetime from dataclasses import dataclass @@ -45,7 +45,7 @@ class FileHeader: """ Last-accessed time """ @classmethod - def read(cls: Type[FH], stream: BinaryIO) -> FH: + def read(cls: Type[FH], stream: IO[bytes]) -> FH: """ Read and construct a header from the provided stream. @@ -63,7 +63,7 @@ class FileHeader: return cls(mod_time=mod_time, acc_time=acc_time, name=name, user_units_per_db_unit=uu, meters_per_db_unit=dbu) - def write(self, stream: BinaryIO) -> int: + def write(self, stream: IO[bytes]) -> int: """ Write the header to a stream @@ -80,7 +80,7 @@ class FileHeader: return b -def scan_structs(stream: BinaryIO) -> Dict[bytes, int]: +def scan_structs(stream: IO[bytes]) -> Dict[bytes, int]: """ Scan through a GDS file, building a table of {b'structure_name': byte_offset}. @@ -107,7 +107,7 @@ def scan_structs(stream: BinaryIO) -> Dict[bytes, int]: return positions -def try_read_struct(stream: BinaryIO) -> Optional[Tuple[bytes, List[Element]]]: +def try_read_struct(stream: IO[bytes]) -> Optional[Tuple[bytes, List[Element]]]: """ Skip to the next structure and attempt to read it. @@ -125,7 +125,7 @@ def try_read_struct(stream: BinaryIO) -> Optional[Tuple[bytes, List[Element]]]: return name, elements -def write_struct(stream: BinaryIO, +def write_struct(stream: IO[bytes], name: bytes, elements: List[Element], cre_time: datetime = datetime(1900, 1, 1), @@ -150,7 +150,7 @@ def write_struct(stream: BinaryIO, return b -def read_elements(stream: BinaryIO) -> List[Element]: +def read_elements(stream: IO[bytes]) -> List[Element]: """ Read elements from the stream until an ENDSTR record is encountered. The ENDSTR record is also @@ -186,7 +186,7 @@ def read_elements(stream: BinaryIO) -> List[Element]: return data -def scan_hierarchy(stream: BinaryIO) -> Dict[bytes, Dict[bytes, int]]: +def scan_hierarchy(stream: IO[bytes]) -> Dict[bytes, Dict[bytes, int]]: """ Scan through a GDS file, building a table of instance counts `{b'structure_name': {b'ref_name': count}}`. diff --git a/klamath/record.py b/klamath/record.py index e6fb5fb..69769dd 100644 --- a/klamath/record.py +++ b/klamath/record.py @@ -1,7 +1,7 @@ """ Generic record-level read/write functionality. """ -from typing import Optional, Sequence, BinaryIO +from typing import Optional, Sequence, IO from typing import TypeVar, List, Tuple, ClassVar, Type import struct import io @@ -19,7 +19,7 @@ from .basic import parse_ascii, pack_ascii, read _RECORD_HEADER_FMT = struct.Struct('>HH') -def write_record_header(stream: BinaryIO, data_size: int, tag: int) -> int: +def write_record_header(stream: IO[bytes], data_size: int, tag: int) -> int: record_size = data_size + 4 if record_size > 0xFFFF: raise KlamathError(f'Record size is too big: {record_size}') @@ -27,7 +27,7 @@ def write_record_header(stream: BinaryIO, data_size: int, tag: int) -> int: return stream.write(header) -def read_record_header(stream: BinaryIO) -> Tuple[int, int]: +def read_record_header(stream: IO[bytes]) -> Tuple[int, int]: """ Read a record's header (size and tag). Args: @@ -46,7 +46,7 @@ def read_record_header(stream: BinaryIO) -> Tuple[int, int]: return data_size, tag -def expect_record(stream: BinaryIO, tag: int) -> int: +def expect_record(stream: IO[bytes], tag: int) -> int: data_size, actual_tag = read_record_header(stream) if tag != actual_tag: raise KlamathError(f'Unexpected record! Got tag 0x{actual_tag:04x}, expected 0x{tag:04x}') @@ -71,7 +71,7 @@ class Record(metaclass=ABCMeta): @classmethod @abstractmethod - def read_data(cls, stream: BinaryIO, size: int): + def read_data(cls, stream: IO[bytes], size: int): pass @classmethod @@ -80,15 +80,15 @@ class Record(metaclass=ABCMeta): pass @staticmethod - def read_header(stream: BinaryIO) -> Tuple[int, int]: + def read_header(stream: IO[bytes]) -> Tuple[int, int]: return read_record_header(stream) @classmethod - def write_header(cls, stream: BinaryIO, data_size: int) -> int: + def write_header(cls, stream: IO[bytes], data_size: int) -> int: return write_record_header(stream, data_size, cls.tag) @classmethod - def skip_past(cls, stream: BinaryIO) -> bool: + def skip_past(cls, stream: IO[bytes]) -> bool: """ Skip to the end of the next occurence of this record. @@ -110,7 +110,7 @@ class Record(metaclass=ABCMeta): return True @classmethod - def skip_and_read(cls, stream: BinaryIO): + def skip_and_read(cls, stream: IO[bytes]): size, tag = Record.read_header(stream) while tag != cls.tag: stream.seek(size, io.SEEK_CUR) @@ -119,13 +119,13 @@ class Record(metaclass=ABCMeta): return data @classmethod - def read(cls: Type[R], stream: BinaryIO): + def read(cls: Type[R], stream: IO[bytes]): size = expect_record(stream, cls.tag) data = cls.read_data(stream, size) return data @classmethod - def write(cls, stream: BinaryIO, data) -> int: + def write(cls, stream: IO[bytes], data) -> int: data_bytes = cls.pack_data(data) b = cls.write_header(stream, len(data_bytes)) b += stream.write(data_bytes) @@ -136,7 +136,7 @@ class NoDataRecord(Record): expected_size: ClassVar[Optional[int]] = 0 @classmethod - def read_data(cls, stream: BinaryIO, size: int) -> None: + def read_data(cls, stream: IO[bytes], size: int) -> None: stream.read(size) @classmethod @@ -150,7 +150,7 @@ class BitArrayRecord(Record): expected_size: ClassVar[Optional[int]] = 2 @classmethod - def read_data(cls, stream: BinaryIO, size: int) -> int: + def read_data(cls, stream: IO[bytes], size: int) -> int: return parse_bitarray(read(stream, 2)) @classmethod @@ -160,7 +160,7 @@ class BitArrayRecord(Record): class Int2Record(Record): @classmethod - def read_data(cls, stream: BinaryIO, size: int) -> numpy.ndarray: + def read_data(cls, stream: IO[bytes], size: int) -> numpy.ndarray: return parse_int2(read(stream, size)) @classmethod @@ -170,7 +170,7 @@ class Int2Record(Record): class Int4Record(Record): @classmethod - def read_data(cls, stream: BinaryIO, size: int) -> numpy.ndarray: + def read_data(cls, stream: IO[bytes], size: int) -> numpy.ndarray: return parse_int4(read(stream, size)) @classmethod @@ -180,7 +180,7 @@ class Int4Record(Record): class Real8Record(Record): @classmethod - def read_data(cls, stream: BinaryIO, size: int) -> numpy.ndarray: + def read_data(cls, stream: IO[bytes], size: int) -> numpy.ndarray: return parse_real8(read(stream, size)) @classmethod @@ -190,7 +190,7 @@ class Real8Record(Record): class ASCIIRecord(Record): @classmethod - def read_data(cls, stream: BinaryIO, size: int) -> bytes: + def read_data(cls, stream: IO[bytes], size: int) -> bytes: return parse_ascii(read(stream, size)) @classmethod @@ -200,7 +200,7 @@ class ASCIIRecord(Record): class DateTimeRecord(Record): @classmethod - def read_data(cls, stream: BinaryIO, size: int) -> List[datetime]: + def read_data(cls, stream: IO[bytes], size: int) -> List[datetime]: return parse_datetime(read(stream, size)) @classmethod