Coverage for fastagency/ui/mesop/main.py: 68%

145 statements  

« prev     ^ index     » next       coverage.py v7.8.0, created at 2025-04-19 12:16 +0000

1import time 1bcd

2from collections.abc import Iterator 1bcd

3from dataclasses import dataclass 1bcd

4from typing import TYPE_CHECKING, Callable, Optional 1bcd

5 

6import mesop as me 1bcd

7 

8from fastagency.base import ProviderProtocol 1bcd

9 

10from ...logging import get_logger 1bcd

11from .auth import AuthProtocol 1bcd

12from .data_model import Conversation, State 1bcd

13from .message import consume_responses, message_box 1bcd

14from .send_prompt import send_prompt_to_autogen 1bcd

15from .styles import MesopHomePageStyles 1bcd

16 

17if TYPE_CHECKING: 1bcd

18 from .mesop import MesopUI 

19 

20__all__ = ["me"] 1bcd

21 

22# Get the logger 

23logger = get_logger(__name__) 1bcd

24 

25 

26DEFAULT_SECURITY_POLICY = me.SecurityPolicy( 1bcd

27 allowed_script_srcs=["https://cdn.jsdelivr.net"], 

28 allowed_iframe_parents=["https://fastagency.ai"], 

29) 

30 

31 

32def create_home_page( 1bcd

33 ui: "MesopUI", 

34 *, 

35 styles: Optional[MesopHomePageStyles] = None, 

36 security_policy: Optional[me.SecurityPolicy] = None, 

37 auth: Optional[AuthProtocol] = None, 

38) -> Callable[[], None]: 

39 mhp = MesopHomePage(ui, styles=styles, security_policy=security_policy, auth=auth) 1bcd

40 

41 return mhp.build() 1bcd

42 

43 

44@dataclass 1bcd

45class MesopHomePageParams: 1bcd

46 # header_title: str = "FastAgency - Mesop" 

47 conv_starter_text: str = "Select workflow to use with FastAgency team" 1bcd

48 no_workflows_text: str = "No workflows found, click to retry" 1bcd

49 workflows_exception_text: str = ( 1bcd

50 "An exception occurred while discovering workflows, click to retry" 

51 ) 

52 

53 

54class MesopHomePage: 1bcd

55 def __init__( 1bcd

56 self, 

57 ui: "MesopUI", 

58 *, 

59 params: Optional[MesopHomePageParams] = None, 

60 styles: Optional[MesopHomePageStyles] = None, 

61 security_policy: Optional[me.SecurityPolicy] = None, 

62 auth: Optional[AuthProtocol] = None, 

63 ) -> None: 

64 self._ui = ui 1bcd

65 self._params = params or MesopHomePageParams() 1bcd

66 self._styles = styles or MesopHomePageStyles() 1bcd

67 self.auth = auth 1bcd

68 self._security_policy = self._create_security_policy( 1bcd

69 base_policy=security_policy or DEFAULT_SECURITY_POLICY, auth=auth 

70 ) 

71 

72 def _create_security_policy( 1bcd

73 self, base_policy: me.SecurityPolicy, auth: Optional[AuthProtocol] 

74 ) -> me.SecurityPolicy: 

75 """Create a security policy by combining the base policy with auth-specific policies. 

76 

77 Args: 

78 base_policy: The base security policy to start with 

79 auth: Optional authentication protocol implementation 

80 

81 Returns: 

82 The final security policy 

83 """ 

84 if auth is None: 1bcd

85 return base_policy 1bcd

86 

87 return auth.create_security_policy(base_policy) 1bcd

88 

89 def build(self) -> Callable[[], None]: 1bcd

90 stylesheets = ( 1bcd

91 self._styles.stylesheets + self._styles.firebase_stylesheets 

92 if self.auth 

93 else self._styles.stylesheets 

94 ) 

95 

96 @me.page( # type: ignore[misc] 1bcd

97 path="/", 

98 stylesheets=stylesheets, 

99 security_policy=self._security_policy, 

100 ) 

101 def home_page() -> None: 1bcd

102 self.home_page() 1a

103 

104 return home_page # type: ignore[no-any-return] 1bcd

105 

106 def home_page(self) -> None: 1bcd

107 try: 1a

108 state = me.state(State) 1a

109 if self.auth and not state.authenticated_user: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true1a

110 self.auth.auth_component() 

111 else: 

112 with me.box(style=self._styles.root): 1a

113 self.past_conversations_box() 1a

114 if state.in_conversation: 1a

115 self.conversation_box() 1a

116 else: 

117 self.conversation_starter_box() 1a

118 if self.auth and state.authenticated_user: 118 ↛ anywhereline 118 didn't jump anywhere: it always raised an exception.1a

119 self.auth.auth_component() 

120 except Exception as e: 

121 # import traceback 

122 # tb = traceback.format_exc() 

123 # print("Inside except") 

124 # print(tb) 

125 logger.error(f"home_page(): Error rendering home page: {e}") 

126 me.text(text="Error: Something went wrong, please check logs for details.") 

127 

128 def header(self) -> None: 1bcd

129 with me.box( 1a

130 style=self._styles.header, 

131 ): 

132 me.text( 1a

133 self._ui.app.title, 

134 style=self._styles.header_text, 

135 ) 

136 

137 def past_conversations_box(self) -> None: 1bcd

138 def conversation_display_title(full_name: str, max_length: int) -> str: 1a

139 if len(full_name) <= max_length: 

140 return full_name 

141 else: 

142 return full_name[: max_length - 3] + "..." 

143 

