Coverage for core / resonance_identifier.py: 100.0%

1""" 

2ResonanceIdentifier — Thin proxy for biometric user identification. 

3 

4All biometric ML (face embedding extraction, voice embedding extraction, 

5cosine similarity matching) lives in the HevolveAI sibling repo. 

6 

7HARTOS dispatches raw biometric data (frame bytes, audio bytes) to 

8HevolveAI via WorldModelBridge and receives identification results back. 

9When HevolveAI is unavailable, all operations gracefully return None/False 

10and the system falls back to channel identity (user_id from API/WebSocket). 

11 

12Priority: 

13 1. Channel identity (user_id from API/WebSocket) — always available 

14 2. Face match (dispatched to HevolveAI) 

15 3. Voice match (dispatched to HevolveAI) 

16""" 

17 

18import base64 

19import logging 

20from typing import Optional, Tuple 

21 

22logger = logging.getLogger(__name__) 

23 

24 

25class ResonanceIdentifier: 

26 """Dispatch biometric operations to HevolveAI. No local ML.""" 

27 

28 def identify_by_face(self, frame_bytes: bytes) -> Optional[Tuple[str, float]]: 

29 """Request face identification from HevolveAI. 

30 

31 Returns (user_id, confidence) or None if HevolveAI unavailable. 

32 """ 

33 return self._dispatch_identification('face', frame_bytes) 

34 

35 def identify_by_voice(self, audio_bytes: bytes) -> Optional[Tuple[str, float]]: 

36 """Request voice identification from HevolveAI. 

37 

38 Returns (user_id, confidence) or None if HevolveAI unavailable. 

39 """ 

40 return self._dispatch_identification('voice', audio_bytes) 

41 

42 def enroll_face(self, user_id: str, frame_bytes: bytes, 

43 base_dir: str = None) -> bool: 

44 """Dispatch face enrollment to HevolveAI.""" 

45 return self._dispatch_enrollment('face', user_id, frame_bytes) 

46 

47 def enroll_voice(self, user_id: str, audio_bytes: bytes, 

48 base_dir: str = None) -> bool: 

49 """Dispatch voice enrollment to HevolveAI.""" 

50 return self._dispatch_enrollment('voice', user_id, audio_bytes) 

51 

52 def _dispatch_identification(self, modality: str, 

53 raw_bytes: bytes) -> Optional[Tuple[str, float]]: 

54 """Send biometric data to HevolveAI for identification.""" 

55 try: 

56 from integrations.agent_engine.world_model_bridge import get_world_model_bridge 

57 bridge = get_world_model_bridge() 

58 bridge.record_interaction( 

59 user_id='biometric_probe', 

60 prompt_id='biometric_identification', 

61 prompt=base64.b64encode(raw_bytes[:4096]).decode('ascii'), 

62 response='', 

63 model_id=f'biometric_{modality}_identify', 

64 ) 

65 # HevolveAI processes asynchronously; result flows back 

66 # via apply_hevolveai_corrections() in resonance_tuner 

67 return None 

68 except ImportError: 

69 return None 

70 except Exception as e: 

71 logger.debug(f"Biometric identification dispatch failed: {e}") 

72 return None 

73 

74 def _dispatch_enrollment(self, modality: str, user_id: str, 

75 raw_bytes: bytes) -> bool: 

76 """Send biometric data to HevolveAI for enrollment.""" 

77 try: 

78 from integrations.agent_engine.world_model_bridge import get_world_model_bridge 

79 bridge = get_world_model_bridge() 

80 bridge.record_interaction( 

81 user_id=user_id, 

82 prompt_id='biometric_enrollment', 

83 prompt=base64.b64encode(raw_bytes[:4096]).decode('ascii'), 

84 response='', 

85 model_id=f'biometric_{modality}_enroll', 

86 ) 

87 return True 

88 except ImportError: 

89 return False 

90 except Exception as e: 

91 logger.debug(f"Biometric enrollment dispatch failed: {e}") 

92 return False