Coverage for src / invariant / types.py: 97.30%

1"""Domain types implementing ICacheable protocol. 

2 

3Native types (int, str, Decimal, dict, list) are supported directly without wrappers. 

4Only domain types that require custom serialization implement ICacheable. 

5""" 

6 

7import hashlib 

8from typing import BinaryIO 

9 

10from invariant.protocol import ICacheable 

11 

12 

13class Polynomial(ICacheable): 

14 """A cacheable polynomial type. 

15 

16 Represents a polynomial as a tuple of integer coefficients, where the index 

17 represents the degree (coefficient at index i is the coefficient of x^i). 

18 

19 Canonical form: Trailing zeros are stripped to ensure a unique representation. 

20 For example, [1, 2, 0, 0] is canonicalized to [1, 2]. 

21 """ 

22 

23 def __init__(self, coefficients: tuple[int, ...] | list[int]) -> None: 

24 """Initialize with coefficient list. 

25 

26 Args: 

27 coefficients: List or tuple of integer coefficients. Index i represents 

28 the coefficient of x^i. Trailing zeros are automatically 

29 stripped for canonical form. 

30 """ 

31 # Convert to tuple and strip trailing zeros for canonical form 

32 coeffs = tuple(coefficients) 

33 # Strip trailing zeros 

34 while len(coeffs) > 0 and coeffs[-1] == 0: 

35 coeffs = coeffs[:-1] 

36 # If all zeros, keep at least one zero 

37 if len(coeffs) == 0: 

38 coeffs = (0,) 

39 self.coefficients = coeffs 

40 

41 def get_stable_hash(self) -> str: 

42 """Return SHA-256 hash of the canonical coefficient tuple.""" 

43 # Hash the canonical coefficient tuple 

44 coeff_bytes = b",".join(str(c).encode("utf-8") for c in self.coefficients) 

45 return hashlib.sha256(coeff_bytes).hexdigest() 

46 

47 def to_stream(self, stream: BinaryIO) -> None: 

48 """Serialize polynomial to stream. 

49 

50 Format: length-prefixed sequence of 8-byte signed integers (big-endian). 

51 """ 

52 # Write length (number of coefficients) 

53 stream.write(len(self.coefficients).to_bytes(8, byteorder="big", signed=False)) 

54 # Write each coefficient as 8-byte signed big-endian integer 

55 for coeff in self.coefficients: 

56 stream.write(coeff.to_bytes(8, byteorder="big", signed=True)) 

57 

58 @classmethod 

59 def from_stream(cls, stream: BinaryIO) -> "Polynomial": 

60 """Deserialize polynomial from stream. 

61 

62 Reads length-prefixed sequence of 8-byte signed integers and strips 

63 trailing zeros for canonical form. 

64 """ 

65 # Read length 

66 length = int.from_bytes(stream.read(8), byteorder="big", signed=False) 

67 # Read coefficients 

68 coefficients = [] 

69 for _ in range(length): 

70 coeff = int.from_bytes(stream.read(8), byteorder="big", signed=True) 

71 coefficients.append(coeff) 

72 # Create polynomial (constructor will strip trailing zeros) 

73 return cls(tuple(coefficients)) 

74 

75 def __eq__(self, other: object) -> bool: 

76 """Equality comparison.""" 

77 if not isinstance(other, Polynomial): 

78 return False 

79 return self.coefficients == other.coefficients 

80 

81 def __repr__(self) -> str: 

82 """String representation.""" 

83 return f"Polynomial({list(self.coefficients)})" 

84 

85 def to_json_value(self) -> dict: 

86 """Return JSON-serializable dict for graph serialization (IJsonRepresentable).""" 

87 return {"coefficients": list(self.coefficients)} 

88 

89 @classmethod 

90 def from_json_value(cls, obj: dict) -> "Polynomial": 

91 """Reconstruct from JSON dict (IJsonRepresentable).""" 

92 if "coefficients" not in obj: 

93 raise ValueError("Polynomial from_json_value requires 'coefficients' key") 

94 return cls(obj["coefficients"])