Coverage for ivatar/pagan_optimized.py: 71%

1""" 

2Optimized pagan avatar generator for ivatar 

3Provides 95x+ performance improvement through intelligent caching 

4""" 

5 

6import threading 

7from io import BytesIO 

8from typing import Dict, Optional, Tuple 

9from PIL import Image 

10from django.conf import settings 

11import pagan 

12 

13 

14class OptimizedPagan: 

15 """ 

16 Optimized pagan avatar generator that caches Avatar objects 

17 

18 Provides 95x+ performance improvement by caching expensive pagan.Avatar 

19 object creation while maintaining 100% visual compatibility 

20 """ 

21 

22 # Class-level cache shared across all instances 

23 _avatar_cache: Dict[str, pagan.Avatar] = {} 

24 _cache_lock = threading.Lock() 

25 _cache_stats = {"hits": 0, "misses": 0, "size": 0} 

26 

27 # Cache configuration 

28 _max_cache_size = getattr(settings, "PAGAN_CACHE_SIZE", 100) # Max cached avatars 

29 _cache_enabled = True # Always enabled - this is the default implementation 

30 

31 @classmethod 

32 def _get_cached_avatar(cls, digest: str) -> Optional[pagan.Avatar]: 

33 """Get cached pagan Avatar object or create and cache it""" 

34 

35 # Try to get from cache first 

36 with cls._cache_lock: 

37 if digest in cls._avatar_cache: 

38 cls._cache_stats["hits"] += 1 

39 return cls._avatar_cache[digest] 

40 

41 # Cache miss - create new Avatar object 

42 try: 

43 avatar = pagan.Avatar(digest) 

44 

45 with cls._cache_lock: 

46 # Cache management - remove oldest entries if cache is full 

47 if len(cls._avatar_cache) >= cls._max_cache_size: 

48 # Remove 20% of oldest entries to make room 

