مقدمه — از دوربین منفعل تا عامل تصمیم‌گیرنده مستقل

دوربین مداربسته دیگر فقط ضبط نیست. در معماری‌های نظارتی نسل جدید، دوربین نقطه‌ی ورود یک Autonomous Surveillance Agent است؛ سیستمی که می‌بیند، درک می‌کند، استدلال می‌کند و اقدام می‌کند — بدون دخالت انسان در حلقه.

این تحول از یک پارادایم ساده نشأت می‌گیرد: ترکیب استریم ویدیویی خام با مدل‌های Vision-Language مثل GPT-4o، Claude 3.5 Sonnet، یا YOLOv10 که قادر به فهم معنایی صحنه هستند، نه صرفاً تشخیص حرکت.

چرا داهوا؟

دوربین‌های IPC داهوا به دلایل زیر بهترین کاندید برای این ادغام هستند:

• پشتیبانی کامل از ONVIF Profile S, G, T — استاندارد بین‌المللی برای دوربین‌های IP که اتصال vendor-agnostic را ممکن می‌کند
• پشتیبانی از Sub-stream با رزولوشن مستقل — حیاتی برای inference در لایه Edge
• PTZ SDK باز — امکان کنترل SpeedDome از طریق ONVIF مستقیماً در پK باز — امکان کنترل SpeedDome از طریق ONVIF مستقیماً در پ.265، مناسب برای OpenCV و FFmpeg

تهران داهوا به عنوان تأمین‌کننده رسمی دوربین‌های شبکه داهوا در ایران، طیف کاملی از دوربین‌های IPC سری Lite، Pro و Ultra را با گارانتی معتبر ارائه می‌دهد — دوربین‌هایی که به طور خاص برای پیاده‌سازی این معماری آزمایش شده‌اند.

پیش‌نیازهای سخت‌افزاری و نر3 id="hardware-prereqs">سخت‌افزار دوربینخت‌افزار دوربین

برای پیاده‌سازی AI Agent روی دوربین داهوا، دوربین باید این مشخصات را داشته باشد:

سری‌های IPC-HDW، IPC-HFW، و SD داهوا همگی این الزامات را برآورده می‌کنند.

زیرساخت شبکه

• سوئیچ PoE با استاندارد IEEE 802.3af/at (مثل سوئیچ‌های داهوا PFS سری 3000)
• آدرس‌دهی ثابت برای دوربین‌ها در زیرشبکه مجزا (مثلاً 192.168.1.0/24)
• اطمینان از اینکه MTU شبکه روی 1500 تنظیم است
• پهنای باند رزرو‌شده: حداقل 4 Mbps به ازای"># Python 3.10+ الزامی است python --version # Python 3.10.x یا بالاتر # نصب وابستگی‌های اصلی pip install onvif-zeep==0.2.12 pip install opencv-python==4.9.0.80 pip install numpy==1.26.4 pip install langchain==0.2.16 pip install langgraph==0.1.19 pip install ultralytics==8.2.87 # YOLOv10 pip install httpx==0.27.0 pip install Pillow==10.4.0

  توجه: نسخه‌های دقیق بالا برای compatibility آزمایش‌شده‌اند. از open range مثل >= در production استفاده نکنید.

گام 1 — فعال‌سازی و پیکربندی ONVIF در دوربین IP داهوا

ورود به وب پنل و فعال‌سازی ONVIF

1. مرورگر را باز کنید و به آدرس IP دوربین بروید (پیش‌فرض: 192.168.1.108)
2. با اعتبارسنجی پیش‌فرض وارد شوید: admin / admin — بلافاصله رمز عبور را تغییر دهید
3. مسیر: Setup → Network → Advanced → Integration Protocol
4. گزینه Enable ONVIF را فعال کنید
5. یک کاربر ONVIF اختصاصی بسازید:
   • Username: onvif_agent
   • Password: حاربر ONVIF اختصاصی بسازید:
     • Username: onvif_agent
     • Password: حr

تفاوت RTSP و ONVIF — درک معماری پروتکل

