direct init cam

Author: DIYer22

MANIFEST.in

@@ -0,0 +1 @@
+include requirements.txt hik_camera/MvCameraNode-CH.csv README.md

hik_camera/hik_camera.py

@@ -20,12 +20,101 @@
 )
 
 
+def get_host_ip_by_target_ip(target_ip):
+    import socket
+
+    return [
+        (s.connect((target_ip, 80)), s.getsockname()[0], s.close())
+        for s in [socket.socket(socket.AF_INET, socket.SOCK_DGRAM)]
+    ][0][1]
+
+
+def get_setting_df():
+    csv = boxx.pd.read_csv(__file__.replace("hik_camera.py", "MvCameraNode-CH.csv"))
+    setting_df = boxx.pd.DataFrame()
+
+    def to_key(key):
+        if "[" in key:
+            key = key[: key.index("[")]
+        return key.strip()
+
+    def get_depend(key):
+        key = key.strip()
+        if "[" in key:
+            return key[key.index("[") + 1 : -1]
+        return ""
+
+    setting_df["key"] = csv[list(csv)[1]].map(to_key)
+    setting_df["depend"] = csv[list(csv)[1]].map(get_depend)
+    setting_df["dtype"] = csv[list(csv)[2]].map(lambda x: x.strip().lower())
+    return setting_df
+
+
 class HikCamera(hik.MvCamera):
-    def __init__(self):
+    def __init__(self, ip=None, host_ip=None):
         super().__init__()
-        self.get_setting_df()
+        if ip is None:
+            return
+            ip = self.get_all_ips()[0]
+        if host_ip is None:
+            host_ip = get_host_ip_by_target_ip(ip)
+        self._ip = ip
+        self.host_ip = host_ip
+        # self.init_by_ip()
+        self.init_by_enum()
+
+    def setting(self):
+        self.set_exposure(250000)
+        self.setitem("ExposureAuto", "Continuous")
+        # self.adjust_auto_exposure()
+
+        self.pixel_format = "RGB8"
+        self.pixel_format = "BayerGB12Packed"
+        self.setitem("PixelFormat", self.pixel_format)
+
+        self.setitem("GevSCPD", 2000)  # 包延时 ns
+
+    def init_by_spec_ip(self,):
+        """
+        SDK 有 Bug, 调用完"枚举设备" 接口后, 再调用"无枚举连接相机" 会无法打开相机.
+        暂时不能用
+        """
+        stDevInfo = hik.MV_CC_DEVICE_INFO()
+        stGigEDev = hik.MV_GIGE_DEVICE_INFO()
+
+        stGigEDev.nCurrentIp = ip_to_int(self.ip)
+        stGigEDev.nNetExport = ip_to_int(self.host_ip)
+        stDevInfo.nTLayerType = hik.MV_GIGE_DEVICE
+        stDevInfo.SpecialInfo.stGigEInfo = stGigEDev
+        assert not self.MV_CC_CreateHandle(stDevInfo)
+
+    def init_by_enum(self):
+        stDevInfo = self.get_dev_info(self.ip)
+        assert not self.MV_CC_CreateHandle(stDevInfo)
+
+    @classmethod
+    def get_dev_info(cls, ip=None):
+        if not hasattr(cls, "ip_to_dev_info"):
+            ip_to_dev_info = {}
+            deviceList = hik.MV_CC_DEVICE_INFO_LIST()
+            tlayerType = hik.MV_GIGE_DEVICE  # | MV_USB_DEVICE
+            assert not hik.MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
+            for i in range(0, deviceList.nDeviceNum):
+                mvcc_dev_info = cast(
+                    deviceList.pDeviceInfo[i], POINTER(hik.MV_CC_DEVICE_INFO)
+                ).contents
+                _ip = int_to_ip(mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp)
+                if mvcc_dev_info.nTLayerType == hik.MV_GIGE_DEVICE:
+                    ip_to_dev_info[_ip] = mvcc_dev_info
+            cls.ip_to_dev_info = {
+                ip: ip_to_dev_info[ip] for ip in sorted(ip_to_dev_info)
+            }
+        if ip is None:
+            return cls.ip_to_dev_info
+        return cls.ip_to_dev_info[ip]
 
     high_speed_lock = Lock()
+    setting_df = get_setting_df()
 
     def get_frame(self):
         stFrameInfo = self.stFrameInfo
@@ -65,25 +154,6 @@ def get_frame(self):
             )
         return img
 
-    def get_setting_df(self):
-        csv = boxx.pd.read_csv(__file__.replace("hik_camera.py", "MvCameraNode-CH.csv"))
-        self.setting_df = boxx.pd.DataFrame()
-
-        def to_key(key):
-            if "[" in key:
-                key = key[: key.index("[")]
-            return key.strip()
-
-        def get_depend(key):
-            key = key.strip()
-            if "[" in key:
-                return key[key.index("[") + 1 : -1]
-            return ""
-
-        self.setting_df["key"] = csv[list(csv)[1]].map(to_key)
-        self.setting_df["depend"] = csv[list(csv)[1]].map(get_depend)
-        self.setting_df["dtype"] = csv[list(csv)[2]].map(lambda x: x.strip().lower())
-
     def getitem(self, key):
         df = self.setting_df
         dtype = df[df.key == key]["dtype"].iloc[0]
@@ -150,25 +220,6 @@ def adjust_auto_exposure(self, t=2):
         finally:
             self.MV_CC_StopGrabbing()
 
