Eduardo_Scripniciuc/st-ten-1
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master'

Browse Source
This commit is contained in:
neonb 2025-01-22 12:27:30 +01:00
commit 45df1552a1
2 changed files with 108 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
src/components/hikrobot_dll.py
View File

@ -1,4 +1,5 @@
import ctypes import ctypes
import socket
import struct import struct
from ctypes import * from ctypes import *
import sys import sys
@ -144,6 +145,25 @@ class MV_VS_DEVICE_INFO_LIST(ctypes.Structure):
] ]
class PROCESS_MEMORY_COUNTERS(ctypes.Structure):
_fields_ = [
("cb", ctypes.c_ulong),
("PageFaultCount", ctypes.c_ulong),
("PeakWorkingSetSize", ctypes.c_size_t),
("WorkingSetSize", ctypes.c_size_t),
("QuotaPeakPagedPoolUsage", ctypes.c_size_t),
("QuotaPagedPoolUsage", ctypes.c_size_t),
("QuotaPeakNonPagedPoolUsage", ctypes.c_size_t),
("QuotaNonPagedPoolUsage", ctypes.c_size_t),
("PagefileUsage", ctypes.c_size_t),
("PeakPagefileUsage", ctypes.c_size_t),
]
# PPROCESS_MEMORY_COUNTERS is simply a pointer to the struct
PPROCESS_MEMORY_COUNTERS = ctypes.POINTER(PROCESS_MEMORY_COUNTERS)
# Define function signatures # Define function signatures
mv_lib.MV_VS_GetResultData.argtypes = [c_void_p, POINTER(MV_VS_DATA), c_int] mv_lib.MV_VS_GetResultData.argtypes = [c_void_p, POINTER(MV_VS_DATA), c_int]
@ -185,15 +205,27 @@ def MV_VS_GetFrame(handle):
# Process chunk data # Process chunk data
currentDataTag = 0 currentDataTag = 0
while stFrameData.nChunkDataLen > currentDataTag: while stFrameData.nChunkDataLen > currentDataTag:
chunkLength = \ chunkLength = ctypes.c_uint32()
struct.unpack_from("<I", stFrameData.pChunkData, stFrameData.nChunkDataLen - 4 - currentDataTag)[0] chunkID = ctypes.c_uint32()
chunkID = struct.unpack_from("<I", stFrameData.pChunkData, stFrameData.nChunkDataLen - 8 - currentDataTag)[
0]
print(f"chunkLength [{chunkLength}], chunkID: [{chunkID}]") # Calculate source addresses for chunkLength and chunkID
source_address_length = ctypes.addressof(
stFrameData.pChunkData.contents) + stFrameData.nChunkDataLen - 4 - currentDataTag
source_address_id = ctypes.addressof(
stFrameData.pChunkData.contents) + stFrameData.nChunkDataLen - 8 - currentDataTag
if chunkLength <= 0 or chunkLength > stFrameData.nChunkDataLen - 8 - currentDataTag: # Copy data from source to chunkLength and chunkID
# Handle exception ctypes.memmove(ctypes.byref(chunkLength), source_address_length, 4)
ctypes.memmove(ctypes.byref(chunkID), source_address_id, 4)
# Convert network byte order to host byte order
chunkLength.value = socket.ntohl(chunkLength.value)
chunkID.value = socket.ntohl(chunkID.value)
# Print the values
print(f"chunkLength [{chunkLength.value}], chunkID: [{chunkID.value}]")
if chunkLength.value <= 0 or chunkLength.value > stFrameData.nChunkDataLen - 8 - currentDataTag:
break break
if chunkID == CHUNK_RESULT_PORT: # Result data in JSON format if chunkID == CHUNK_RESULT_PORT: # Result data in JSON format
@ -203,24 +235,42 @@ def MV_VS_GetFrame(handle):
print(f"chResultInfo [{chResultInfo.value.decode()}] end") print(f"chResultInfo [{chResultInfo.value.decode()}] end")
elif chunkID == CHUNK_MASK_IMAGE_PORT: # Mask image data elif chunkID == CHUNK_MASK_IMAGE_PORT: # Mask image data
maskModID = struct.unpack_from("<I", stFrameData.pChunkData, maskModID = ctypes.c_uint32()
stFrameData.nChunkDataLen - 8 - chunkLength - currentDataTag)[0] maskModFormat = ctypes.c_uint32()
maskModFormat = struct.unpack_from("<I", stFrameData.pChunkData, maskModWidth = ctypes.c_uint32()
stFrameData.nChunkDataLen - 8 - chunkLength - currentDataTag + 4)[0] maskModHeight = ctypes.c_uint32()
maskModWidth = struct.unpack_from("<I", stFrameData.pChunkData,
stFrameData.nChunkDataLen - 8 - chunkLength - currentDataTag + 8)[0] # Calculate source addresses
maskModHeight = struct.unpack_from("<I", stFrameData.pChunkData, source_address_id = ctypes.addressof(
stFrameData.nChunkDataLen - 8 - chunkLength - currentDataTag + 12)[0] stFrameData.pChunkData.contents) + stFrameData.nChunkDataLen - 8 - chunkLength.value - currentDataTag
source_address_format = source_address_id + 4
source_address_width = source_address_id + 8
source_address_height = source_address_id + 12
# Copy data from source addresses to corresponding variables
ctypes.memmove(ctypes.byref(maskModID), source_address_id, 4)
ctypes.memmove(ctypes.byref(maskModFormat), source_address_format, 4)
ctypes.memmove(ctypes.byref(maskModWidth), source_address_width, 4)
ctypes.memmove(ctypes.byref(maskModHeight), source_address_height, 4)
print( print(
f"Mask ModID[{maskModID}], ModFormat[{maskModFormat}], ModWidth[{maskModWidth}], ModHeight[{maskModHeight}], ") f"Mask ModID[{maskModID}], ModFormat[{maskModFormat}], ModWidth[{maskModWidth}], ModHeight[{maskModHeight}], ")
chMaskModImageData = ctypes.cast( try:
stFrameData.pChunkData + (stFrameData.nChunkDataLen - 8 - chunkLength - currentDataTag + 16), chMaskModImageData = ctypes.cast(ctypes.create_string_buffer(chunkLength.value - 16 + 1),
ctypes.POINTER(ctypes.c_char)) ctypes.POINTER(ctypes.c_char))
print(f"Mask data length [{chunkLength - 16}]") if chMaskModImageData:
ctypes.memset(chMaskModImageData, 0, chunkLength.value - 16 + 1)
chImageData = ctypes.cast(ctypes.addressof(
stFrameData.pChunkData.contents) + stFrameData.nChunkDataLen - 8 - chunkLength.value - currentDataTag + 16,
ctypes.POINTER(ctypes.c_char))
ctypes.memmove(chMaskModImageData, chImageData, chunkLength.value - 16)
print(f"Mask data length [{chunkLength.value - 16}]")
currentDataTag += 8 + chunkLength except (MemoryError, TypeError) as e:
print(f"Error allocating or copying memory: {e}")
currentDataTag = currentDataTag + 8 + chunkLength.value
else: else:
print(f"Get frame buffer fail! nRet [0x{nRet:x}]") print(f"Get frame buffer fail! nRet [0x{nRet:x}]")
@ -230,4 +280,10 @@ def MV_VS_GetFrame(handle):
if nRet != MV_VS_OK: if nRet != MV_VS_OK:
print(f"Release frame buffer fail! nRet [0x{nRet:x}]") print(f"Release frame buffer fail! nRet [0x{nRet:x}]")
else: else:
print(f"MV_VS_GetResultData fail! nRet [0x{nRet & 0xFFFFFFFF:x}]") print(f"MV_VS_GetResultData fail! nRet [0x{nRet&0xFFFFFFFF:x}]")
nRet = mv_lib.MV_VS_ReleaseResultData(handle, byref(stFrameData))
if nRet != MV_VS_OK:
print(f"Release frame buffer fail! nRet [0x{nRet&0xFFFFFFFF:x}]")
else:
print("Release frame buffer OK")