این تمایز در طراحی AI Agent حیاتی است:

RTSP (Real-Time Streaming Protocol): فقط برای انتقال فریم‌های ویدیویی خام، کانال یک‌طرفه (دوربین → سرور).

• آدرس Main Stream داهوا: rtsp://user:pass@192.168.1.108:554/cam/realmonitor?channel=1&subtype=0
• آدرس Sub-stream داهوا: rtsp://user:pass@192.168.1.108:554/cam/realmonitor?channel=1&subtype=1

ONVIF (Open Network Video Interface Forum): پروتکل مبتنی بر SOAP/WS برای کنترل و متادیتا، کانال دو‌طرفه (کنترل PTZ، دریافت رویدادها، کوئری مشخصات دستگاه). پورت پیش‌فرض داهوا: 80 (HTTP) یا 443 (HTTPS).

در معماری AI Agent، RTSP داده می‌دهد و ONVIF فرمان می‌گیرد.

گام 2 — اتصال به استریم RTSP و کنترل ONVIF با پایتون

کد اتصال به استریم RTSP دوربین داهوا

"""
dahua_rtsp_reader.py
اتصال پایدار به استریم RTSP دوربین مداربسته داهوا با OpenCV
"""
import cv2
import time
import logging
from dataclasses import dataclass
from typing import Generator, Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class DahuaCamera:
    """پارامترهای اتصال به دوربین داهوا"""
    host: str
    port: int = 554
    username: str = ""
    password: str = ""
    channel: int = 1
    use_substream: bool = True  # True = Sub-stream، کمتر مصرف پردازشی

    @property
    def rtsp_url(self) -> str:
        subtype = 1 if self.use_substream else 0
        return (
            f"rtsp://{self.username}:{self.password}@"
            f"{self.host}:{self.port}/cam/realmonitor"
            f"?channel={self.channel}&subtype={subtype}"
        )


class RTSPStreamReader:
    """
    مدیریت استریم RTSP با قابلیت reconnect خودکار.
    از Sub-stream برای کاهش latency استفاده می‌کند.
    """

    def __init__(self, camera: DahuaCamera, max_retries: int = 5):
        self.camera = camera
        self.max_retries = max_retries
        self._cap: Optional[cv2.VideoCapture] = None

    def _build_capture(self) -> cv2.VideoCapture:
        """ساخت VideoCapture با تنظیمات بهینه برای latency پای"""
        url = self.camera.rtsp_ = self.camera.rtsp_url
        cap = cv2.VideoCapture(url, cv2.CAP_FFMPEG)
        cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
        cap.set(cv2.CAP_PROP_FPS, 15)self.camera.host} برError(f"اتصال به {self.camera.host} برقرار نشد.")

        logger.info(
            "اتصال RTSP برقرار شد: %s | رزولوشن: %dx%d",
            self.camera.host,
            int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
            int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
        )
        return cap

    def frame_generator(self) -> Generator:
        """
        ژنراتور فریم با مدیریت خودکار reconnect.
        هر بار yield می‌کند یک numpy array به شکل (H, W, 3) BGR.
        """
        retry_count = 0

        while retry_count < self.max_retries:
            try:
                self._cap = self._build_capture()
                retry_count = 0

                while True:
                    ret, frame = self._cap.read()
                    if not ret:
                        logger.warning("فریم دریافت نشد — تلاش برای reconnect...")
                        break
                    yield frame

            except ConnectionError as exc:
                retry_count += 1
                wait = min(2 ** retry_count, 30)
                logger.error("%s | تلاش %d/%d | انتظار %ds", exc, retry_count, self.max_retries, wait)
                time.sleep(wait)

            finally:
                if self._cap:
                    self._cap.release()

        raise RuntimeError("حداکثر تعداد تلاش برای اتصال مصرف شد.")


if __name__ == "__main__":
    cam = DahuaCamera(
        host="192.168.1.108",
        username="onvif_agent",
        password="YourStrongPass!23",
        use_substream=True,
    )

    reader = RTSPStreamReader(cam)
    for i, frame in enumerate(reader.frame_generator()):
        cv2.imshow("Dahua Live", frame)
        if cv2.waitKey(1) & 0xFF == ord("q"):
            break
        if i >= 300:
            break

    cv2.destroyAllWindows()

کد کنترل ONVIF — واکشی مشخصات و حرکت PTZ

"""
dahua_onvif_controller.py
کنترل دوربین داهوا از طریق پروتکل ONVIF — مشخصات دستگاه و PTZ Continuous Move
"""
from onvif import ONVIFCamera
import logging
import time

logger = logging.getLogger(__name__)


class DahuaONVIFController:
    """
    رابط ONVIF برای دوربین‌های داهوا.
    پشتیبانی از: واکشی مشخصات، کنترل PTZ، دریافت پروفایل‌های استریم.
    """

    def __init__(self, host: str, port: int, username: str, password: str):
        self.host = host
        self._ port, username, password)
        self._ptz = None
        self._media =tz = None
        self._media = None
        self._profile_token: str = ""

    def connect(self) -> None:
        self._media = self._cam.create_media_service()
        self._ptz = self._cam.create_ptz_service()
        profiles = self._media.GetProfiles()
        if not profiles:
            raise RuntimeError("پروفایل رسانه‌ای یافت نشد.")
        self._profile_token = profiles[0].token
        logger.info("ONVIF متصل شد. Profile token: %s", self._profile_token)

    def get_device_info(self) -> dict:
        device_service = self._cam.create_devicemgmt_service()
        info = device_service.GetDeviceInformation()
        return {
            "manufacturer": info.Manufacturer,
            "model": info.Model,
            "firmware": info.FirmwareVersion,
            "serial": info.SerialNumber,
            "hardware_id": info.HardwareId,
        }

    def ptz_continuous_move(
        self,
        pan: float = 0.0,
        tilt: float = 0.0,
        zoom: float = 0.0,
        duration_seconds: float = 1.0,
    ) -> None:
        """
        حرکت پیوسته دوربین PTZ داهوا.

        Args:
            pan: سرعت چرخش افقی (-1.0 تا 1.0)
            tilt: سرعت چرخش عمودی (-1.0 تا 1.0)
            zoom: سرعت زوم (-1.0 تا 1.0)
            duration_seconds: مدت زمان حرکت
        """
        if not self._ptz:
            raise RuntimeError("ابتدا متد connect() را فراخوانی کنید.")

        request = self._ptz.create_type("ContinuousMove")
        request.ProfileToken = self._profile_token
        request.Velocity = {
            "PanTilt": {"x": pan, "y": tilt},
            "Zoom": {"x": zoom},
        }

        self._ptz.ContinuousMove(request)
        logger.info("PTZ در حال حرکت: pan=%.2f, tilt=%.2f, zoom=%.2f", pan, tilt, zoom)
        time.sleep(duration_seconds)
        self._ptz.Stop({"ProfileToken": self._profile_token})
        logger.info("PTZ متوقف شد.")

    def get_rtsp_url(self, use_substream: bool = True) -> str:
        stream_setup = {
            "Stream": "RTP-Unicast",
            "Transport": {"Protocol": "RTSP"},
        }
        uri_response = self._media.GetStreamUri(
            {"StreamSetup": stream_setup, "ProfileToken": self._profile_token}
        )
        return uri_response.Uri


if __name__ == "__main__":
    controller = DahuaONVIFController(
        host="192.168.1.108",
        port=80,
        username="onvif_agent",
        password="YourStrongPass!23",
    )
    controller.connect()
    info = controller.get_device_info()
    print(f"مدل دوربین: {info['model']} | فریم‌ور: {info['firmware']}")
    controller.ptz_continuous_move(pan=0.3, tilt=0.0, duration_seconds=2.0)

گام 3 — معماری کامل AI Agent هوشمند برای دوربین مداربسته

این معماری سه‌لایه‌ای استاندارد در حوزه Agentic AI Vision Systems است:

┌─────────────────────────────────────────────────────────┐
│                    SURVEILLANCE AI AGENT                │
├─────────────────┬───────────────────┬───────────────────┤
│  PERCEPTION     │    REASONING      │     ACTION        │
│  Layer 1        │    Layer 2        │     Layer 3       │
│                 │                   │                   │
│  RTSP Stream    │   LLM / VLM       │  PTZ Control      │
│  OpenCV Decode  │   (GPT-4o /       │  Webhook Call     │
│  YOLOv10 Infer  │   YOLOv10+NLP)    │  Alarm Trigger    │
│  Frame Buffer   │   Scene Context   │  Notification     │
└─────────────────┴───────────────────┴───────────────────┘

لایه ادراک (Perception) — تشخیص اشیاء با YOLOv10

"""
perception_layer.py
استخراج فریم و تشخیص اشیاء با YOLOv10 — لایه ادراک AI Agent
"""
import cv2
import numpy as np
from ultralytics import YOLO
from dataclasses import dataclass, field
from typing import List, Optional
import base64
import io
from PIL import Image


@dataclass
class Detection:
    label: str
    confidence: float
    bbox: tuple  # (x1, y1, x2, y2) نرمال‌شده
    is_person: bool = False


@dataclass
class PerceptionResult:
    frame_id: int
    detections: List[Detection] = field(default_factory=list)
    has_suspicious_activity: bool = False
    encoded_frame: Optional[str] = None  # Base64 برای ارسال به VLM


class PerceptionLayer:
    """
    لایه ادراک: دریافت فریم، اجرای YOLOv10، فیلتر رویدادهای مشکوک.
    از YOLOv10n (nano) برای تعادل سرعت/دقت در Edge استفاده می‌شود.
    """

    SUSPICIOUS_LABELS = {"person", "backpack", "suitcase", "knife"}

    def __init__(self, model_path: str = "yolov10n.pt", confidence_threshold: float = 0.45):
        self.model = YOLO(model_path)
        self.conf_threshold = confidence_threshold
        self._frame_counter = 0

    def process_frame(self, frame: np.ndarray) -> PerceptionResult:
        self._frame_counter += 1
        h, w = frame.shape[:2]
        results = self.model(frame, conf=self.conf_threshold, verbose=False)
        detections: List[Detection] = []

        for result in results:
            for box in result.boxes:
                label = self.model.names[int(box.cls[0])]
                confidence = float(box.conf[0])
                x1, y1, x2, y2 = box.xyxy[0].tolist()
                norm_bbox = (x1 / w, y1 / h, x2 / w, y2 / h)
                detections.append(
                    Detection(label=label, confidence=confidence,
                              bbox=norm_bbox, is_person=(label == "person"))
                )

        detected_labels = {d.label for d in detections}
        has_suspicious = bool(detected_labels & self.SUSPICIOUS_LABELS)
        result_obj = PerceptionResult(
            frame_id=self._frame_counter,
            detections=detections,
            has_suspicious_activity=has_suspicious,
        )

        if has_suspicious:
            result_obj.encoded_frame = self._encode_frame(frame)

        return result_obj

    @staticmethod
    def _encode_frame(frame: np.ndarray) -> str:
        rgb_frame = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
        pil_image = Image.fromarray(rgb_frame)
        buffer = io.BytesIO()
        pil_image.save(buffer, format="JPEG", quality=85)
        return base64.b64encode(buffer.getvalue()).decode("utf-8")

لایه استدلال (Reasoning) — تحلیل صحنه با VLM

"""
reasoning_layer.py
لایه استدلال: ارسال فریم و متادیتا به VLM جهت درک معنایی صحنه
"""
import httpx
import json
import logging
from dataclasses import dataclass
from typing import Optional
from perception_layer import PerceptionResult

logger = logging.getLogger(__name__)


@dataclass
class SceneAnalysis:
    threat_level: int  # 0=بدون تهدید، 1=مشکوک، 2=خطرناک
    description: str
    recommended_action: str
    requires_ptz_track: bool = False
    target_bbox: Optional[tuple] = None


