""" gRPC client for SDK Bridge communication """ import grpc from typing import Optional, List import structlog from config import settings # Import generated protobuf classes from protos import camera_pb2, camera_pb2_grpc from protos import monitor_pb2, monitor_pb2_grpc from protos import crossswitch_pb2, crossswitch_pb2_grpc from protos import action_mapping_pb2, action_mapping_pb2_grpc from protos import configuration_pb2, configuration_pb2_grpc logger = structlog.get_logger() class SDKBridgeClient: """gRPC client for communicating with SDK Bridge""" def __init__(self): self._channel: Optional[grpc.aio.Channel] = None self._camera_stub = None self._monitor_stub = None self._crossswitch_stub = None self._action_mapping_stub = None self._configuration_stub = None async def connect(self): """Initialize gRPC channel to SDK Bridge""" try: logger.info("sdk_bridge_connecting", url=settings.sdk_bridge_url) # Create async gRPC channel self._channel = grpc.aio.insecure_channel( settings.sdk_bridge_url, options=[ ('grpc.max_send_message_length', 50 * 1024 * 1024), # 50MB ('grpc.max_receive_message_length', 50 * 1024 * 1024), # 50MB ('grpc.keepalive_time_ms', 30000), # 30 seconds ('grpc.keepalive_timeout_ms', 10000), # 10 seconds ] ) # Initialize service stubs self._camera_stub = camera_pb2_grpc.CameraServiceStub(self._channel) self._monitor_stub = monitor_pb2_grpc.MonitorServiceStub(self._channel) self._crossswitch_stub = crossswitch_pb2_grpc.CrossSwitchServiceStub(self._channel) self._action_mapping_stub = action_mapping_pb2_grpc.ActionMappingServiceStub(self._channel) self._configuration_stub = configuration_pb2_grpc.ConfigurationServiceStub(self._channel) logger.info("sdk_bridge_connected") except Exception as e: logger.error("sdk_bridge_connection_failed", error=str(e)) raise async def close(self): """Close gRPC channel""" try: if self._channel: await self._channel.close() logger.info("sdk_bridge_closed") except Exception as e: logger.error("sdk_bridge_close_failed", error=str(e)) async def health_check(self) -> dict: """Check SDK Bridge health""" try: logger.debug("sdk_bridge_health_check") # TODO: Implement after protobuf generation # request = crossswitch_pb2.Empty() # response = await self._crossswitch_stub.HealthCheck(request, timeout=5.0) # return { # "is_healthy": response.is_healthy, # "sdk_status": response.sdk_status, # "geviserver_host": response.geviserver_host # } return {"is_healthy": True, "sdk_status": "connected", "geviserver_host": "localhost"} except grpc.RpcError as e: logger.error("sdk_bridge_health_check_failed", error=str(e)) return {"is_healthy": False, "sdk_status": "error", "error": str(e)} async def list_cameras(self) -> List[dict]: """List all cameras from GeViServer""" try: logger.debug("sdk_bridge_list_cameras") request = camera_pb2.ListCamerasRequest() response = await self._camera_stub.ListCameras(request, timeout=10.0) return [ { "id": camera.id, "name": camera.name, "description": camera.description, "has_ptz": camera.has_ptz, "has_video_sensor": camera.has_video_sensor, "status": camera.status, "last_seen": None # TODO: Convert protobuf timestamp to datetime } for camera in response.cameras ] except grpc.RpcError as e: logger.error("sdk_bridge_list_cameras_failed", error=str(e)) raise async def get_camera(self, camera_id: int) -> Optional[dict]: """Get camera details""" try: logger.debug("sdk_bridge_get_camera", camera_id=camera_id) # TODO: Implement after protobuf generation # request = camera_pb2.GetCameraRequest(camera_id=camera_id) # response = await self._camera_stub.GetCamera(request, timeout=5.0) # return { # "id": response.id, # "name": response.name, # "description": response.description, # "has_ptz": response.has_ptz, # "has_video_sensor": response.has_video_sensor, # "status": response.status # } return None # Placeholder except grpc.RpcError as e: if e.code() == grpc.StatusCode.NOT_FOUND: return None logger.error("sdk_bridge_get_camera_failed", camera_id=camera_id, error=str(e)) raise async def list_monitors(self) -> List[dict]: """List all monitors from GeViServer""" try: logger.debug("sdk_bridge_list_monitors") request = monitor_pb2.ListMonitorsRequest() response = await self._monitor_stub.ListMonitors(request, timeout=10.0) return [ { "id": monitor.id, "name": monitor.name, "description": monitor.description, "is_active": monitor.is_active, "current_camera_id": monitor.current_camera_id, "status": monitor.status } for monitor in response.monitors ] except grpc.RpcError as e: logger.error("sdk_bridge_list_monitors_failed", error=str(e)) raise async def execute_crossswitch(self, camera_id: int, monitor_id: int, mode: int = 0) -> dict: """Execute cross-switch operation""" try: logger.info("sdk_bridge_crossswitch", camera_id=camera_id, monitor_id=monitor_id, mode=mode) request = crossswitch_pb2.CrossSwitchRequest( camera_id=camera_id, monitor_id=monitor_id, mode=mode ) response = await self._crossswitch_stub.ExecuteCrossSwitch(request, timeout=10.0) return { "success": response.success, "message": response.message, "camera_id": response.camera_id, "monitor_id": response.monitor_id } except grpc.RpcError as e: logger.error("sdk_bridge_crossswitch_failed", error=str(e)) raise async def clear_monitor(self, monitor_id: int) -> dict: """Clear monitor (stop video)""" try: logger.info("sdk_bridge_clear_monitor", monitor_id=monitor_id) request = crossswitch_pb2.ClearMonitorRequest(monitor_id=monitor_id) response = await self._crossswitch_stub.ClearMonitor(request, timeout=10.0) return { "success": response.success, "message": response.message, "monitor_id": response.monitor_id } except grpc.RpcError as e: logger.error("sdk_bridge_clear_monitor_failed", error=str(e)) raise async def get_routing_state(self) -> dict: """Get current routing state""" try: logger.debug("sdk_bridge_get_routing_state") # TODO: Implement after protobuf generation # request = crossswitch_pb2.GetRoutingStateRequest() # response = await self._crossswitch_stub.GetRoutingState(request, timeout=10.0) # return { # "routes": [ # { # "camera_id": route.camera_id, # "monitor_id": route.monitor_id, # "camera_name": route.camera_name, # "monitor_name": route.monitor_name # } # for route in response.routes # ], # "total_routes": response.total_routes # } return {"routes": [], "total_routes": 0} # Placeholder except grpc.RpcError as e: logger.error("sdk_bridge_get_routing_state_failed", error=str(e)) raise async def get_action_mappings(self, enabled_only: bool = False) -> dict: """Get action mappings from GeViServer via SDK Bridge""" try: logger.debug("sdk_bridge_get_action_mappings", enabled_only=enabled_only) request = action_mapping_pb2.GetActionMappingsRequest(enabled_only=enabled_only) response = await self._action_mapping_stub.GetActionMappings(request, timeout=30.0) return { "mappings": [ { "id": mapping.id, "name": mapping.name, "description": mapping.description, "input_action": mapping.input_action, "output_actions": list(mapping.output_actions), "enabled": mapping.enabled, "execution_count": mapping.execution_count, "last_executed": mapping.last_executed if mapping.last_executed else None, "created_at": mapping.created_at, "updated_at": mapping.updated_at } for mapping in response.mappings ], "total_count": response.total_count, "enabled_count": response.enabled_count, "disabled_count": response.disabled_count } except grpc.RpcError as e: logger.error("sdk_bridge_get_action_mappings_failed", error=str(e)) raise async def read_configuration(self) -> dict: """Read and parse configuration from GeViServer""" try: logger.debug("sdk_bridge_read_configuration") request = configuration_pb2.ReadConfigurationRequest() response = await self._configuration_stub.ReadConfiguration(request, timeout=30.0) return { "success": response.success, "error_message": response.error_message if response.error_message else None, "file_size": response.file_size, "header": response.header, "nodes": [ { "start_offset": node.start_offset, "end_offset": node.end_offset, "node_type": node.node_type, "name": node.name if node.name else None, "value": node.value if node.value else None, "value_type": node.value_type if node.value_type else None } for node in response.nodes ], "statistics": { "total_nodes": response.statistics.total_nodes, "boolean_count": response.statistics.boolean_count, "integer_count": response.statistics.integer_count, "string_count": response.statistics.string_count, "property_count": response.statistics.property_count, "marker_count": response.statistics.marker_count, "rules_section_count": response.statistics.rules_section_count } } except grpc.RpcError as e: logger.error("sdk_bridge_read_configuration_failed", error=str(e)) raise async def export_configuration_json(self) -> dict: """Export configuration as JSON""" try: logger.debug("sdk_bridge_export_configuration_json") request = configuration_pb2.ExportJsonRequest() response = await self._configuration_stub.ExportConfigurationJson(request, timeout=30.0) return { "success": response.success, "error_message": response.error_message if response.error_message else None, "json_data": response.json_data, "json_size": response.json_size } except grpc.RpcError as e: logger.error("sdk_bridge_export_configuration_json_failed", error=str(e)) raise async def modify_configuration(self, modifications: List[dict]) -> dict: """Modify configuration and write back to server""" try: logger.info("sdk_bridge_modify_configuration", count=len(modifications)) request = configuration_pb2.ModifyConfigurationRequest() for mod in modifications: modification = configuration_pb2.NodeModification( start_offset=mod["start_offset"], node_type=mod["node_type"], new_value=mod["new_value"] ) request.modifications.append(modification) response = await self._configuration_stub.ModifyConfiguration(request, timeout=60.0) return { "success": response.success, "error_message": response.error_message if response.error_message else None, "modifications_applied": response.modifications_applied } except grpc.RpcError as e: logger.error("sdk_bridge_modify_configuration_failed", error=str(e)) raise async def import_configuration(self, json_data: str) -> dict: """Import complete configuration from JSON and write to GeViServer""" try: logger.info("sdk_bridge_import_configuration", json_size=len(json_data)) request = configuration_pb2.ImportConfigurationRequest(json_data=json_data) response = await self._configuration_stub.ImportConfiguration(request, timeout=60.0) return { "success": response.success, "error_message": response.error_message if response.error_message else None, "bytes_written": response.bytes_written, "nodes_imported": response.nodes_imported } except grpc.RpcError as e: logger.error("sdk_bridge_import_configuration_failed", error=str(e)) raise async def read_action_mappings(self) -> dict: """ Read ONLY action mappings (Rules markers) from GeViServer Much faster than full configuration export - selective parsing Returns structured format with input_actions and output_actions with parameters """ try: logger.info("sdk_bridge_read_action_mappings") request = configuration_pb2.ReadActionMappingsRequest() response = await self._configuration_stub.ReadActionMappings(request, timeout=30.0) # Convert protobuf response to dict with structured format mappings = [] for mapping in response.mappings: # Convert input actions with parameters input_actions = [] for action_def in mapping.input_actions: parameters = {} for param in action_def.parameters: parameters[param.name] = param.value input_actions.append({ "action": action_def.action, "parameters": parameters }) # Convert output actions with parameters output_actions = [] for action_def in mapping.output_actions: parameters = {} for param in action_def.parameters: parameters[param.name] = param.value output_actions.append({ "action": action_def.action, "parameters": parameters }) mappings.append({ "name": mapping.name, "input_actions": input_actions, "output_actions": output_actions, "start_offset": mapping.start_offset, "end_offset": mapping.end_offset, # Keep old format for backward compatibility "actions": list(mapping.actions) }) return { "success": response.success, "error_message": response.error_message if response.error_message else None, "mappings": mappings, "total_count": response.total_count } except grpc.RpcError as e: logger.error("sdk_bridge_read_action_mappings_failed", error=str(e)) raise async def read_specific_markers(self, marker_names: List[str]) -> dict: """ Read specific configuration markers by name Extensible method for reading any configuration type """ try: logger.info("sdk_bridge_read_specific_markers", markers=marker_names) request = configuration_pb2.ReadSpecificMarkersRequest(marker_names=marker_names) response = await self._configuration_stub.ReadSpecificMarkers(request, timeout=30.0) # Convert protobuf response to dict nodes = [] for node in response.extracted_nodes: nodes.append({ "start_offset": node.start_offset, "end_offset": node.end_offset, "node_type": node.node_type, "name": node.name, "value": node.value, "value_type": node.value_type }) return { "success": response.success, "error_message": response.error_message if response.error_message else None, "file_size": response.file_size, "requested_markers": list(response.requested_markers), "extracted_nodes": nodes, "markers_found": response.markers_found } except grpc.RpcError as e: logger.error("sdk_bridge_read_specific_markers_failed", error=str(e)) raise async def create_action_mapping(self, mapping_data: dict) -> dict: """ Create a new action mapping Args: mapping_data: Dict with name, input_actions, output_actions Returns: Dict with success status and created mapping """ try: logger.info("sdk_bridge_create_action_mapping", name=mapping_data.get("name")) # Build protobuf request mapping_input = configuration_pb2.ActionMappingInput( name=mapping_data.get("name", "") ) # Add output actions for action_data in mapping_data.get("output_actions", []): action_def = configuration_pb2.ActionDefinition(action=action_data["action"]) # Add parameters for param_name, param_value in action_data.get("parameters", {}).items(): action_def.parameters.add(name=param_name, value=str(param_value)) mapping_input.output_actions.append(action_def) request = configuration_pb2.CreateActionMappingRequest(mapping=mapping_input) response = await self._configuration_stub.CreateActionMapping(request, timeout=60.0) # Convert response result = { "success": response.success, "error_message": response.error_message if response.error_message else None, "message": response.message } if response.mapping: result["mapping"] = { "id": len([]), # ID will be assigned by the system "name": response.mapping.name, "offset": response.mapping.start_offset, "output_actions": [] } for action_def in response.mapping.output_actions: result["mapping"]["output_actions"].append({ "action": action_def.action, "parameters": {p.name: p.value for p in action_def.parameters} }) return result except grpc.RpcError as e: logger.error("sdk_bridge_create_action_mapping_failed", error=str(e)) raise async def update_action_mapping(self, mapping_id: int, mapping_data: dict) -> dict: """ Update an existing action mapping Args: mapping_id: 1-based ID of mapping to update mapping_data: Dict with updated fields (name, input_actions, output_actions) Returns: Dict with success status and updated mapping """ try: logger.info("sdk_bridge_update_action_mapping", mapping_id=mapping_id) # Build protobuf request mapping_input = configuration_pb2.ActionMappingInput() if "name" in mapping_data: mapping_input.name = mapping_data["name"] # Add output actions if provided if "output_actions" in mapping_data: for action_data in mapping_data["output_actions"]: action_def = configuration_pb2.ActionDefinition(action=action_data["action"]) # Add parameters for param_name, param_value in action_data.get("parameters", {}).items(): action_def.parameters.add(name=param_name, value=str(param_value)) mapping_input.output_actions.append(action_def) request = configuration_pb2.UpdateActionMappingRequest( mapping_id=mapping_id, mapping=mapping_input ) response = await self._configuration_stub.UpdateActionMapping(request, timeout=60.0) # Convert response result = { "success": response.success, "error_message": response.error_message if response.error_message else None, "message": response.message } if response.mapping: result["mapping"] = { "id": mapping_id, "name": response.mapping.name, "offset": response.mapping.start_offset, "output_actions": [] } for action_def in response.mapping.output_actions: result["mapping"]["output_actions"].append({ "action": action_def.action, "parameters": {p.name: p.value for p in action_def.parameters} }) return result except grpc.RpcError as e: logger.error("sdk_bridge_update_action_mapping_failed", error=str(e)) raise async def delete_action_mapping(self, mapping_id: int) -> dict: """ Delete an action mapping by ID Args: mapping_id: 1-based ID of mapping to delete Returns: Dict with success status and message """ try: logger.info("sdk_bridge_delete_action_mapping", mapping_id=mapping_id) request = configuration_pb2.DeleteActionMappingRequest(mapping_id=mapping_id) response = await self._configuration_stub.DeleteActionMapping(request, timeout=60.0) return { "success": response.success, "error_message": response.error_message if response.error_message else None, "message": response.message } except grpc.RpcError as e: logger.error("sdk_bridge_delete_action_mapping_failed", error=str(e)) raise async def read_configuration_tree(self) -> dict: """ Read configuration as hierarchical folder tree (RECOMMENDED) Returns: Dict with tree structure """ try: logger.info("sdk_bridge_read_configuration_tree") request = configuration_pb2.ReadConfigurationTreeRequest() response = await self._configuration_stub.ReadConfigurationTree(request, timeout=30.0) if not response.success: return { "success": False, "error_message": response.error_message } # Convert protobuf TreeNode to dict def convert_tree_node(node): result = { "type": node.type, "name": node.name } # Add value based on type if node.type == "string": result["value"] = node.string_value elif node.type in ("bool", "byte", "int16", "int32", "int64"): result["value"] = node.int_value # Add children recursively if node.type == "folder" and len(node.children) > 0: result["children"] = [convert_tree_node(child) for child in node.children] return result tree_dict = convert_tree_node(response.root) if response.root else None return { "success": True, "tree": tree_dict, "total_nodes": response.total_nodes } except grpc.RpcError as e: logger.error("sdk_bridge_read_configuration_tree_failed", error=str(e)) raise async def write_configuration_tree(self, tree: dict) -> dict: """ Write modified configuration tree back to GeViServer Args: tree: Modified tree structure (dict) Returns: Dict with success status and write statistics """ try: import json logger.info("sdk_bridge_write_configuration_tree") # Convert tree to JSON string json_data = json.dumps(tree, indent=2) # Use import_configuration to write the tree result = await self.import_configuration(json_data) return result except Exception as e: logger.error("sdk_bridge_write_configuration_tree_failed", error=str(e)) raise # Global SDK Bridge client instance sdk_bridge_client = SDKBridgeClient() # Convenience functions async def init_sdk_bridge(): """Initialize SDK Bridge connection (call on startup)""" await sdk_bridge_client.connect() async def close_sdk_bridge(): """Close SDK Bridge connection (call on shutdown)""" await sdk_bridge_client.close()