43
src/test/hikrobot_sc/test_hikrobot_sc.py
View File

@ -1,5 +1,6 @@
import ctypes import ctypes
import os import os
import threading
import time import time
from ctypes import * from ctypes import *
@ -68,26 +69,44 @@ def main():
else: else:
print("CommandImageMode OK") print("CommandImageMode OK")
nRet1 = mv_lib.MV_VS_SetEnumValue(pHandle1, b"AcquisitionMode", 2) nRet1 = mv_lib.MV_VS_SetEnumValue(pHandle1, b"AcquisitionMode", 0)
nRet2 = mv_lib.MV_VS_SetEnumValue(pHandle2, b"AcquisitionMode", 2) nRet2 = mv_lib.MV_VS_SetEnumValue(pHandle2, b"AcquisitionMode", 0)
if nRet1 != MV_VS_OK or nRet2 != MV_VS_OK: if nRet1 != MV_VS_OK or nRet2 != MV_VS_OK:
print(f"AcquisitionMode failed! [{nRet1 & 0xFFFFFFFF:#x} {nRet2 & 0xFFFFFFFF:#x}]") print(f"AcquisitionMode failed! [{nRet1 & 0xFFFFFFFF:#x} {nRet2 & 0xFFFFFFFF:#x}]")
return return
else: else:
print("AcquisitionMode OK") print("AcquisitionMode OK")
nRet1 = mv_lib.MV_VS_StartRun(pHandle1) stMemory = PROCESS_MEMORY_COUNTERS()
nRet2 = mv_lib.MV_VS_StartRun(pHandle2) ctypes.windll.psapi.GetProcessMemoryInfo(
time.sleep(1) ctypes.windll.kernel32.GetCurrentProcess(), ctypes.byref(stMemory), ctypes.sizeof(stMemory)
)
if nRet1 != MV_VS_OK or nRet2 != MV_VS_OK: # Get the current time
print(f"Start run failed! [{nRet1&0xFFFFFFFF:#x} {nRet2&0xFFFFFFFF:#x}]") current_time = time.localtime()
return
else:
print("Start run OK")
MV_VS_GetFrame(pHandle1) # Print the information
MV_VS_GetFrame(pHandle2) print(
f"Start test Get/Release result data, begin time [{current_time.tm_hour + 8:02d}:{current_time.tm_min:02d}:{current_time.tm_sec:02d}], "
f"current memory [{stMemory.WorkingSetSize / 1024} KB]"
)
# Create the thread
thread = threading.Thread(target=MV_VS_GetFrame, args=(pHandle1,))
# Start the thread
thread.start()
while True:
print("main loop cycle")
nRet1 = mv_lib.MV_VS_StartRun(pHandle1)
nRet2 = mv_lib.MV_VS_StartRun(pHandle2)
if nRet1 != MV_VS_OK or nRet2 != MV_VS_OK:
print(f"Start run failed! [{nRet1 & 0xFFFFFFFF:#x} {nRet2 & 0xFFFFFFFF:#x}]")
return
else:
print("Start run OK")
time.sleep(1)
return 0 return 0