Compare commits
No commits in common. "a50d53b5085c388457f0b33ef98f07689221d53a" and "6a0019010fb3b47761a5761535418ab400c43402" have entirely different histories.
a50d53b508
...
6a0019010f
10 changed files with 124 additions and 133 deletions
30
.flake8
Normal file
30
.flake8
Normal file
|
|
@ -0,0 +1,30 @@
|
||||||
|
[flake8]
|
||||||
|
ignore =
|
||||||
|
# E501 line too long
|
||||||
|
E501,
|
||||||
|
# W391 newlines at EOF
|
||||||
|
W391,
|
||||||
|
# E241 multiple spaces after comma
|
||||||
|
E241,
|
||||||
|
# E302 expected 2 newlines
|
||||||
|
E302,
|
||||||
|
# W503 line break before binary operator (to be deprecated)
|
||||||
|
W503,
|
||||||
|
# E265 block comment should start with '# '
|
||||||
|
E265,
|
||||||
|
# E123 closing bracket does not match indentation of opening bracket's line
|
||||||
|
E123,
|
||||||
|
# E124 closing bracket does not match visual indentation
|
||||||
|
E124,
|
||||||
|
# E221 multiple spaces before operator
|
||||||
|
E221,
|
||||||
|
# E201 whitespace after '['
|
||||||
|
E201,
|
||||||
|
# # E741 ambiguous variable name 'I'
|
||||||
|
# E741,
|
||||||
|
|
||||||
|
|
||||||
|
per-file-ignores =
|
||||||
|
# F401 import without use
|
||||||
|
*/__init__.py: F401,
|
||||||
|
__init__.py: F401,
|
||||||
|
|
@ -50,7 +50,7 @@ The goal is to keep this library simple:
|
||||||
## Installation
|
## Installation
|
||||||
|
|
||||||
Requirements:
|
Requirements:
|
||||||
* python >= 3.11
|
* python >= 3.10 (written and tested with 3.11)
|
||||||
* numpy
|
* numpy
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -27,14 +27,12 @@ The goal is to keep this library simple:
|
||||||
tools for working with hierarchical design data and supports multiple
|
tools for working with hierarchical design data and supports multiple
|
||||||
file formats.
|
file formats.
|
||||||
"""
|
"""
|
||||||
from . import (
|
from . import basic
|
||||||
basic as basic,
|
from . import record
|
||||||
record as record,
|
from . import records
|
||||||
records as records,
|
from . import elements
|
||||||
elements as elements,
|
from . import library
|
||||||
library as library,
|
|
||||||
)
|
|
||||||
|
|
||||||
__author__ = 'Jan Petykiewicz'
|
__author__ = 'Jan Petykiewicz'
|
||||||
__version__ = '1.4'
|
__version__ = '1.3'
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,7 @@
|
||||||
"""
|
"""
|
||||||
Functionality for encoding/decoding basic datatypes
|
Functionality for encoding/decoding basic datatypes
|
||||||
"""
|
"""
|
||||||
from typing import IO
|
from typing import Sequence, IO
|
||||||
from collections.abc import Sequence
|
|
||||||
import struct
|
import struct
|
||||||
import logging
|
import logging
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
@ -93,15 +92,15 @@ def pack_bitarray(data: int) -> bytes:
|
||||||
return struct.pack('>H', data)
|
return struct.pack('>H', data)
|
||||||
|
|
||||||
|
|
||||||
def pack_int2(data: NDArray[numpy.integer] | Sequence[int] | int) -> bytes:
|
def pack_int2(data: Sequence[int]) -> bytes:
|
||||||
arr = numpy.asarray(data)
|
arr = numpy.array(data)
|
||||||
if (arr > 32767).any() or (arr < -32768).any():
|
if (arr > 32767).any() or (arr < -32768).any():
|
||||||
raise KlamathError(f'int2 data out of range: {arr}')
|
raise KlamathError(f'int2 data out of range: {arr}')
|
||||||
return arr.astype('>i2').tobytes()
|
return arr.astype('>i2').tobytes()
|
||||||
|
|
||||||
|
|
||||||
def pack_int4(data: NDArray[numpy.integer] | Sequence[int] | int) -> bytes:
|
def pack_int4(data: Sequence[int]) -> bytes:
|
||||||
arr = numpy.asarray(data)
|
arr = numpy.array(data)
|
||||||
if (arr > 2147483647).any() or (arr < -2147483648).any():
|
if (arr > 2147483647).any() or (arr < -2147483648).any():
|
||||||
raise KlamathError(f'int4 data out of range: {arr}')
|
raise KlamathError(f'int4 data out of range: {arr}')
|
||||||
return arr.astype('>i4').tobytes()
|
return arr.astype('>i4').tobytes()
|
||||||
|
|
@ -165,8 +164,8 @@ def encode_real8(fnums: NDArray[numpy.float64]) -> NDArray[numpy.uint64]:
|
||||||
return real8.astype(numpy.uint64, copy=False)
|
return real8.astype(numpy.uint64, copy=False)
|
||||||
|
|
||||||
|
|
||||||
def pack_real8(data: NDArray[numpy.floating] | Sequence[float] | float) -> bytes:
|
def pack_real8(data: Sequence[float]) -> bytes:
|
||||||
return encode_real8(numpy.asarray(data)).astype('>u8').tobytes()
|
return encode_real8(numpy.array(data)).astype('>u8').tobytes()
|
||||||
|
|
||||||
|
|
||||||
def pack_ascii(data: bytes) -> bytes:
|
def pack_ascii(data: bytes) -> bytes:
|
||||||
|
|
|
||||||
|
|
@ -2,7 +2,7 @@
|
||||||
Functionality for reading/writing elements (geometry, text labels,
|
Functionality for reading/writing elements (geometry, text labels,
|
||||||
structure references) and associated properties.
|
structure references) and associated properties.
|
||||||
"""
|
"""
|
||||||
from typing import IO, TypeVar
|
from typing import Optional, IO, TypeVar, Type, Union
|
||||||
from abc import ABCMeta, abstractmethod
|
from abc import ABCMeta, abstractmethod
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
||||||
|
|
@ -51,7 +51,7 @@ def read_properties(stream: IO[bytes]) -> dict[int, bytes]:
|
||||||
value = PROPVALUE.read(stream)
|
value = PROPVALUE.read(stream)
|
||||||
if key in properties:
|
if key in properties:
|
||||||
raise KlamathError(f'Duplicate property key: {key!r}')
|
raise KlamathError(f'Duplicate property key: {key!r}')
|
||||||
properties[key] = value
|
properties[key] = value
|
||||||
size, tag = Record.read_header(stream)
|
size, tag = Record.read_header(stream)
|
||||||
return properties
|
return properties
|
||||||
|
|
||||||
|
|
@ -78,7 +78,7 @@ class Element(metaclass=ABCMeta):
|
||||||
"""
|
"""
|
||||||
@classmethod
|
@classmethod
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def read(cls: type[E], stream: IO[bytes]) -> E:
|
def read(cls: Type[E], stream: IO[bytes]) -> E:
|
||||||
"""
|
"""
|
||||||
Read from a stream to construct this object.
|
Read from a stream to construct this object.
|
||||||
Consumes up to (and including) the ENDEL record.
|
Consumes up to (and including) the ENDEL record.
|
||||||
|
|
@ -151,7 +151,7 @@ class Reference(Element):
|
||||||
""" Properties associated with this reference. """
|
""" Properties associated with this reference. """
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[R], stream: IO[bytes]) -> R:
|
def read(cls: Type[R], stream: IO[bytes]) -> R:
|
||||||
invert_y = False
|
invert_y = False
|
||||||
mag = 1
|
mag = 1
|
||||||
angle_deg = 0
|
angle_deg = 0
|
||||||
|
|
@ -211,7 +211,7 @@ class Reference(Element):
|
||||||
if self.colrow is not None:
|
if self.colrow is not None:
|
||||||
if self.xy.size != 6:
|
if self.xy.size != 6:
|
||||||
raise KlamathError(f'colrow is not None, so expected size-6 xy. Got {self.xy}')
|
raise KlamathError(f'colrow is not None, so expected size-6 xy. Got {self.xy}')
|
||||||
else: # noqa: PLR5501
|
else:
|
||||||
if self.xy.size != 2:
|
if self.xy.size != 2:
|
||||||
raise KlamathError(f'Expected size-2 xy. Got {self.xy}')
|
raise KlamathError(f'Expected size-2 xy. Got {self.xy}')
|
||||||
|
|
||||||
|
|
@ -233,7 +233,7 @@ class Boundary(Element):
|
||||||
""" Properties for the element. """
|
""" Properties for the element. """
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[B], stream: IO[bytes]) -> B:
|
def read(cls: Type[B], stream: IO[bytes]) -> B:
|
||||||
layer = LAYER.skip_and_read(stream)[0]
|
layer = LAYER.skip_and_read(stream)[0]
|
||||||
dtype = DATATYPE.read(stream)[0]
|
dtype = DATATYPE.read(stream)[0]
|
||||||
xy = XY.read(stream).reshape(-1, 2)
|
xy = XY.read(stream).reshape(-1, 2)
|
||||||
|
|
@ -279,7 +279,7 @@ class Path(Element):
|
||||||
""" Properties for the element. """
|
""" Properties for the element. """
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[P], stream: IO[bytes]) -> P:
|
def read(cls: Type[P], stream: IO[bytes]) -> P:
|
||||||
path_type = 0
|
path_type = 0
|
||||||
width = 0
|
width = 0
|
||||||
bgn_ext = 0
|
bgn_ext = 0
|
||||||
|
|
@ -344,7 +344,7 @@ class Box(Element):
|
||||||
""" Properties for the element. """
|
""" Properties for the element. """
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[X], stream: IO[bytes]) -> X:
|
def read(cls: Type[X], stream: IO[bytes]) -> X:
|
||||||
layer = LAYER.skip_and_read(stream)[0]
|
layer = LAYER.skip_and_read(stream)[0]
|
||||||
dtype = BOXTYPE.read(stream)[0]
|
dtype = BOXTYPE.read(stream)[0]
|
||||||
xy = XY.read(stream).reshape(-1, 2)
|
xy = XY.read(stream).reshape(-1, 2)
|
||||||
|
|
@ -378,7 +378,7 @@ class Node(Element):
|
||||||
""" Properties for the element. """
|
""" Properties for the element. """
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[N], stream: IO[bytes]) -> N:
|
def read(cls: Type[N], stream: IO[bytes]) -> N:
|
||||||
layer = LAYER.skip_and_read(stream)[0]
|
layer = LAYER.skip_and_read(stream)[0]
|
||||||
dtype = NODETYPE.read(stream)[0]
|
dtype = NODETYPE.read(stream)[0]
|
||||||
xy = XY.read(stream).reshape(-1, 2)
|
xy = XY.read(stream).reshape(-1, 2)
|
||||||
|
|
@ -438,7 +438,7 @@ class Text(Element):
|
||||||
""" Properties for the element. """
|
""" Properties for the element. """
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[T], stream: IO[bytes]) -> T:
|
def read(cls: Type[T], stream: IO[bytes]) -> T:
|
||||||
path_type = 0
|
path_type = 0
|
||||||
presentation = 0
|
presentation = 0
|
||||||
invert_y = False
|
invert_y = False
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,7 @@
|
||||||
"""
|
"""
|
||||||
File-level read/write functionality.
|
File-level read/write functionality.
|
||||||
"""
|
"""
|
||||||
from typing import IO, Self, TYPE_CHECKING
|
from typing import IO, TypeVar, Type, MutableMapping
|
||||||
import io
|
import io
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
|
@ -15,8 +15,8 @@ from .records import BGNSTR, STRNAME, ENDSTR, SNAME, COLROW, ENDEL
|
||||||
from .records import BOX, BOUNDARY, NODE, PATH, TEXT, SREF, AREF
|
from .records import BOX, BOUNDARY, NODE, PATH, TEXT, SREF, AREF
|
||||||
from .elements import Element, Reference, Text, Box, Boundary, Path, Node
|
from .elements import Element, Reference, Text, Box, Boundary, Path, Node
|
||||||
|
|
||||||
if TYPE_CHECKING:
|
|
||||||
from collections.abc import MutableMapping
|
FH = TypeVar('FH', bound='FileHeader')
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|
@ -45,7 +45,7 @@ class FileHeader:
|
||||||
""" Last-accessed time """
|
""" Last-accessed time """
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[Self], stream: IO[bytes]) -> Self:
|
def read(cls: Type[FH], stream: IO[bytes]) -> FH:
|
||||||
"""
|
"""
|
||||||
Read and construct a header from the provided stream.
|
Read and construct a header from the provided stream.
|
||||||
|
|
||||||
|
|
@ -176,7 +176,9 @@ def read_elements(stream: IO[bytes]) -> list[Element]:
|
||||||
data.append(Box.read(stream))
|
data.append(Box.read(stream))
|
||||||
elif tag == TEXT.tag:
|
elif tag == TEXT.tag:
|
||||||
data.append(Text.read(stream))
|
data.append(Text.read(stream))
|
||||||
elif tag in (SREF.tag, AREF.tag):
|
elif tag == SREF.tag:
|
||||||
|
data.append(Reference.read(stream))
|
||||||
|
elif tag == AREF.tag:
|
||||||
data.append(Reference.read(stream))
|
data.append(Reference.read(stream))
|
||||||
else:
|
else:
|
||||||
# don't care, skip
|
# don't care, skip
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,7 @@
|
||||||
"""
|
"""
|
||||||
Generic record-level read/write functionality.
|
Generic record-level read/write functionality.
|
||||||
"""
|
"""
|
||||||
from typing import IO, ClassVar, Self, Generic, TypeVar
|
from typing import Sequence, IO, TypeVar, ClassVar, Type
|
||||||
from collections.abc import Sequence
|
|
||||||
import struct
|
import struct
|
||||||
import io
|
import io
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
|
@ -18,8 +17,6 @@ from .basic import parse_ascii, pack_ascii, read
|
||||||
|
|
||||||
|
|
||||||
_RECORD_HEADER_FMT = struct.Struct('>HH')
|
_RECORD_HEADER_FMT = struct.Struct('>HH')
|
||||||
II = TypeVar('II') # Input type
|
|
||||||
OO = TypeVar('OO') # Output type
|
|
||||||
|
|
||||||
|
|
||||||
def write_record_header(stream: IO[bytes], data_size: int, tag: int) -> int:
|
def write_record_header(stream: IO[bytes], data_size: int, tag: int) -> int:
|
||||||
|
|
@ -56,27 +53,30 @@ def expect_record(stream: IO[bytes], tag: int) -> int:
|
||||||
return data_size
|
return data_size
|
||||||
|
|
||||||
|
|
||||||
class Record(Generic[II, OO], metaclass=ABCMeta):
|
R = TypeVar('R', bound='Record')
|
||||||
|
|
||||||
|
|
||||||
|
class Record(metaclass=ABCMeta):
|
||||||
tag: ClassVar[int] = -1
|
tag: ClassVar[int] = -1
|
||||||
expected_size: ClassVar[int | None] = None
|
expected_size: ClassVar[int | None] = None
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def check_size(cls: type[Self], size: int) -> None:
|
def check_size(cls, size: int):
|
||||||
if cls.expected_size is not None and size != cls.expected_size:
|
if cls.expected_size is not None and size != cls.expected_size:
|
||||||
raise KlamathError(f'Expected size {cls.expected_size}, got {size}')
|
raise KlamathError(f'Expected size {cls.expected_size}, got {size}')
|
||||||
|
|
||||||
@classmethod # noqa: B027 Intentionally non-abstract
|
@classmethod
|
||||||
def check_data(cls: type[Self], data: II) -> None:
|
def check_data(cls, data):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> OO:
|
def read_data(cls, stream: IO[bytes], size: int):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def pack_data(cls: type[Self], data: II) -> bytes:
|
def pack_data(cls, data) -> bytes:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
|
|
@ -84,11 +84,11 @@ class Record(Generic[II, OO], metaclass=ABCMeta):
|
||||||
return read_record_header(stream)
|
return read_record_header(stream)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def write_header(cls: type[Self], stream: IO[bytes], data_size: int) -> int:
|
def write_header(cls, stream: IO[bytes], data_size: int) -> int:
|
||||||
return write_record_header(stream, data_size, cls.tag)
|
return write_record_header(stream, data_size, cls.tag)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def skip_past(cls: type[Self], stream: IO[bytes]) -> bool:
|
def skip_past(cls, stream: IO[bytes]) -> bool:
|
||||||
"""
|
"""
|
||||||
Skip to the end of the next occurence of this record.
|
Skip to the end of the next occurence of this record.
|
||||||
|
|
||||||
|
|
@ -110,7 +110,7 @@ class Record(Generic[II, OO], metaclass=ABCMeta):
|
||||||
return True
|
return True
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def skip_and_read(cls: type[Self], stream: IO[bytes]) -> OO:
|
def skip_and_read(cls, stream: IO[bytes]):
|
||||||
size, tag = Record.read_header(stream)
|
size, tag = Record.read_header(stream)
|
||||||
while tag != cls.tag:
|
while tag != cls.tag:
|
||||||
stream.seek(size, io.SEEK_CUR)
|
stream.seek(size, io.SEEK_CUR)
|
||||||
|
|
@ -119,90 +119,90 @@ class Record(Generic[II, OO], metaclass=ABCMeta):
|
||||||
return data
|
return data
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read(cls: type[Self], stream: IO[bytes]) -> OO:
|
def read(cls: Type[R], stream: IO[bytes]):
|
||||||
size = expect_record(stream, cls.tag)
|
size = expect_record(stream, cls.tag)
|
||||||
data = cls.read_data(stream, size)
|
data = cls.read_data(stream, size)
|
||||||
return data
|
return data
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def write(cls: type[Self], stream: IO[bytes], data: II) -> int:
|
def write(cls, stream: IO[bytes], data) -> int:
|
||||||
data_bytes = cls.pack_data(data)
|
data_bytes = cls.pack_data(data)
|
||||||
b = cls.write_header(stream, len(data_bytes))
|
b = cls.write_header(stream, len(data_bytes))
|
||||||
b += stream.write(data_bytes)
|
b += stream.write(data_bytes)
|
||||||
return b
|
return b
|
||||||
|
|
||||||
|
|
||||||
class NoDataRecord(Record[None, None]):
|
class NoDataRecord(Record):
|
||||||
expected_size: ClassVar[int | None] = 0
|
expected_size: ClassVar[int | None] = 0
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> None:
|
def read_data(cls, stream: IO[bytes], size: int) -> None:
|
||||||
stream.read(size)
|
stream.read(size)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def pack_data(cls: type[Self], data: None) -> bytes:
|
def pack_data(cls, data: None) -> bytes:
|
||||||
if data is not None:
|
if data is not None:
|
||||||
raise KlamathError('?? Packing {data} into NoDataRecord??')
|
raise KlamathError('?? Packing {data} into NoDataRecord??')
|
||||||
return b''
|
return b''
|
||||||
|
|
||||||
|
|
||||||
class BitArrayRecord(Record[int, int]):
|
class BitArrayRecord(Record):
|
||||||
expected_size: ClassVar[int | None] = 2
|
expected_size: ClassVar[int | None] = 2
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> int: # noqa: ARG003 size unused
|
def read_data(cls, stream: IO[bytes], size: int) -> int:
|
||||||
return parse_bitarray(read(stream, 2))
|
return parse_bitarray(read(stream, 2))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def pack_data(cls: type[Self], data: int) -> bytes:
|
def pack_data(cls, data: int) -> bytes:
|
||||||
return pack_bitarray(data)
|
return pack_bitarray(data)
|
||||||
|
|
||||||
|
|
||||||
class Int2Record(Record[NDArray[numpy.integer] | Sequence[int] | int, NDArray[numpy.int16]]):
|
class Int2Record(Record):
|
||||||
@classmethod
|
@classmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> NDArray[numpy.int16]:
|
def read_data(cls, stream: IO[bytes], size: int) -> NDArray[numpy.int16]:
|
||||||
return parse_int2(read(stream, size))
|
return parse_int2(read(stream, size))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def pack_data(cls: type[Self], data: NDArray[numpy.integer] | Sequence[int] | int) -> bytes:
|
def pack_data(cls, data: Sequence[int]) -> bytes:
|
||||||
return pack_int2(data)
|
return pack_int2(data)
|
||||||
|
|
||||||
|
|
||||||
class Int4Record(Record[NDArray[numpy.integer] | Sequence[int] | int, NDArray[numpy.int32]]):
|
class Int4Record(Record):
|
||||||
@classmethod
|
@classmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> NDArray[numpy.int32]:
|
def read_data(cls, stream: IO[bytes], size: int) -> NDArray[numpy.int32]:
|
||||||
return parse_int4(read(stream, size))
|
return parse_int4(read(stream, size))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def pack_data(cls: type[Self], data: NDArray[numpy.integer] | Sequence[int] | int) -> bytes:
|
def pack_data(cls, data: Sequence[int]) -> bytes:
|
||||||
return pack_int4(data)
|
return pack_int4(data)
|
||||||
|
|
||||||
|
|
||||||
class Real8Record(Record[Sequence[float] | float, NDArray[numpy.float64]]):
|
class Real8Record(Record):
|
||||||
@classmethod
|
@classmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> NDArray[numpy.float64]:
|
def read_data(cls, stream: IO[bytes], size: int) -> NDArray[numpy.float64]:
|
||||||
return parse_real8(read(stream, size))
|
return parse_real8(read(stream, size))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def pack_data(cls: type[Self], data: Sequence[float] | float) -> bytes:
|
def pack_data(cls, data: Sequence[int]) -> bytes:
|
||||||
return pack_real8(data)
|
return pack_real8(data)
|
||||||
|
|
||||||
|
|
||||||
class ASCIIRecord(Record[bytes, bytes]):
|
class ASCIIRecord(Record):
|
||||||
@classmethod
|
@classmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> bytes:
|
def read_data(cls, stream: IO[bytes], size: int) -> bytes:
|
||||||
return parse_ascii(read(stream, size))
|
return parse_ascii(read(stream, size))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def pack_data(cls: type[Self], data: bytes) -> bytes:
|
def pack_data(cls, data: bytes) -> bytes:
|
||||||
return pack_ascii(data)
|
return pack_ascii(data)
|
||||||
|
|
||||||
|
|
||||||
class DateTimeRecord(Record[Sequence[datetime], list[datetime]]):
|
class DateTimeRecord(Record):
|
||||||
@classmethod
|
@classmethod
|
||||||
def read_data(cls: type[Self], stream: IO[bytes], size: int) -> list[datetime]:
|
def read_data(cls, stream: IO[bytes], size: int) -> list[datetime]:
|
||||||
return parse_datetime(read(stream, size))
|
return parse_datetime(read(stream, size))
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def pack_data(cls: type[Self], data: Sequence[datetime]) -> bytes:
|
def pack_data(cls, data: Sequence[datetime]) -> bytes:
|
||||||
return pack_datetime(data)
|
return pack_datetime(data)
|
||||||
|
|
|
||||||
|
|
@ -1,12 +1,8 @@
|
||||||
"""
|
"""
|
||||||
Record type and tag definitions
|
Record type and tag definitions
|
||||||
"""
|
"""
|
||||||
from typing import Self
|
from typing import Sequence
|
||||||
from collections.abc import Sequence, Sized
|
|
||||||
import numpy
|
|
||||||
from numpy.typing import NDArray
|
|
||||||
|
|
||||||
from .basic import KlamathError
|
|
||||||
from .record import NoDataRecord, BitArrayRecord, Int2Record, Int4Record, Real8Record
|
from .record import NoDataRecord, BitArrayRecord, Int2Record, Int4Record, Real8Record
|
||||||
from .record import ASCIIRecord, DateTimeRecord
|
from .record import ASCIIRecord, DateTimeRecord
|
||||||
|
|
||||||
|
|
@ -148,18 +144,18 @@ class REFLIBS(ASCIIRecord):
|
||||||
tag = 0x1f06
|
tag = 0x1f06
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def check_size(cls: type[Self], size: int) -> None:
|
def check_size(cls, size: int):
|
||||||
if size != 0 and size % 44 != 0:
|
if size != 0 and size % 44 != 0:
|
||||||
raise KlamathError(f'Expected size to be multiple of 44, got {size}')
|
raise Exception(f'Expected size to be multiple of 44, got {size}')
|
||||||
|
|
||||||
|
|
||||||
class FONTS(ASCIIRecord):
|
class FONTS(ASCIIRecord):
|
||||||
tag = 0x2006
|
tag = 0x2006
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def check_size(cls: type[Self], size: int) -> None:
|
def check_size(cls, size: int):
|
||||||
if size != 0 and size % 44 != 0:
|
if size != 0 and size % 44 != 0:
|
||||||
raise KlamathError(f'Expected size to be multiple of 44, got {size}')
|
raise Exception(f'Expected size to be multiple of 44, got {size}')
|
||||||
|
|
||||||
|
|
||||||
class PATHTYPE(Int2Record):
|
class PATHTYPE(Int2Record):
|
||||||
|
|
@ -172,18 +168,18 @@ class GENERATIONS(Int2Record):
|
||||||
expected_size = 2
|
expected_size = 2
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def check_data(cls: type[Self], data: NDArray[numpy.integer] | Sequence[int] | int) -> None:
|
def check_data(cls, data: Sequence[int]):
|
||||||
if not isinstance(data, Sized) or len(data) != 1:
|
if len(data) != 1:
|
||||||
raise KlamathError(f'Expected exactly one integer, got {data}')
|
raise Exception(f'Expected exactly one integer, got {data}')
|
||||||
|
|
||||||
|
|
||||||
class ATTRTABLE(ASCIIRecord):
|
class ATTRTABLE(ASCIIRecord):
|
||||||
tag = 0x2306
|
tag = 0x2306
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def check_size(cls: type[Self], size: int) -> None:
|
def check_size(cls, size: int):
|
||||||
if size > 44:
|
if size > 44:
|
||||||
raise KlamathError(f'Expected size <= 44, got {size}')
|
raise Exception(f'Expected size <= 44, got {size}')
|
||||||
|
|
||||||
|
|
||||||
class STYPTABLE(ASCIIRecord):
|
class STYPTABLE(ASCIIRecord):
|
||||||
|
|
@ -270,9 +266,9 @@ class FORMAT(Int2Record):
|
||||||
expected_size = 2
|
expected_size = 2
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def check_data(cls: type[Self], data: NDArray[numpy.integer] | Sequence[int] | int) -> None:
|
def check_data(cls, data: Sequence[int]):
|
||||||
if not isinstance(data, Sized) or len(data) != 1:
|
if len(data) != 1:
|
||||||
raise KlamathError(f'Expected exactly one integer, got {data}')
|
raise Exception(f'Expected exactly one integer, got {data}')
|
||||||
|
|
||||||
|
|
||||||
class MASK(ASCIIRecord):
|
class MASK(ASCIIRecord):
|
||||||
|
|
|
||||||
|
|
@ -12,7 +12,7 @@ from .basic import decode_real8, encode_real8, parse_datetime
|
||||||
from .basic import KlamathError
|
from .basic import KlamathError
|
||||||
|
|
||||||
|
|
||||||
def test_parse_bitarray() -> None:
|
def test_parse_bitarray():
|
||||||
assert parse_bitarray(b'59') == 13625
|
assert parse_bitarray(b'59') == 13625
|
||||||
assert parse_bitarray(b'\0\0') == 0
|
assert parse_bitarray(b'\0\0') == 0
|
||||||
assert parse_bitarray(b'\xff\xff') == 65535
|
assert parse_bitarray(b'\xff\xff') == 65535
|
||||||
|
|
@ -26,7 +26,7 @@ def test_parse_bitarray() -> None:
|
||||||
parse_bitarray(b'')
|
parse_bitarray(b'')
|
||||||
|
|
||||||
|
|
||||||
def test_parse_int2() -> None:
|
def test_parse_int2():
|
||||||
assert_array_equal(parse_int2(b'59\xff\xff\0\0'), (13625, -1, 0))
|
assert_array_equal(parse_int2(b'59\xff\xff\0\0'), (13625, -1, 0))
|
||||||
|
|
||||||
# odd length
|
# odd length
|
||||||
|
|
@ -38,7 +38,7 @@ def test_parse_int2() -> None:
|
||||||
parse_int2(b'')
|
parse_int2(b'')
|
||||||
|
|
||||||
|
|
||||||
def test_parse_int4() -> None:
|
def test_parse_int4():
|
||||||
assert_array_equal(parse_int4(b'4321'), (875770417,))
|
assert_array_equal(parse_int4(b'4321'), (875770417,))
|
||||||
|
|
||||||
# length % 4 != 0
|
# length % 4 != 0
|
||||||
|
|
@ -50,7 +50,7 @@ def test_parse_int4() -> None:
|
||||||
parse_int4(b'')
|
parse_int4(b'')
|
||||||
|
|
||||||
|
|
||||||
def test_decode_real8() -> None:
|
def test_decode_real8():
|
||||||
# zeroes
|
# zeroes
|
||||||
assert decode_real8(numpy.array([0x0])) == 0
|
assert decode_real8(numpy.array([0x0])) == 0
|
||||||
assert decode_real8(numpy.array([1 << 63])) == 0 # negative
|
assert decode_real8(numpy.array([1 << 63])) == 0 # negative
|
||||||
|
|
@ -60,7 +60,7 @@ def test_decode_real8() -> None:
|
||||||
assert decode_real8(numpy.array([0xC120 << 48])) == -2.0
|
assert decode_real8(numpy.array([0xC120 << 48])) == -2.0
|
||||||
|
|
||||||
|
|
||||||
def test_parse_real8() -> None:
|
def test_parse_real8():
|
||||||
packed = struct.pack('>3Q', 0x0, 0x4110_0000_0000_0000, 0xC120_0000_0000_0000)
|
packed = struct.pack('>3Q', 0x0, 0x4110_0000_0000_0000, 0xC120_0000_0000_0000)
|
||||||
assert_array_equal(parse_real8(packed), (0.0, 1.0, -2.0))
|
assert_array_equal(parse_real8(packed), (0.0, 1.0, -2.0))
|
||||||
|
|
||||||
|
|
@ -73,7 +73,7 @@ def test_parse_real8() -> None:
|
||||||
parse_real8(b'')
|
parse_real8(b'')
|
||||||
|
|
||||||
|
|
||||||
def test_parse_ascii() -> None:
|
def test_parse_ascii():
|
||||||
# # empty data Now allowed!
|
# # empty data Now allowed!
|
||||||
# with pytest.raises(KlamathError):
|
# with pytest.raises(KlamathError):
|
||||||
# parse_ascii(b'')
|
# parse_ascii(b'')
|
||||||
|
|
@ -82,40 +82,40 @@ def test_parse_ascii() -> None:
|
||||||
assert parse_ascii(b'12345\0') == b'12345' # strips trailing null byte
|
assert parse_ascii(b'12345\0') == b'12345' # strips trailing null byte
|
||||||
|
|
||||||
|
|
||||||
def test_pack_bitarray() -> None:
|
def test_pack_bitarray():
|
||||||
packed = pack_bitarray(321)
|
packed = pack_bitarray(321)
|
||||||
assert len(packed) == 2
|
assert len(packed) == 2
|
||||||
assert packed == struct.pack('>H', 321)
|
assert packed == struct.pack('>H', 321)
|
||||||
|
|
||||||
|
|
||||||
def test_pack_int2() -> None:
|
def test_pack_int2():
|
||||||
packed = pack_int2((3, 2, 1))
|
packed = pack_int2((3, 2, 1))
|
||||||
assert len(packed) == 3 * 2
|
assert len(packed) == 3 * 2
|
||||||
assert packed == struct.pack('>3h', 3, 2, 1)
|
assert packed == struct.pack('>3h', 3, 2, 1)
|
||||||
assert pack_int2([-3, 2, -1]) == struct.pack('>3h', -3, 2, -1)
|
assert pack_int2([-3, 2, -1]) == struct.pack('>3h', -3, 2, -1)
|
||||||
|
|
||||||
|
|
||||||
def test_pack_int4() -> None:
|
def test_pack_int4():
|
||||||
packed = pack_int4((3, 2, 1))
|
packed = pack_int4((3, 2, 1))
|
||||||
assert len(packed) == 3 * 4
|
assert len(packed) == 3 * 4
|
||||||
assert packed == struct.pack('>3l', 3, 2, 1)
|
assert packed == struct.pack('>3l', 3, 2, 1)
|
||||||
assert pack_int4([-3, 2, -1]) == struct.pack('>3l', -3, 2, -1)
|
assert pack_int4([-3, 2, -1]) == struct.pack('>3l', -3, 2, -1)
|
||||||
|
|
||||||
|
|
||||||
def test_encode_real8() -> None:
|
def test_encode_real8():
|
||||||
assert encode_real8(numpy.array([0.0])) == 0
|
assert encode_real8(numpy.array([0.0])) == 0
|
||||||
arr = numpy.array((1.0, -2.0, 1e-9, 1e-3, 1e-12))
|
arr = numpy.array((1.0, -2.0, 1e-9, 1e-3, 1e-12))
|
||||||
assert_array_equal(decode_real8(encode_real8(arr)), arr)
|
assert_array_equal(decode_real8(encode_real8(arr)), arr)
|
||||||
|
|
||||||
|
|
||||||
def test_pack_real8() -> None:
|
def test_pack_real8():
|
||||||
reals = (0, 1, -1, 0.5, 1e-9, 1e-3, 1e-12)
|
reals = (0, 1, -1, 0.5, 1e-9, 1e-3, 1e-12)
|
||||||
packed = pack_real8(reals)
|
packed = pack_real8(reals)
|
||||||
assert len(packed) == len(reals) * 8
|
assert len(packed) == len(reals) * 8
|
||||||
assert_array_equal(parse_real8(packed), reals)
|
assert_array_equal(parse_real8(packed), reals)
|
||||||
|
|
||||||
|
|
||||||
def test_pack_ascii() -> None:
|
def test_pack_ascii():
|
||||||
assert pack_ascii(b'4321') == b'4321'
|
assert pack_ascii(b'4321') == b'4321'
|
||||||
assert pack_ascii(b'321') == b'321\0'
|
assert pack_ascii(b'321') == b'321\0'
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -45,48 +45,14 @@ classifiers = [
|
||||||
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
|
"License :: OSI Approved :: GNU General Public License v3 (GPLv3)",
|
||||||
"Topic :: Scientific/Engineering :: Electronic Design Automation (EDA)",
|
"Topic :: Scientific/Engineering :: Electronic Design Automation (EDA)",
|
||||||
]
|
]
|
||||||
requires-python = ">=3.11"
|
requires-python = ">=3.8"
|
||||||
include = [
|
include = [
|
||||||
"LICENSE.md"
|
"LICENSE.md"
|
||||||
]
|
]
|
||||||
dynamic = ["version"]
|
dynamic = ["version"]
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"numpy>=1.26",
|
"numpy~=1.21",
|
||||||
]
|
]
|
||||||
|
|
||||||
[tool.hatch.version]
|
[tool.hatch.version]
|
||||||
path = "klamath/__init__.py"
|
path = "klamath/__init__.py"
|
||||||
|
|
||||||
|
|
||||||
[tool.ruff]
|
|
||||||
exclude = [
|
|
||||||
".git",
|
|
||||||
"dist",
|
|
||||||
]
|
|
||||||
line-length = 145
|
|
||||||
indent-width = 4
|
|
||||||
lint.dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
|
|
||||||
lint.select = [
|
|
||||||
"NPY", "E", "F", "W", "B", "ANN", "UP", "SLOT", "SIM", "LOG",
|
|
||||||
"C4", "ISC", "PIE", "PT", "RET", "TCH", "PTH", "INT",
|
|
||||||
"ARG", "PL", "R", "TRY",
|
|
||||||
"G010", "G101", "G201", "G202",
|
|
||||||
"Q002", "Q003", "Q004",
|
|
||||||
]
|
|
||||||
lint.ignore = [
|
|
||||||
#"ANN001", # No annotation
|
|
||||||
"ANN002", # *args
|
|
||||||
"ANN003", # **kwargs
|
|
||||||
"ANN401", # Any
|
|
||||||
"ANN101", # self: Self
|
|
||||||
"SIM108", # single-line if / else assignment
|
|
||||||
"RET504", # x=y+z; return x
|
|
||||||
"PIE790", # unnecessary pass
|
|
||||||
"ISC003", # non-implicit string concatenation
|
|
||||||
"C408", # dict(x=y) instead of {'x': y}
|
|
||||||
"PLR09", # Too many xxx
|
|
||||||
"PLR2004", # magic number
|
|
||||||
"PLC0414", # import x as x
|
|
||||||
"TRY003", # Long exception message
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue