Arena SDK

from arena_api.system import system

# 별도의 open/close 호출 불필요
# 모듈 언로드 시 자동으로 정리됨

device_infos = system.device_infos
print(f"발견된 장치: {len(device_infos)}대")

for info in device_infos:
    print(f"  모델: {info['model']}")
    print(f"  시리얼: {info['serial']}")
    print(f"  IP: {info['ip']}")
    print(f"  MAC: {info['mac']}")

# 기본값: 100ms
# LUCID 카메라는 100ms 이내 응답, GigE Vision 스펙상 최대 1초
system.DEVICE_INFOS_TIMEOUT_MILLISEC = 200  # 필요시 늘릴 수 있음
device_infos = system.device_infos

devices = system.create_device()

import time

tries = 0
tries_max = 6
sleep_time_secs = 10

while tries < tries_max:
    devices = system.create_device()
    if devices:
        print(f"{len(devices)}대 연결 성공")
        break
    print(f"시도 {tries+1}/{tries_max}: {sleep_time_secs}초 대기 중...")
    time.sleep(sleep_time_secs)
    tries += 1
else:
    raise Exception("장치를 찾을 수 없습니다")

# MAC 주소로 필터링
target_mac = "00:30:F1:02:03:04"
target_info = [d for d in system.device_infos if d['mac'] == target_mac]
devices = system.create_device(device_infos=target_info)

# 특정 인덱스의 장치만 연결
device_info = system.device_infos[0]
devices = system.create_device(device_infos=device_info)

# 여러 장치를 선택적으로 연결
all_infos = system.device_infos
devices = system.create_device(device_infos=all_infos[:2])  # 처음 2대만

devices = system.create_device()
device = system.select_device(devices)  # 1대면 자동 선택, 여러 대면 사용자 선택

# 단일 장치
system.force_ip({
    'mac': '00:30:F1:02:03:04',
    'ip': '192.168.1.50',
    'subnetmask': '255.255.255.0',
    'defaultgateway': '192.168.1.1'
})

# 다중 장치 동시 설정
system.force_ip([
    {'mac': '00:30:F1:02:03:04', 'ip': '192.168.1.50',
     'subnetmask': '255.255.255.0', 'defaultgateway': '192.168.1.1'},
    {'mac': '00:30:F1:02:03:05', 'ip': '192.168.1.51',
     'subnetmask': '255.255.255.0', 'defaultgateway': '192.168.1.1'}
])

import ipaddress

devices = system.create_device()
device = system.select_device(devices)
nodemap = device.nodemap

# IP 주소 설정
new_ip = ipaddress.IPv4Address("169.254.3.2")
new_mask = ipaddress.IPv4Address("255.255.0.0")

nodemap.get_node("GevPersistentIPAddress").value = int(new_ip)
nodemap.get_node("GevPersistentSubnetMask").value = int(new_mask)

# 고정 IP 활성화, DHCP 비활성화
nodemap.get_node("GevCurrentIPConfigurationPersistentIP").value = True
nodemap.get_node("GevCurrentIPConfigurationDHCP").value = False

# (선택) ARP 충돌 감지 비활성화
nodemap.get_node("GevPersistentARPConflictDetectionEnable").value = False

# 유니캐스트 대상 추가
system.add_unicast_discovery_device("192.168.2.50")

# 검색 시 원격 장치도 포함됨
device_infos = system.device_infos

# 유니캐스트 대상 제거
system.remove_unicast_discovery_device("192.168.2.50")

# 모든 유니캐스트 대상 제거
system.remove_unicast_discovery_device(None)

tl_stream = device.tl_stream_nodemap

# 패킷 크기 자동 협상 (권장)
tl_stream['StreamAutoNegotiatePacketSize'].value = True

# 패킷 재전송 활성화 (UDP 손실 복구)
tl_stream['StreamPacketResendEnable'].value = True

# 버퍼 핸들링 모드 설정
tl_stream['StreamBufferHandlingMode'].value = 'NewestOnly'

# 컨텍스트 매니저 사용 (권장)
with device.start_stream():
    buffer = device.get_buffer()
    print(f"이미지: {buffer.width}x{buffer.height}, 포맷: {buffer.pixel_format.name}")
    device.requeue_buffer(buffer)