144 def select_past_conversation(ev: me.ClickEvent) -> Iterator[None]: 1a

145 id = ev.key 

146 state = me.state(State) 

147 conversations_with_id = list( 

148 filter(lambda c: c.id == id, state.past_conversations) 

149 ) 

150 conversation = conversations_with_id[0] 

151 state.conversation = conversation 

152 state.in_conversation = True 

153 yield 

154 time.sleep(1) 

155 yield 

156 me.scroll_into_view(key="end_of_messages") 

157 yield 

158 

159 def on_show_hide(ev: me.ClickEvent) -> None: 1a

160 state.hide_past = not state.hide_past 

161 

162 def on_start_new_conversation(ev: me.ClickEvent) -> None: 1a

163 state.in_conversation = False 

164 state.prompt = "" 

165 

166 state = me.state(State) 1a

167 style = ( 1a

168 self._styles.past_chats_hide 

169 if state.hide_past 

170 else self._styles.past_chats_show 

171 ) 

172 with me.box(style=style): 1a

173 with me.box( 1a

174 style=self._styles.past_chats_inner, 

175 ): 

176 with me.content_button( 1a

177 on_click=on_show_hide, disabled=not state.past_conversations 

178 ): 

179 me.icon("menu") 1a

180 with me.content_button( 1a

181 on_click=on_start_new_conversation, 

182 disabled=not state.conversation.completed, 

183 ): 

184 me.icon("rate_review") 1a

185 if not state.hide_past: 185 ↛ anywhereline 185 didn't jump anywhere: it always raised an exception.1a

186 for conversation in state.past_conversations: 

187 with me.box( 

188 key=conversation.id, # they are GUIDs so should not clash with anything other on the page 

189 on_click=select_past_conversation, 

190 style=self._styles.past_chats_conv, 

191 ): 

192 me.text( 

193 text=conversation_display_title(conversation.title, 128) 

194 ) 

195 

196 def conversation_starter_box(self) -> None: 1bcd

197 def retry(ev: me.ClickEvent) -> None: 1a

198 state = me.state(State) 

199 try: 

200 state.available_workflows = provider.names 

201 state.available_workflows_exception = False 

202 except Exception: 

203 state.available_workflows_exception = False 

204 

205 provider = self.get_provider() 1a

206 with me.box(style=self._styles.chat_starter): 1a

207 self.header() 1a

208 with me.box( 1a

209 style=self._styles.conv_starter, 

210 ): 

211 me.text( 1a

212 self._params.conv_starter_text, 

213 style=self._styles.conv_starter_text, 

214 ) 

215 with me.box(style=self._styles.conv_starter_wf_box): 1a

216 state = me.state(State) 1a

217 if not state.available_workflows_initialized: 217 ↛ 226line 217 didn't jump to line 226 because the condition on line 217 was always true1a

218 state.available_workflows_initialized = True 1a

219 try: 1a

220 state.available_workflows = provider.names 1a

221 state.available_workflows_exception = False 1a

222 except Exception: 

223 state.available_workflows = [] 

224 state.available_workflows_exception = True 

225 

226 names = state.available_workflows 1a

227 if names and not state.available_workflows_exception: 227 ↛ 237line 227 didn't jump to line 237 because the condition on line 227 was always true1a

228 try: 1a

229 for wf_name in names: 1a

230 wf_description = provider.get_description(wf_name) 1a

231 with me.content_button( 1a

232 key=wf_name, on_click=lambda e: self.send_prompt(e) 

233 ): 

234 me.text(wf_description) 1a

235 except Exception: 

236 state.available_workflows_exception = True 

237 if not names or state.available_workflows_exception: 237 ↛ anywhereline 237 didn't jump anywhere: it always raised an exception.1a

238 with me.content_button(on_click=retry): 

239 if state.available_workflows_exception: 

240 me.text(self._params.workflows_exception_text) 

241 else: 

242 me.text(self._params.no_workflows_text) 

243 

244 def get_provider(self) -> ProviderProtocol: 1bcd

245 ui = self._ui 1a

246 return ui.app.provider 1a

247 

248 def send_prompt(self, ev: me.ClickEvent) -> Iterator[None]: 1bcd

249 name = ev.key 1a

250 provider = self.get_provider() 1a

251 state = me.state(State) 1a

252 conversation = Conversation( 1a

253 title="New Conversation", completed=False, waiting_for_feedback=False 

254 ) 

255 state.conversation = conversation 1a

256 state.in_conversation = True 1a

257 yield 1a

258 responses = send_prompt_to_autogen(provider=provider, name=name) 1a

259 yield from consume_responses(responses) 1a

260 try: 1a

261 state.available_workflows = provider.names 1a

262 except Exception: 

263 state.available_workflows = [] 

264 state.available_workflows_exception = True 

265 

266 def conversation_box(self) -> None: 1bcd

267 state = me.state(State) 1a

268 conversation = state.conversation 1a

269 with me.box(style=self._styles.chat_starter): 1a

270 self.header() 1a

271 messages = conversation.messages 1a

272 with me.box( 1a

273 style=self._styles.conv_list, 

274 ): 

275 me.box( 1a

276 key="conversationtop", 

277 style=self._styles.conv_top, 

278 ) 

279 for message in messages: 1a

280 message_box( 1a

281 message, conversation.is_from_the_past, styles=self._styles 

282 ) 

283 if messages: 283 ↛ exitline 283 didn't return from function 'conversation_box' because the condition on line 283 was always true1a

284 me.box( 1a

285 key="end_of_messages", 

286 style=self._styles.conv_top, 

287 )