mirror of
https://github.com/intrepidcs/libicsneo.git
synced 2026-08-05 09:28:40 +02:00
Docs: Refactor icsneopy examples
This commit is contained in:
committed by
Kyle Schwarz
parent
cf2cf3e28b
commit
3c0c0dd44c
@@ -0,0 +1,125 @@
|
||||
"""
|
||||
Complete CAN example using icsneopy library.
|
||||
|
||||
Demonstrates device setup and CAN frame transmission/reception.
|
||||
"""
|
||||
|
||||
import icsneopy
|
||||
import time
|
||||
|
||||
|
||||
def setup_device():
|
||||
"""Initialize CAN device."""
|
||||
devices = icsneopy.find_all_devices()
|
||||
if not devices:
|
||||
raise RuntimeError("No devices found")
|
||||
|
||||
device = devices[0]
|
||||
print(f"Using device: {device}")
|
||||
return device
|
||||
|
||||
|
||||
def open_device(device):
|
||||
"""Open device connection."""
|
||||
try:
|
||||
if not device.open():
|
||||
raise RuntimeError("Failed to open device")
|
||||
|
||||
if not device.go_online():
|
||||
device.close()
|
||||
raise RuntimeError("Failed to go online")
|
||||
|
||||
print("Device initialized successfully")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"Device setup failed: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def transmit_can_frame(device, arbid, data):
|
||||
"""Transmit a CAN frame."""
|
||||
frame = icsneopy.CANMessage()
|
||||
frame.network = icsneopy.Network(icsneopy.Network.NetID.DWCAN_01)
|
||||
frame.arbid = arbid
|
||||
frame.data = data
|
||||
|
||||
success = device.transmit(frame)
|
||||
if success:
|
||||
print(f"Frame transmitted: ID=0x{arbid:03X}, Data={list(data)}")
|
||||
else:
|
||||
print(f"Failed to transmit frame ID=0x{arbid:03X}")
|
||||
|
||||
return success
|
||||
|
||||
|
||||
def setup_can_reception(device):
|
||||
"""Configure CAN frame reception with callback."""
|
||||
frame_count = 0
|
||||
|
||||
def frame_handler(frame):
|
||||
nonlocal frame_count
|
||||
frame_count += 1
|
||||
print(f"[RX {frame_count}] ID: 0x{frame.arbid:03X}, "
|
||||
f"Data: {[hex(b) for b in frame.data]}, "
|
||||
f"Length: {len(frame.data)}")
|
||||
|
||||
frame_filter = icsneopy.MessageFilter(icsneopy.Network.NetID.DWCAN_01)
|
||||
callback = icsneopy.MessageCallback(frame_handler, frame_filter)
|
||||
device.add_message_callback(callback)
|
||||
|
||||
print("CAN frame reception configured")
|
||||
return 0
|
||||
|
||||
|
||||
def cleanup_device(device):
|
||||
"""Close device connection."""
|
||||
if device:
|
||||
device.close()
|
||||
print("Device connection closed")
|
||||
|
||||
|
||||
def main():
|
||||
"""Complete CAN example with proper error handling."""
|
||||
device = None
|
||||
|
||||
try:
|
||||
# Setup device
|
||||
device = setup_device()
|
||||
|
||||
# Open device
|
||||
if not open_device(device):
|
||||
raise RuntimeError("Failed to initialize device")
|
||||
|
||||
# Setup frame reception
|
||||
setup_can_reception(device)
|
||||
|
||||
# Transmit test frames
|
||||
test_frames = [
|
||||
(0x123, (0x01, 0x02, 0x03, 0x04)),
|
||||
(0x456, (0x05, 0x06, 0x07, 0x08)),
|
||||
(0x789, (0x09, 0x0A, 0x0B, 0x0C))
|
||||
]
|
||||
|
||||
for arbid, data in test_frames:
|
||||
transmit_result = transmit_can_frame(device, arbid, data)
|
||||
if not transmit_result:
|
||||
print(f"Warning: Failed to transmit frame ID=0x{arbid:03X}")
|
||||
time.sleep(0.1)
|
||||
|
||||
# Listen for responses
|
||||
print("Listening for CAN frames for 5 seconds...")
|
||||
time.sleep(5)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
return 1
|
||||
|
||||
finally:
|
||||
cleanup_device(device)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,47 @@
|
||||
"""
|
||||
Basic CAN frame reception example using icsneopy library.
|
||||
|
||||
Demonstrates how to receive CAN frames on DW CAN 01 using callback handlers.
|
||||
"""
|
||||
|
||||
import icsneopy
|
||||
import time
|
||||
|
||||
|
||||
def receive_can_frames():
|
||||
"""Receive CAN frames with callback handling."""
|
||||
devices = icsneopy.find_all_devices()
|
||||
if not devices:
|
||||
raise RuntimeError("No devices found")
|
||||
|
||||
device = devices[0]
|
||||
frame_count = 0
|
||||
|
||||
def on_frame(frame):
|
||||
nonlocal frame_count
|
||||
frame_count += 1
|
||||
print(f"[RX {frame_count}] ID: 0x{frame.arbid:03X}, "
|
||||
f"Data: {[hex(b) for b in frame.data]}")
|
||||
|
||||
frame_filter = icsneopy.MessageFilter(icsneopy.Network.NetID.DWCAN_01)
|
||||
callback = icsneopy.MessageCallback(on_frame, frame_filter)
|
||||
|
||||
try:
|
||||
if not device.open():
|
||||
raise RuntimeError("Failed to open device")
|
||||
|
||||
if not device.go_online():
|
||||
raise RuntimeError("Failed to go online")
|
||||
|
||||
device.add_message_callback(callback)
|
||||
print("Listening for CAN frames for 10 seconds...")
|
||||
time.sleep(10)
|
||||
|
||||
print(f"Total frames received: {frame_count}")
|
||||
|
||||
finally:
|
||||
device.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
receive_can_frames()
|
||||
@@ -0,0 +1,41 @@
|
||||
"""
|
||||
Basic CAN frame transmission example using icsneopy library.
|
||||
|
||||
Demonstrates how to transmit CAN frames on DW CAN 01.
|
||||
"""
|
||||
|
||||
import icsneopy
|
||||
|
||||
|
||||
def transmit_can_frame():
|
||||
"""Transmit a CAN frame."""
|
||||
devices = icsneopy.find_all_devices()
|
||||
if not devices:
|
||||
raise RuntimeError("No devices found")
|
||||
|
||||
device = devices[0]
|
||||
|
||||
try:
|
||||
if not device.open():
|
||||
raise RuntimeError("Failed to open device")
|
||||
|
||||
if not device.go_online():
|
||||
raise RuntimeError("Failed to go online")
|
||||
|
||||
frame = icsneopy.CANMessage()
|
||||
frame.network = icsneopy.Network(icsneopy.Network.NetID.DWCAN_01)
|
||||
frame.arbid = 0x123
|
||||
frame.data = (0x01, 0x02, 0x03, 0x04)
|
||||
|
||||
success = device.transmit(frame)
|
||||
if success:
|
||||
print(f"Frame transmitted: ID=0x{frame.arbid:03X}")
|
||||
else:
|
||||
print("Failed to transmit frame")
|
||||
|
||||
finally:
|
||||
device.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
transmit_can_frame()
|
||||
@@ -0,0 +1,53 @@
|
||||
"""
|
||||
DoIP activation control example using icsneopy library.
|
||||
|
||||
Demonstrates DoIP (Diagnostics over Internet Protocol) Ethernet activation
|
||||
control with comprehensive error handling and state validation.
|
||||
"""
|
||||
|
||||
import icsneopy
|
||||
import time
|
||||
|
||||
|
||||
def doip_activation_demo():
|
||||
"""Demonstrate DoIP Ethernet activation control with error handling."""
|
||||
devices = icsneopy.find_all_devices()
|
||||
if not devices:
|
||||
raise RuntimeError("No devices found")
|
||||
|
||||
device = devices[0]
|
||||
print(f"Using {device} for DoIP activation control")
|
||||
|
||||
try:
|
||||
if not device.open():
|
||||
raise RuntimeError("Failed to open device")
|
||||
|
||||
if not device.go_online():
|
||||
raise RuntimeError("Failed to go online")
|
||||
|
||||
initial_state = device.get_digital_io(icsneopy.IO.EthernetActivation, 1)
|
||||
print(f"Initial DoIP activation state: {initial_state}")
|
||||
|
||||
print("Activating DoIP Ethernet...")
|
||||
device.set_digital_io(icsneopy.IO.EthernetActivation, 1, True)
|
||||
time.sleep(1)
|
||||
|
||||
active_state = device.get_digital_io(icsneopy.IO.EthernetActivation, 1)
|
||||
print(f"DoIP activated: {active_state}")
|
||||
time.sleep(2)
|
||||
|
||||
print("Deactivating DoIP Ethernet...")
|
||||
device.set_digital_io(icsneopy.IO.EthernetActivation, 1, False)
|
||||
time.sleep(1)
|
||||
|
||||
final_state = device.get_digital_io(icsneopy.IO.EthernetActivation, 1)
|
||||
print(f"DoIP deactivated: {final_state}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"DoIP control error: {e}")
|
||||
finally:
|
||||
device.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
doip_activation_demo()
|
||||
@@ -0,0 +1,127 @@
|
||||
"""
|
||||
Complete Ethernet example using icsneopy library.
|
||||
|
||||
Demonstrates device setup and Ethernet frame transmission/reception.
|
||||
"""
|
||||
|
||||
import icsneopy
|
||||
import time
|
||||
|
||||
|
||||
def setup_device():
|
||||
"""Initialize Ethernet device."""
|
||||
devices = icsneopy.find_all_devices()
|
||||
if not devices:
|
||||
raise RuntimeError("No devices found")
|
||||
|
||||
device = devices[0]
|
||||
print(f"Using device: {device}")
|
||||
return device
|
||||
|
||||
|
||||
def open_device(device):
|
||||
"""Open device connection."""
|
||||
try:
|
||||
if not device.open():
|
||||
raise RuntimeError("Failed to open device")
|
||||
|
||||
if not device.go_online():
|
||||
device.close()
|
||||
raise RuntimeError("Failed to go online")
|
||||
|
||||
print("Device initialized successfully")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
print(f"Device setup failed: {e}")
|
||||
return False
|
||||
|
||||
def on_status_message(message):
|
||||
print(f"info: network: {message.network}, state: {message.state}, speed: {message.speed}, duplex: {message.duplex}, mode: {message.mode}")
|
||||
|
||||
|
||||
def setup_ethernet_reception(device):
|
||||
"""Configure Ethernet frame reception with callback."""
|
||||
frame_count = 0
|
||||
|
||||
def frame_handler(frame):
|
||||
nonlocal frame_count
|
||||
frame_count += 1
|
||||
print(f"[RX {frame_count}], "
|
||||
f"Data: {[hex(b) for b in frame.data]}, "
|
||||
f"Length: {len(frame.data)}")
|
||||
frame_filter = icsneopy.MessageFilter(icsneopy.Network.NetID.ETHERNET_01)
|
||||
callback = icsneopy.MessageCallback(frame_handler, frame_filter)
|
||||
device.add_message_callback(callback)
|
||||
|
||||
print("CAN frame reception configured")
|
||||
return 0
|
||||
|
||||
|
||||
def transmit_ethernet_frame(device):
|
||||
"""Transmit an Ethernet frame."""
|
||||
frame = icsneopy.EthernetMessage()
|
||||
frame.network = icsneopy.Network(icsneopy.Network.NetID.ETHERNET_01)
|
||||
frame.data = [
|
||||
0x00, 0xFC, 0x70, 0x00, 0x01, 0x02,
|
||||
0x00, 0xFC, 0x70, 0x00, 0x01, 0x01,
|
||||
0x08, 0x00,
|
||||
0x01, 0xC5, 0x01, 0xC5
|
||||
]
|
||||
|
||||
success = device.transmit(frame)
|
||||
if success:
|
||||
print("Frame transmitted successfully")
|
||||
else:
|
||||
print("Failed to transmit frame")
|
||||
|
||||
return success
|
||||
|
||||
def cleanup_device(device):
|
||||
"""Close device connection."""
|
||||
if device:
|
||||
device.close()
|
||||
print("Device connection closed")
|
||||
|
||||
def main():
|
||||
"""Complete Ethernet example"""
|
||||
device = None
|
||||
|
||||
try:
|
||||
# Initialize device
|
||||
device = setup_device()
|
||||
|
||||
# Open device
|
||||
if not open_device(device):
|
||||
raise RuntimeError("Failed to initialize device")
|
||||
|
||||
filter = icsneopy.MessageFilter(icsneopy.Message.Type.EthernetStatus)
|
||||
status_callback = icsneopy.MessageCallback(on_status_message, filter)
|
||||
device.add_message_callback(status_callback)
|
||||
|
||||
#Setup Ethernet Callback
|
||||
setup_ethernet_reception(device)
|
||||
|
||||
# Transmit an Ethernet frame
|
||||
transmit_result = transmit_ethernet_frame(device)
|
||||
if not transmit_result:
|
||||
print("Warning: Failed to transmit frame")
|
||||
|
||||
# Monitor for a period
|
||||
print("Monitoring for 10 seconds...")
|
||||
time.sleep(10)
|
||||
|
||||
print(f"Monitoring completed.")
|
||||
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
return 1
|
||||
|
||||
finally:
|
||||
cleanup_device(device)
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,37 @@
|
||||
"""
|
||||
Ethernet status monitoring example using icsneopy library.
|
||||
|
||||
Demonstrates how to monitor Ethernet link status changes.
|
||||
"""
|
||||
|
||||
import icsneopy
|
||||
import time
|
||||
|
||||
def main():
|
||||
devices = icsneopy.find_all_devices()
|
||||
if len(devices) == 0:
|
||||
print("error: no devices found")
|
||||
return False
|
||||
|
||||
device = devices[0]
|
||||
print(f"info: monitoring Ethernet status on {device}")
|
||||
|
||||
def on_message(message):
|
||||
print(f"info: network: {message.network}, state: {message.state}, speed: {message.speed}, duplex: {message.duplex}, mode: {message.mode}")
|
||||
|
||||
filter = icsneopy.MessageFilter(icsneopy.Message.Type.EthernetStatus)
|
||||
callback = icsneopy.MessageCallback(on_message, filter)
|
||||
device.add_message_callback(callback)
|
||||
|
||||
if not device.open():
|
||||
print("error: unable to open device")
|
||||
return False
|
||||
|
||||
if not device.go_online():
|
||||
print("error: unable to go online")
|
||||
return False
|
||||
|
||||
while True:
|
||||
time.sleep(1)
|
||||
|
||||
main()
|
||||
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
Basic Ethernet frame transmission example using icsneopy library.
|
||||
|
||||
Demonstrates how to transmit Ethernet frames on Ethernet 01.
|
||||
"""
|
||||
|
||||
import icsneopy
|
||||
|
||||
|
||||
def transmit_ethernet_frame():
|
||||
"""Transmit an Ethernet frame."""
|
||||
devices = icsneopy.find_all_devices()
|
||||
if not devices:
|
||||
raise RuntimeError("No devices found")
|
||||
|
||||
device = devices[0]
|
||||
|
||||
try:
|
||||
if not device.open():
|
||||
raise RuntimeError("Failed to open device")
|
||||
|
||||
if not device.go_online():
|
||||
raise RuntimeError("Failed to go online")
|
||||
|
||||
frame = icsneopy.EthernetMessage()
|
||||
frame.network = icsneopy.Network(icsneopy.Network.NetID.ETHERNET_01)
|
||||
frame.data = [
|
||||
0x00, 0xFC, 0x70, 0x00, 0x01, 0x02,
|
||||
0x00, 0xFC, 0x70, 0x00, 0x01, 0x01,
|
||||
0x08, 0x00,
|
||||
0x01, 0xC5, 0x01, 0xC5
|
||||
]
|
||||
|
||||
success = device.transmit(frame)
|
||||
if success:
|
||||
print("Frame transmitted successfully")
|
||||
else:
|
||||
print("Failed to transmit frame")
|
||||
|
||||
finally:
|
||||
device.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
transmit_ethernet_frame()
|
||||
Reference in New Issue
Block a user