class ReasoningLayer:
    SYSTEM_PROMPT = """
    تو یک سیستم تحلیل نظارت تصویری هوشمند هستی.
    صحنه تصویری و لیست اشیاء تشخیص‌داده‌شده را دریافت می‌کنی.
    باید پاسخ JSON با این فیلدها برگردانی:
    - threat_level: عدد صحیح 0 تا 2
    - description: توضیح فارسی صحنه در یک جمله
    - recommended_action: یکی از: "monitor", "track_ptz", "trigger_alarm", "notify_operator"
    - requires_ptz_track: true/false
    - target_description: توضیح هدف اصلی برای دنبال کردن (اگر نیاز باشد)
    """.strip()

    def __init__(self, api_key: str, model: str = "gpt-4o",
                 endpoint: str = "https://api.openai.com/v1"):
        self.api_key = api_key
        self.model = model
        self.endpoint = endpoint

    def analyze(self, perception_result: PerceptionResult) -> Optional[SceneAnalysis]:
        if not perception_result.has_suspicious_activity:
            return None

        detection_summary = self._build_detection_summary(perception_result)
        messages = [
            {"role": "system", "content": self.SYSTEM_PROMPT},
            {"role": "user", "content": self._build_user_message(
                detection_summary, perception_result.encoded_frame)},
        ]

        try:
            response = self._call_vlm(messages)
            return self._parse_response(response, perception_result)
        except Exception as exc:
            logger.error("خطا در تحلیل VLM: %s", exc)
            return None

    def _build_detection_summary(self, result: PerceptionResult) -> str:
        lines = [f"فریم شماره {result.frame_id}:"]
        for det in result.detections:
            lines.append(f"  - {det.label} با اطمینان {det.confidence:.0%} در موقعیت {det.bbox}")
        return "\n".join(lines)

    def _build_user_message(self, summary: str, encoded_frame: Optional[str]) -> list:
        content = [{"type": "text", "text": f"اطلاعات تشخیص:\n{summary}"}]
        if encoded_frame:
            content.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/jpeg;base64,{encoded_frame}", "detail": "low"},
            })
        return content

    def _call_vlm(self, messages: list) -> dict:
        with httpx.Client(timeout=15.0) as client:
            response = client.post(
                f"{self.endpoint}/chat/completions",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={"model": self.model, "messages": messages,
                      "max_tokens": 256, "response_format": {"type": "json_object"}},
            )
            response.raise_for_status()
            return response.json()

    def _parse_response(self, response: dict,
                        perception_result: PerceptionResult) -> SceneAnalysis:
        content = response["choices"][0]["message"]["content"]
        data = json.loads(content)
        target_bbox = None
        if data.get("requires_ptz_track"):
            persons = [d for d in perception_result.detections if d.is_person]
            if persons:
                target_bbox = persons[0].bbox
        return SceneAnalysis(
            threat_level=int(data.get("threat_level", 0)),
            description=data.get("description", ""),
            recommended_action=data.get("recommended_action", "monitor"),
            requires_ptz_track=bool(data.get("requires_ptz_track", False)),
            target_bbox=target_bbox,
        )

لایه اقدام (Action) — واکنش خودکار Agent

"""
action_layer.py
لایه اقدام: کنترل PTZ، ارسال نوتیفیکیشن، فعال‌سازی آژیر
"""
import httpx
import logging
from reasoning_layer import SceneAnalysis
from dahua_onvif_controller import DahuaONVIFController

logger = logging.getLogger(__name__)


