Spaces:
Runtime error
Runtime error
| # ---------------------------------------------------------------------------- | |
| # | |
| # Welcome to Baml! To use this generated code, please run the following: | |
| # | |
| # $ pip install baml | |
| # | |
| # ---------------------------------------------------------------------------- | |
| # This file was generated by BAML: please do not edit it. Instead, edit the | |
| # BAML files and re-generate this code using: baml-cli generate | |
| # baml-cli is available with the baml package. | |
| import typing | |
| import typing_extensions | |
| import baml_py | |
| from . import stream_types, types, type_builder | |
| from .parser import LlmResponseParser, LlmStreamParser | |
| from .runtime import DoNotUseDirectlyCallManager, BamlCallOptions | |
| from .globals import DO_NOT_USE_DIRECTLY_UNLESS_YOU_KNOW_WHAT_YOURE_DOING_RUNTIME as __runtime__ | |
| class BamlSyncClient: | |
| __options: DoNotUseDirectlyCallManager | |
| __stream_client: "BamlStreamClient" | |
| __http_request: "BamlHttpRequestClient" | |
| __http_stream_request: "BamlHttpStreamRequestClient" | |
| __llm_response_parser: LlmResponseParser | |
| __llm_stream_parser: LlmStreamParser | |
| def __init__(self, options: DoNotUseDirectlyCallManager): | |
| self.__options = options | |
| self.__stream_client = BamlStreamClient(options) | |
| self.__http_request = BamlHttpRequestClient(options) | |
| self.__http_stream_request = BamlHttpStreamRequestClient(options) | |
| self.__llm_response_parser = LlmResponseParser(options) | |
| self.__llm_stream_parser = LlmStreamParser(options) | |
| def __getstate__(self): | |
| # Return state needed for pickling | |
| return {"options": self.__options} | |
| def __setstate__(self, state): | |
| # Restore state from pickling | |
| self.__options = state["options"] | |
| self.__stream_client = BamlStreamClient(self.__options) | |
| self.__http_request = BamlHttpRequestClient(self.__options) | |
| self.__http_stream_request = BamlHttpStreamRequestClient(self.__options) | |
| self.__llm_response_parser = LlmResponseParser(self.__options) | |
| self.__llm_stream_parser = LlmStreamParser(self.__options) | |
| def with_options(self, | |
| tb: typing.Optional[type_builder.TypeBuilder] = None, | |
| client_registry: typing.Optional[baml_py.baml_py.ClientRegistry] = None, | |
| client: typing.Optional[str] = None, | |
| collector: typing.Optional[typing.Union[baml_py.baml_py.Collector, typing.List[baml_py.baml_py.Collector]]] = None, | |
| env: typing.Optional[typing.Dict[str, typing.Optional[str]]] = None, | |
| tags: typing.Optional[typing.Dict[str, str]] = None, | |
| on_tick: typing.Optional[typing.Callable[[str, baml_py.baml_py.FunctionLog], None]] = None, | |
| ) -> "BamlSyncClient": | |
| options: BamlCallOptions = {} | |
| if tb is not None: | |
| options["tb"] = tb | |
| if client_registry is not None: | |
| options["client_registry"] = client_registry | |
| if client is not None: | |
| options["client"] = client | |
| if collector is not None: | |
| options["collector"] = collector | |
| if env is not None: | |
| options["env"] = env | |
| if tags is not None: | |
| options["tags"] = tags | |
| if on_tick is not None: | |
| options["on_tick"] = on_tick | |
| return BamlSyncClient(self.__options.merge_options(options)) | |
| def stream(self): | |
| return self.__stream_client | |
| def request(self): | |
| return self.__http_request | |
| def stream_request(self): | |
| return self.__http_stream_request | |
| def parse(self): | |
| return self.__llm_response_parser | |
| def parse_stream(self): | |
| return self.__llm_stream_parser | |
| def AssessDetections(self, mission: str,detections: typing.List["types.DetectionInfo"],frame_image: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> typing.List["types.DetectionVerdict"]: | |
| # Check if on_tick is provided | |
| if 'on_tick' in baml_options: | |
| __stream__ = self.stream.AssessDetections(mission=mission,detections=detections,frame_image=frame_image, | |
| baml_options=baml_options) | |
| return __stream__.get_final_response() | |
| else: | |
| # Original non-streaming code | |
| __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="AssessDetections", args={ | |
| "mission": mission,"detections": detections,"frame_image": frame_image, | |
| }) | |
| return typing.cast(typing.List["types.DetectionVerdict"], __result__.cast_to(types, types, stream_types, False, __runtime__)) | |
| def PlanMission(self, mission_text: str, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> types.MissionPlan: | |
| # Check if on_tick is provided | |
| if 'on_tick' in baml_options: | |
| __stream__ = self.stream.PlanMission(mission_text=mission_text, | |
| baml_options=baml_options) | |
| return __stream__.get_final_response() | |
| else: | |
| # Original non-streaming code | |
| __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="PlanMission", args={ | |
| "mission_text": mission_text, | |
| }) | |
| return typing.cast(types.MissionPlan, __result__.cast_to(types, types, stream_types, False, __runtime__)) | |
| def SuggestMissions(self, frame: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> typing.List["types.MissionSuggestion"]: | |
| # Check if on_tick is provided | |
| if 'on_tick' in baml_options: | |
| __stream__ = self.stream.SuggestMissions(frame=frame, | |
| baml_options=baml_options) | |
| return __stream__.get_final_response() | |
| else: | |
| # Original non-streaming code | |
| __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="SuggestMissions", args={ | |
| "frame": frame, | |
| }) | |
| return typing.cast(typing.List["types.MissionSuggestion"], __result__.cast_to(types, types, stream_types, False, __runtime__)) | |
| class BamlStreamClient: | |
| __options: DoNotUseDirectlyCallManager | |
| def __init__(self, options: DoNotUseDirectlyCallManager): | |
| self.__options = options | |
| def AssessDetections(self, mission: str,detections: typing.List["types.DetectionInfo"],frame_image: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.BamlSyncStream[typing.List["stream_types.DetectionVerdict"], typing.List["types.DetectionVerdict"]]: | |
| __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="AssessDetections", args={ | |
| "mission": mission,"detections": detections,"frame_image": frame_image, | |
| }) | |
| return baml_py.BamlSyncStream[typing.List["stream_types.DetectionVerdict"], typing.List["types.DetectionVerdict"]]( | |
| __result__, | |
| lambda x: typing.cast(typing.List["stream_types.DetectionVerdict"], x.cast_to(types, types, stream_types, True, __runtime__)), | |
| lambda x: typing.cast(typing.List["types.DetectionVerdict"], x.cast_to(types, types, stream_types, False, __runtime__)), | |
| __ctx__, | |
| ) | |
| def PlanMission(self, mission_text: str, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.BamlSyncStream[stream_types.MissionPlan, types.MissionPlan]: | |
| __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="PlanMission", args={ | |
| "mission_text": mission_text, | |
| }) | |
| return baml_py.BamlSyncStream[stream_types.MissionPlan, types.MissionPlan]( | |
| __result__, | |
| lambda x: typing.cast(stream_types.MissionPlan, x.cast_to(types, types, stream_types, True, __runtime__)), | |
| lambda x: typing.cast(types.MissionPlan, x.cast_to(types, types, stream_types, False, __runtime__)), | |
| __ctx__, | |
| ) | |
| def SuggestMissions(self, frame: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.BamlSyncStream[typing.List["stream_types.MissionSuggestion"], typing.List["types.MissionSuggestion"]]: | |
| __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="SuggestMissions", args={ | |
| "frame": frame, | |
| }) | |
| return baml_py.BamlSyncStream[typing.List["stream_types.MissionSuggestion"], typing.List["types.MissionSuggestion"]]( | |
| __result__, | |
| lambda x: typing.cast(typing.List["stream_types.MissionSuggestion"], x.cast_to(types, types, stream_types, True, __runtime__)), | |
| lambda x: typing.cast(typing.List["types.MissionSuggestion"], x.cast_to(types, types, stream_types, False, __runtime__)), | |
| __ctx__, | |
| ) | |
| class BamlHttpRequestClient: | |
| __options: DoNotUseDirectlyCallManager | |
| def __init__(self, options: DoNotUseDirectlyCallManager): | |
| self.__options = options | |
| def AssessDetections(self, mission: str,detections: typing.List["types.DetectionInfo"],frame_image: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.baml_py.HTTPRequest: | |
| __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="AssessDetections", args={ | |
| "mission": mission,"detections": detections,"frame_image": frame_image, | |
| }, mode="request") | |
| return __result__ | |
| def PlanMission(self, mission_text: str, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.baml_py.HTTPRequest: | |
| __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="PlanMission", args={ | |
| "mission_text": mission_text, | |
| }, mode="request") | |
| return __result__ | |
| def SuggestMissions(self, frame: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.baml_py.HTTPRequest: | |
| __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="SuggestMissions", args={ | |
| "frame": frame, | |
| }, mode="request") | |
| return __result__ | |
| class BamlHttpStreamRequestClient: | |
| __options: DoNotUseDirectlyCallManager | |
| def __init__(self, options: DoNotUseDirectlyCallManager): | |
| self.__options = options | |
| def AssessDetections(self, mission: str,detections: typing.List["types.DetectionInfo"],frame_image: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.baml_py.HTTPRequest: | |
| __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="AssessDetections", args={ | |
| "mission": mission,"detections": detections,"frame_image": frame_image, | |
| }, mode="stream") | |
| return __result__ | |
| def PlanMission(self, mission_text: str, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.baml_py.HTTPRequest: | |
| __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="PlanMission", args={ | |
| "mission_text": mission_text, | |
| }, mode="stream") | |
| return __result__ | |
| def SuggestMissions(self, frame: baml_py.Image, | |
| baml_options: BamlCallOptions = {}, | |
| ) -> baml_py.baml_py.HTTPRequest: | |
| __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="SuggestMissions", args={ | |
| "frame": frame, | |
| }, mode="stream") | |
| return __result__ | |
| b = BamlSyncClient(DoNotUseDirectlyCallManager({})) |