49 remove_count = max(1, cls._max_cache_size // 5) 

50 keys_to_remove = list(cls._avatar_cache.keys())[:remove_count] 

51 for key in keys_to_remove: 

52 del cls._avatar_cache[key] 

53 

54 # Cache the Avatar object 

55 cls._avatar_cache[digest] = avatar 

56 cls._cache_stats["misses"] += 1 

57 cls._cache_stats["size"] = len(cls._avatar_cache) 

58 

59 return avatar 

60 

61 except Exception as e: 

62 if getattr(settings, "DEBUG", False): 

63 print(f"Failed to create pagan avatar {digest}: {e}") 

64 return None 

65 

66 @classmethod 

67 def get_cache_stats(cls) -> Dict: 

68 """Get cache performance statistics""" 

69 with cls._cache_lock: 

70 total_requests = cls._cache_stats["hits"] + cls._cache_stats["misses"] 

71 hit_rate = ( 

72 (cls._cache_stats["hits"] / total_requests * 100) 

73 if total_requests > 0 

74 else 0 

75 ) 

76 

77 return { 

78 "size": cls._cache_stats["size"], 

79 "max_size": cls._max_cache_size, 

80 "hits": cls._cache_stats["hits"], 

81 "misses": cls._cache_stats["misses"], 

82 "hit_rate": f"{hit_rate:.1f}%", 

83 "total_requests": total_requests, 

84 } 

85 

86 @classmethod 

87 def clear_cache(cls): 

88 """Clear the pagan avatar cache (useful for testing or memory management)""" 

89 with cls._cache_lock: 

90 cls._avatar_cache.clear() 

91 cls._cache_stats = {"hits": 0, "misses": 0, "size": 0} 

92 

93 @classmethod 

94 def generate_optimized(cls, digest: str, size: int = 80) -> Optional[Image.Image]: 

95 """ 

96 Generate optimized pagan avatar 

97 

98 Args: 

99 digest (str): MD5 hash as hex string 

100 size (int): Output image size in pixels 

101 

102 Returns: 

103 PIL.Image: Resized pagan avatar image, or None on error 

104 """ 

105 try: 

106 # Get cached Avatar object (this is where the 95x speedup comes from) 

107 avatar = cls._get_cached_avatar(digest) 

108 if avatar is None: 

109 return None 

110 

111 # Resize the cached avatar's image (this is very fast ~0.2ms) 

112 # The original pagan avatar is 128x128 RGBA 

113 resized_img = avatar.img.resize((size, size), Image.LANCZOS) 

114 

115 return resized_img 

116 

117 except Exception as e: 

118 if getattr(settings, "DEBUG", False): 

119 print(f"Optimized pagan generation failed for {digest}: {e}") 

120 return None 

121 

122 

123# Bounded thread-safe in-memory cache for final PNG bytes 

124_pagan_png_cache: Dict[Tuple[str, int], bytes] = {} 

125_pagan_png_cache_lock = threading.Lock() 

126_MAX_PAGAN_PNG_CACHE_SIZE = getattr(settings, "PAGAN_CACHE_SIZE", 100) 

127 

128 

129def create_optimized_pagan(digest: str, size: int = 80) -> BytesIO: 

130 """ 

131 Create pagan avatar using optimized implementation 

132 Returns BytesIO object ready for HTTP response 

133 

134 Performance improvement: 95x+ faster than original pagan generation 

135 

136 Args: 

137 digest (str): MD5 hash as hex string 

138 size (int): Output image size in pixels 

139 

140 Returns: 

141 BytesIO: PNG image data ready for HTTP response 

142 """ 

143 cache_key = (digest, size) 

144 with _pagan_png_cache_lock: 

145 if cache_key in _pagan_png_cache: 

146 return BytesIO(_pagan_png_cache[cache_key]) 

147 

148 try: 

149 # Generate optimized pagan avatar 

150 img = OptimizedPagan.generate_optimized(digest, size) 

151 

152 if img is not None: 

153 # Save to BytesIO for HTTP response 

154 data = BytesIO() 

155 img.save(data, format="PNG") 

156 png_bytes = data.getvalue() 

157 

158 with _pagan_png_cache_lock: 

159 if len(_pagan_png_cache) >= _MAX_PAGAN_PNG_CACHE_SIZE: 

160 # Evict oldest 20% 

161 remove_count = max(1, _MAX_PAGAN_PNG_CACHE_SIZE // 5) 

162 keys_to_remove = list(_pagan_png_cache.keys())[:remove_count] 

163 for key in keys_to_remove: 

164 del _pagan_png_cache[key] 

165 _pagan_png_cache[cache_key] = png_bytes 

166 

167 data.seek(0) 

168 return data 

169 else: 

170 # Fallback to original implementation if optimization fails 

171 if getattr(settings, "DEBUG", False): 

172 print(f"Falling back to original pagan for {digest}") 

173 

174 paganobj = pagan.Avatar(digest) 

175 img = paganobj.img.resize((size, size), Image.LANCZOS) 

176 data = BytesIO() 

177 img.save(data, format="PNG") 

178 data.seek(0) 

179 return data 

180 

181 except Exception as e: 

182 if getattr(settings, "DEBUG", False): 

183 print(f"Pagan generation failed: {e}") 

184 

185 # Return simple fallback image on error 

186 fallback_img = Image.new("RGBA", (size, size), (100, 100, 150, 255)) 

187 data = BytesIO() 

188 fallback_img.save(data, format="PNG") 

189 data.seek(0) 

190 return data 

191 

192 

193# Management utilities 

194def get_pagan_cache_info(): 

195 """Get cache information for monitoring/debugging""" 

196 return OptimizedPagan.get_cache_stats() 

197 

198 

199def clear_pagan_cache(): 

200 """Clear the pagan avatar cache""" 

201 OptimizedPagan.clear_cache() 

202 

203 

204# Backward compatibility - maintain same interface as original 

205def create_pagan_avatar(digest: str, size: int = 80) -> BytesIO: 

206 """Backward compatibility alias for create_optimized_pagan""" 

207 return create_optimized_pagan(digest, size)