class ActionLayer:
    def __init__(self, ptz_controller: DahuaONVIFController,
                 webhook_url: str = "", telegram_bot_token: str = "",
                 telegram_chat_id: str = ""):
        self.ptz = ptz_controller
        self.webhook_url = webhook_url
        self.telegram_bot_token = telegram_bot_token
        self.telegram_chat_id = telegram_chat_id

    def execute(self, analysis: SceneAnalysis) -> None:
        if analysis is None:
            return
        logger.info("اقدام: %s | سطح تهدید: %d | توضیح: %s",
                    analysis.recommended_action, analysis.threat_level, analysis.description)
        action_map = {
            "monitor": self._do_monitor,
            "track_ptz": self._do_ptz_track,
            "trigger_alarm": self._do_trigger_alarm,
            "notify_operator": self._do_notify,
        }
        handler = action_map.get(analysis.recommended_action, self._do_monitor)
        handler(analysis)

    def _do_monitor(self, analysis: SceneAnalysis) -> None:
        logger.info("حالت نظارت: ثبت رویداد — %s", analysis.description)

    def _do_ptz_track(self, analysis: SceneAnalysis) -> None:
        if not analysis.target_bbox:
            logger.warning("هدف برای PTZ tracking مشخص نشد.")
            return
        x1, y1, x2, y2 = analysis.target_bbox
        center_x = (x1 + x2) / 2
        center_y = (y1 + y2) / 2
        pan_error  = center_x - 0.5
        tilt_error = 0.5 - center_y
        gain = 0.6
        pan_velocity  = max(-0.8, min(0.8, pan_error  * gain))
        tilt_velocity = max(-0.8, min(0.8, tilt_error * gain))
        logger.info("PTZ tracking: pan=%.2f, tilt=%.2f", pan_velocity, tilt_velocity)
        self.ptz.ptz_continuous_move(pan=pan_velocity, tilt=tilt_velocity, duration_seconds=0.5)

    def _do_trigger_alarm(self, analysis: SceneAnalysis) -> None:
        if not self.webhook_url:
            logger.warning("Webhook URL تنظیم نشده است.")
            return
        payload = {"event": "alarm_trigger", "threat_level": analysis.threat_level,
                   "description": analysis.description, "timestamp": __import__("time").time()}
        try:
            with httpx.Client(timeout=5.0) as client:
                response = client.post(self.webhook_url, json=payload)
                response.raise_for_status()
                logger.info("Webhook ارسال شد — Status: %d", response.status_code)
        except httpx.HTTPError as exc:
            logger.error("خطا در ارسال Webhook: %s", exc)

    def _do_notify(self, analysis: SceneAnalysis) -> None:
        if not self.telegram_bot_token:
            return
        message = (f"🔔 هشدار دوربین مداربسته\n"
                   f"سطح تهدید: {analysis.threat_level}/2\n"
                   f"توضیح: {analysis.description}")
        url = f"https://api.telegram.org/bot{self.telegram_bot_token}/sendMessage"
        try:
            with httpx.Client(timeout=5.0) as client:
                client.post(url, json={"chat_id": self.telegram_chat_id, "text": message})
        except httpx.HTTPError as exc:
            logger.error("خطا در ارسال نوتیفیکیشن: %s", exc)

اورکسترایتور نهایی — اتصال سه لایه

"""
agent_orchestrator.py
اورکسترایتور اصلی AI Agent — اتصال Perception, Reasoning, Action
"""
import logging
from dahua_rtsp_reader import DahuaCamera, RTSPStreamReader
from dahua_onvif_controller import DahuaONVIFController
from perception_layer import PerceptionLayer
from reasoning_layer import ReasoningLayer
from action_layer import ActionLayer

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def run_agent(
    camera_host: str,
    camera_user: str,
    camera_pass: str,
    openai_api_key: str,
    webhook_url: str = "",
    inference_interval: int = 5,
) -> None:
    camera = DahuaCamera(host=camera_host, username=camera_user,
                         password=camera_pass, use_substream=True)

    ptz_controller = DahuaONVIFController(camera_host, 80, camera_user, camera_pass)
    ptz_controller.connect()

    perception = PerceptionLayer(model_path="yolov10n.pt", confidence_threshold=0.45)
    reasoning  = ReasoningLayer(api_key=openai_api_key)
    action     = ActionLayer(ptz_controller=ptz_controller, webhook_url=webhook_url)

    logger.info("AI Agent راه‌اندازی شد. شروع نظارت...")

    reader = RTSPStreamReader(camera)
    frame_count = 0

    for frame in reader.frame_generator():
        frame_count += 1
        perception_result = perception.process_frame(frame)

        if perception_result.has_suspicious_activity and frame_count % inference_interval == 0:
            logger.info("رویداد مشکوک در فریم %d — ارسال به VLM...", frame_count)
            scene_analysis = reasoning.analyze(perception_result)
            action.execute(scene_analysis)


