#!/usr/bin/env python3
"""عدسة النقوش SAM 2 — نسخة مطورة"""
import cv2
import numpy as np
import base64, json, pickle, os, torch
from http.server import HTTPServer, BaseHTTPRequestHandler
SAM2_DIR = '/var/www/html/sam2'
MODEL_PATH = os.path.join(SAM2_DIR, 'sam2_hiera_large.pt')
CFG_PATH = 'configs/sam2/sam2_hiera_l.yaml'
# Load model once at startup
print("⏳ جاري تحميل SAM 2...")
from sam2.build_sam import build_sam2
from sam2.sam2_image_predictor import SAM2ImagePredictor
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
sam2_model = build_sam2(CFG_PATH, MODEL_PATH, device=device)
predictor = SAM2ImagePredictor(sam2_model, device=device)
print(f"✅ SAM 2 جاهز على {device}")
HTML_PAGE = """
عدسة النقوش SAM 2
⏳ يرجى الانتظار...
⚙️ جاري المعالجة...
"""
def parse_multipart(parts, key):
for p in parts:
if f'name="{key}"'.encode() in p:
i = p.find(b'\r\n\r\n')
if i != -1:
data = p[i+4:]
if data.endswith(b'\r\n'): data = data[:-2]
if data.endswith(b'--'): data = data[:-2]
return data
return None
def parse_text(parts, key, default=''):
v = parse_multipart(parts, key)
if v is None: return default
try: return v.decode().strip()
except: return default
overlay = None
overlay_step = 0
def process_init(img_bytes):
global overlay, overlay_step
arr = np.frombuffer(img_bytes, np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if img is None: return None, "خطأ في قراءة الصورة"
h, w = img.shape[:2]
if w > 800:
ratio = 800/w; img = cv2.resize(img, (800, int(h*ratio)))
_, ob = cv2.imencode('.jpg', img, [cv2.IMWRITE_JPEG_QUALITY, 85])
orig = base64.b64encode(ob).decode()
predictor.set_image(cv2.cvtColor(img, cv2.COLOR_BGR2RGB))
overlay = np.zeros((img.shape[0], img.shape[1], 4), dtype=np.uint8)
overlay_step = 0
return {'original': orig}, None
def process_click(img_bytes, cx, cy, step):
global overlay, overlay_step
arr = np.frombuffer(img_bytes, np.uint8)
img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
if img is None: return None, "خطأ"
h, w = img.shape[:2]
if w > 800:
ratio = 800/w; img = cv2.resize(img, (800, int(h*ratio)))
# Get SAM 2 mask
masks, scores, logits = predictor.predict(
point_coords=np.array([[int(cx/100*img.shape[1]), int(cy/100*img.shape[0])]]),
point_labels=np.array([1]),
multimask_output=False,
)
mask = masks[0]; mu8 = (mask*255).astype(np.uint8)
contours, _ = cv2.findContours(mu8, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
cv2.drawContours(overlay, contours, -1, (255, 255, 255, 255), 2)
overlay_step += 1
# Save history
cv2.imwrite(f'/tmp/sam2_hist_{overlay_step}.png', overlay)
_, rb = cv2.imencode('.png', overlay)
return {'overlay': base64.b64encode(rb).decode()}, None
def process_undo():
global overlay, overlay_step
if overlay_step <= 0: return None, "لا يوجد"
overlay_step -= 1
if overlay_step == 0:
overlay = np.zeros((1,1,4), dtype=np.uint8)
_, rb = cv2.imencode('.png', overlay)
return {'overlay': base64.b64encode(rb).decode()}, None
overlay = cv2.imread(f'/tmp/sam2_hist_{overlay_step}.png', cv2.IMREAD_UNCHANGED)
_, rb = cv2.imencode('.png', overlay)
return {'overlay': base64.b64encode(rb).decode()}, None
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'text/html; charset=utf-8')
self.end_headers()
self.wfile.write(HTML_PAGE.encode('utf-8'))
def do_POST(self):
ct = self.headers.get('Content-Type', '')
if not ct.startswith('multipart/form-data'): return self._json({'error':'نوع غير مدعوم'})
cl = int(self.headers.get('Content-Length', 0))
parts = self.rfile.read(cl).split(b'--' + ct.split('=')[1].encode())
p = self.path
if p == '/init':
ib = parse_multipart(parts, 'image')
if not ib: return self._json({'error':'لا توجد صورة'})
r, e = process_init(ib)
self._json({'error': e} if e else r)
elif p == '/click':
ib = parse_multipart(parts, 'image')
cx = float(parse_text(parts, 'x', '50'))
cy = float(parse_text(parts, 'y', '50'))
st = int(parse_text(parts, 'step', '0'))
if not ib: return self._json({'error':'لا توجد صورة'})
r, e = process_click(ib, cx, cy, st)
self._json({'error': e} if e else r)
elif p == '/undo':
r, e = process_undo()
self._json({'error': e} if e else r)
def _json(self, d):
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.end_headers()
self.wfile.write(json.dumps(d).encode('utf-8'))
if __name__ == '__main__':
HTTPServer(('0.0.0.0', 8086), Handler).serve_forever()