# 여러 프레임 획득
with device.start_stream():
    for i in range(10):
        buffer = device.get_buffer(timeout=2000)  # 2초 타임아웃
        print(f"프레임 {i}: {buffer.width}x{buffer.height}")
        device.requeue_buffer(buffer)

# 모든 장치 해제
system.destroy_device()

# 특정 장치만 해제
system.destroy_device(device)

# 여러 장치 선택 해제
system.destroy_device([device1, device2])

from arena_api.callback import callback, callback_function

# 콜백 함수 정의
@callback_function.system.on_device_disconnected
def on_disconnect(device):
    print(f'장치 연결 해제됨: {device}')

# 콜백 등록
handle = callback.register(system, on_disconnect, watched_device=device)

# 연결 상태 확인
if device.is_connected():
    print("연결됨")
else:
    print("연결 해제됨")

# 콜백 해제
callback.deregister(handle)

import threading

# 모든 장치 생성
devices = system.create_device()

# 각 장치 설정
for device in devices:
    nodemap = device.nodemap
    tl_stream = device.tl_stream_nodemap

    nodemap.get_node("AcquisitionMode").value = "Continuous"
    tl_stream['StreamAutoNegotiatePacketSize'].value = True
    tl_stream['StreamPacketResendEnable'].value = True
    tl_stream['StreamBufferHandlingMode'].value = 'NewestOnly'

# 병렬 획득 함수
def acquire(device, num_frames=10):
    with device.start_stream():
        for i in range(num_frames):
            buffer = device.get_buffer(timeout=2000)
            print(f"[{device}] 프레임 {i}: {buffer.width}x{buffer.height}")
            device.requeue_buffer(buffer)

# 스레드로 병렬 실행
threads = [threading.Thread(target=acquire, args=(d,)) for d in devices]
for t in threads:
    t.start()
for t in threads:
    t.join()

# 정리
system.destroy_device()

import time
from arena_api.system import system

def main():
    # 1. 장치 검색
    print("장치 검색 중...")
    device_infos = system.device_infos
    print(f"발견된 장치: {len(device_infos)}대")

    if not device_infos:
        print("장치를 찾을 수 없습니다")
        return

    # 2. 장치 생성 (재시도 포함)
    tries = 0
    while tries < 6:
        devices = system.create_device()
        if devices:
            break
        print(f"대기 중... ({tries+1}/6)")
        time.sleep(10)
        tries += 1
    else:
        raise Exception("장치 연결 실패")

    # 3. 장치 선택
    device = system.select_device(devices)

    # 4. 스트림 설정
    tl_stream = device.tl_stream_nodemap
    tl_stream['StreamAutoNegotiatePacketSize'].value = True
    tl_stream['StreamPacketResendEnable'].value = True

    # 5. 이미지 획득
    with device.start_stream():
        buffer = device.get_buffer()
        print(f"이미지: {buffer.width}x{buffer.height}")
        print(f"포맷: {buffer.pixel_format.name}")
        device.requeue_buffer(buffer)

    # 6. 정리
    system.destroy_device()
    print("연결 종료")

if __name__ == '__main__':
    main()

nodes = nodemap.get_node(['ExposureAuto', 'ExposureTime'])
nodes['ExposureAuto'].value = 'Off'
nodes['ExposureTime'].value = 4000.0  # 마이크로초

nodemap['ConversionGain'].value = 'Low'

nodemap.get_node("LUTEnable").value = True

node_lut_index = nodemap.get_node("LUTIndex")
node_lut_value = nodemap.get_node("LUTValue")

# 강도 반전 예시
SLOPE = -1
for i in range(node_lut_index.max + 1):
    node_lut_index.value = i
    node_lut_value.value = SLOPE * i + node_lut_index.max

# 노드 값 설정 전 범위 확인
node = nodemap.get_node('ExposureTime')
if node.is_writable:
    value = min(max(desired_value, node.min), node.max)
    node.value = value

# 열거형 노드의 사용 가능한 값 확인
enum_node = nodemap.get_node('ExposureAuto')
print(enum_node.enumentry_names)  # ['Off', 'Once', 'Continuous']

# 커맨드 노드 실행
cmd_node = nodemap.get_node('DefectCorrectionApply')
cmd_node.execute()