Coverage for yield_analysis_sdk\acp.py: 82%

1from typing import Optional 

2from virtuals_acp import VirtualsACP, ACPJob, ACPJobPhase, ACPMemo # type: ignore 

3from .type import ( 

4 AnalysisRequest, 

5 AnalysisResponse, 

6 VaultRegistrationRequest, 

7 VaultRegistrationResponse, 

8) 

9 

10 

11def extract_analysis_request(memos: list[ACPMemo]) -> Optional[AnalysisRequest]: 

12 """ 

13 Extract AnalysisRequest from memos by trying each negotiation memo until one works. 

14 

15 Args: 

16 memos: List of ACP memos to process 

17 

18 Returns: 

19 AnalysisRequest if a valid one is found, otherwise None 

20 """ 

21 if not memos: 

22 return None 

23 

24 for memo in memos: 

25 if memo.next_phase == ACPJobPhase.NEGOTIATION: 

26 if memo.content: 

27 try: 

28 return AnalysisRequest.model_validate_json(memo.content) 

29 except Exception: 

30 # Continue to next memo if this one fails to parse 

31 continue 

32 else: 

33 # Skip empty memos 

34 continue 

35 

36 # No valid memo found 

37 return None 

38 

39 

40def extract_analysis_response(memos: list[ACPMemo]) -> Optional[AnalysisResponse]: 

41 """ 

42 Extract AnalysisResponse from memos by trying each negotiation memo until one works. 

43 

44 Args: 

45 memos: List of ACP memos to process 

46 

47 Returns: 

48 AnalysisResponse if a valid one is found, otherwise None 

49 """ 

50 if not memos: 

51 return None 

52 

53 for memo in memos: 

54 if memo.next_phase == ACPJobPhase.COMPLETED: 

55 if memo.content: 

56 try: 

57 return AnalysisResponse.model_validate_json(memo.content) 

58 except Exception: 

59 # Continue to next memo if this one fails to parse 

60 continue 

61 else: 

62 # Skip empty memos 

63 continue 

64 

65 # No valid memo found 

66 return None 

67 

68 

69def extract_vault_registration_request( 

70 memos: list[ACPMemo], 

71) -> Optional[VaultRegistrationRequest]: 

72 """ 

73 Extract VaultRegistrationRequest from memos by trying each negotiation memo until one works. 

74 

75 Args: 

76 memos: List of ACP memos to process 

77 

78 Returns: 

79 VaultRegistrationRequest if a valid one is found, otherwise None 

80 """ 

81 if not memos: 

82 return None 

83 

84 for memo in memos: 

85 if memo.next_phase == ACPJobPhase.NEGOTIATION: 

86 if memo.content: 

87 try: 

88 return VaultRegistrationRequest.model_validate_json(memo.content) 

89 except Exception: 

90 # Continue to next memo if this one fails to parse 

91 continue 

92 else: 

93 # Skip empty memos 

94 continue 

95 

96 # No valid memo found 

97 return None 

98 

99 

100def extract_vault_registration_response( 

101 memos: list[ACPMemo], 

102) -> Optional[VaultRegistrationResponse]: 

103 """ 

104 Extract VaultRegistrationResponse from memos by trying each negotiation memo until one works. 

105 

106 Args: 

107 memos: List of ACP memos to process 

108 

109 Returns: 

110 VaultRegistrationResponse if a valid one is found, otherwise None 

111 """ 

112 

113 if not memos: 

114 return None 

115 

116 for memo in memos: 

117 if memo.next_phase == ACPJobPhase.COMPLETED: 

118 if memo.content: 

119 try: 

120 return VaultRegistrationResponse.model_validate_json(memo.content) 

121 except Exception: 

122 # Continue to next memo if this one fails to parse 

123 continue 

124 else: 

125 # Skip empty memos 

126 continue 

127 

128 # No valid memo found 

129 return None