-    pixel_format_values = {
-        "RGB8": 0x02180014,
-        "BayerGB8": 0x0108000A,
-        "BayerGB12": 0x01100012,
-        "BayerGB12Packed": 0x010C002C,
-    }
-
-    def setting(self):
-        # self.adjust_auto_exposure()
-        self.setitem("ExposureAuto", "Off")
-        self.setitem("ExposureTime", 250000)
-        self.setitem("ExposureAuto", "Continuous")
-
-        self.pixel_format = "RGB8"
-        self.pixel_format = "BayerGB12Packed"
-        self.setitem("PixelFormat", self.pixel_format_values[self.pixel_format])
-
-        self.setitem("GevSCPD", 2000)  # 包延时 ns
-
     def __enter__(self):
         assert not self.MV_CC_OpenDevice(hik.MV_ACCESS_Exclusive, 0)
 
@@ -207,45 +258,25 @@ def get_shape(self):
     def __exit__(self, *l):
         self.MV_CC_CloseDevice()
 
-    @staticmethod
-    def get_all_ips():
-        deviceList = hik.MV_CC_DEVICE_INFO_LIST()
-        tlayerType = hik.MV_GIGE_DEVICE  # | MV_USB_DEVICE
-
-        # ch:枚举设备 | en:Enum device
-        assert not hik.MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
-        ips = {}
-        for i in range(0, deviceList.nDeviceNum):
-            mvcc_dev_info = cast(
-                deviceList.pDeviceInfo[i], POINTER(hik.MV_CC_DEVICE_INFO)
-            ).contents
-            if mvcc_dev_info.nTLayerType == hik.MV_GIGE_DEVICE:
-                ips[int_to_ip(mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp)] = i
-        return ips
+    @classmethod
+    def get_all_ips(cls):
+        ip_to_dev_info = cls.get_dev_info()
+        return sorted(ip_to_dev_info)
 
     @classmethod
-    def get_all_cams(cls):
-        deviceList = hik.MV_CC_DEVICE_INFO_LIST()
-        tlayerType = hik.MV_GIGE_DEVICE  # | MV_USB_DEVICE
-
-        # ch:枚举设备 | en:Enum device
-        assert not hik.MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
-        cams = MultiHikCamera()
-        for i in range(0, deviceList.nDeviceNum):
-            mvcc_dev_info = cast(
-                deviceList.pDeviceInfo[i], POINTER(hik.MV_CC_DEVICE_INFO)
-            ).contents
-            if mvcc_dev_info.nTLayerType == hik.MV_GIGE_DEVICE:
-                cam = cls()
-                assert not cam.MV_CC_CreateHandle(mvcc_dev_info)
-                ip = cam.ip
-                cams[ip] = cam
-        cams = MultiHikCamera({ip: cams[ip] for ip in sorted(cams)})
+    def get_cams(cls, ips=None):
+        if ips is None:
+            ips = cls.get_all_ips()
+        else:
+            ips = sorted(ips)
+        cams = MultiHikCamera({ip: cls(ip) for ip in ips})
         return cams
 
+    get_all_cams = get_cams
+
     def MV_CC_CreateHandle(self, mvcc_dev_info):
         self.mvcc_dev_info = mvcc_dev_info
-        self.ip = int_to_ip(mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp)
+        self._ip = int_to_ip(mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp)
         assert not super().MV_CC_CreateHandle(mvcc_dev_info)
 
     def __del__(self):
@@ -269,9 +300,18 @@ def raw_to_uint8_rgb(self, raw, poww=1, demosaicing_method="Malvar2004"):
     def is_raw(self):
         return "Bayer" in self.__dict__.get("pixel_format", "RGB8")
 
+    @property
+    def ip(self):
+        if not hasattr(self, "_ip"):
+            self._ip = self.getitem("GevCurrentIPAddress")
+        return self._ip
+
 
 class MultiHikCamera(dict):
     def __getattr__(self, attr):
+        if not callable(getattr(next(iter(self.values())), attr)):
+            return {k: getattr(cam, attr) for k, v in self.items()}
+
         def func(*args, **kwargs):
             threads = []
             res = {}
@@ -304,25 +344,23 @@ def __exit__(self, *l):
     ips = HikCamera.get_all_ips()
     print(ips)
     ip = list(ips)[0]
-    cams = HikCamera.get_all_cams()
-    cam = cams.get("10.9.5.102", cams[ip])
-    with cams:
-        with boxx.timeit("get_frame"):
-            imgs = cams.get_frame()
-        print("imgs = cams.get_frame()")
-        boxx.tree(imgs)
-        # rgbs = [cam.raw_to_uint8_rgb(raw, poww=.3, demosaicing_method="simple") for raw in imgs.values()]
-        # normas = [norma(img) for img in imgs.values()]
-        # shows-[normas, rgbs]
 
+    cam = HikCamera(ip)
+    with cam:
         for i in range(2):
             with boxx.timeit("cam.get_frame"):
                 img = cam.get_frame()
                 print("cam.get_exposure", cam["ExposureTime"])
-            boxx.show(img)
         if cam.is_raw:
             rgbs = [
                 cam.raw_to_uint8_rgb(img, poww=1),
                 cam.raw_to_uint8_rgb(img, poww=0.3),
             ]
             boxx.show(rgbs)
+
+    cams = HikCamera.get_all_cams()
+    with cams:
+        with boxx.timeit("cams.get_frame"):
+            imgs = cams.get_frame()
+        print("imgs = cams.get_frame()")
+        boxx.tree(imgs)