if __name__ == "__main__":
    import os
    from dotenv import load_dotenv
    load_dotenv()

    run_agent(
        camera_host=os.getenv("DAHUA_HOST", "192.168.1.108"),
        camera_user=os.getenv("DAHUA_USER"),
        camera_pass=os.getenv("DAHUA_PASS"),
        openai_api_key=os.getenv("OPENAI_API_KEY"),
        webhook_url=os.getenv("WEBHOOK_URL", ""),
        inference_interval=5,
    )

نکات حیاتی عیب‌یابی و بهینه‌سازی

مدیریت تأخیر (Latency) در تحلیل ریل‌تایم

بزرگ‌ترین چالش عملی در این معماری، تعادل بین تأخیر End-to-End و دقت تحلیل است. فراخوانی VLM (GPT-4o) برای هر فریم ممکن است 500ms تا 2s طول بکشد.

# 1. استفاده از Sub-stream به جای Main Stream
camera = DahuaCamera(use_substream=True)

# 2. پردازش غیرهمزمان با ThreadPoolExecutor
from concurrent.futures import ThreadPoolExecutor
executor = ThreadPoolExecutor(max_workers=2)
future = executor.submit(reasoning.analyze, perception_result)

# 3. Frame skipping هوشمند — فقط فریم‌های با تغییر pixel بالا به VLM ارسال شوند
def has_significant_motion(prev_frame, curr_frame, threshold=0.05):
    diff = cv2.absdiff(prev_frame, curr_frame)
    return diff.mean() / 255.0 > threshold

سخت‌کردن امنیتی (Security Hardening) پورت‌های ONVIF

# 1. محدودسازی دسترسی به ONVIF فقط از IP سرور agent
iptables -A INPUT -p tcp --dport 80 -s 192.168.1.50 -j ACCEPT
iptables -A INPUT -p tcp --dport 80 -j DROP

# 2. غیرفعال کردن سرویس‌های غیرضروری
# Setup → Network → Advanced → Disable: UPnP, Multicast
# Setup → Network → HTTPS → فعال‌سازی TLS

# هیچ‌گاه credentials را در کد hard-code نکنید
import os
from dotenv import load_dotenv

load_dotenv()

CAMERA_USER = os.getenv("DAHUA_USER")
CAMERA_PASS = os.getenv("DAHUA_PASS")
OPENAI_KEY  = os.getenv("OPENAI_API_KEY")

مهم: پورت RTSP (554) نباید در اینترنت expose شود. برای دسترسی از خارج از شبکه محلی، از VPN یا SSH Tunnel استفاده کنید.

نتیجه‌گیری و کال‌تو‌اکشن

پیاده‌سازی AI Agent برای دوربین مداربسته داهوا با معماری سه‌لایه Perception → Reasoning → Action یک گام واقعی در مسیر نظارت مستقل (Autonomous Surveillance) است. ترکیب ONVIF Profile S/G/T با مدل‌های Vision-Language و کنترل PTZ، امکاناتی را فراهم می‌کند که تا چند سال پیش فقط در اختیار سیستم‌های نظامی و صنعتی بود.

تهران داهوا با بیش از یک دهه تجربه در توزیع دوربین‌های شبکه داهوا در ایران، طیف کاملی از دوربین‌های IPC، NVR و SpeedDome را با تضمین اصالت کالا، گارانتی معتبر، و پشتیبانی فنی تخصصی ارائه می‌دهد.

برای استعلام قیمت دوربین داهوا، خرید دستگاه NVR داهوا، یا دریافت مشاوره فنی رایگان برای طراحی سیستم نظارتی هوشمند خود، با ما در تماس باشید.

«بالاتر از نوسانات، حق مشتری را در نظر می‌گیریم»

سوالات متداول