Loading skill
Install any skill in seconds. Free to start, no credit card required.
Get Started Free →Expert in AMQP (Advanced Message Queuing Protocol) middleware for automotive enterprise integration using RabbitMQ and Azure Service Bus. Covers 6 topics across middleware domain. Includes 6 skill files covering AMQP 1.0 OASIS Standard, AUTOSAR Adaptive (comparison), AUTOSAR Adaptive Platform, AWS IoT Core best practices, Apache Qpid Proton, AutomationML for data modeling, Azure IoT Hub protocols, Azure Service Bus protocols and more.
| Test case | Without → With | Effect | Δ tokens | Δ turns |
|---|---|---|---|---|
| case-10 | ✓→✓ | ＝ Same ✓ | 730% | 0% |
| case-02 | ✓→✓ | ＝ Same ✓ | 929% | 0% |
| case-03 | ✓→✓ | ＝ Same ✓ | 910% | 0% |
| case-04 | ✓→✓ | ＝ Same ✓ | 1259% | 0% |
| case-05 | ✓→✓ | ＝ Same ✓ | 942% | 0% |
6 skill files covering middleware domain for automotive software engineering.
You are an expert in AMQP middleware for automotive enterprise systems and factory integration.
Exchange: vehicle.commands Binding: remote_lock → queue.remote_lock Message routing_key: remote_lock → delivered to queue.remote_lock
Exchange: vehicle.telemetry Binding: vehicle.*.battery → queue.battery_all Binding: vehicle.tesla.# → queue.tesla_fleet
Exchange: ota.broadcast All queues bound to exchange receive message
Producer: Robot PLC Exchange: factory.station (direct) Routing Key: station.battery_install.complete Queue: station.quality_check Consumer: QA workstation
Producer: Order management system Exchange: vehicle.config (topic) Message: {"vin": "...", "trim": "premium", "color": "blue"} Routing Key: vehicle.model_s.premium Bindings:
Exchange: ota.release (fanout) Queues: ota.batch_1, ota.batch_2, ..., ota.batch_100 Each queue has 10,000 vehicle IDs Workers consume from queues at controlled rate
Exchange: supply.events (topic) Routing Key: supply.battery.LG.shipped Queue: inventory.battery (TTL=48h, DLX for unprocessed) Consumer: ERP system
python import pika import json
def publish_vehicle_config(vin, config): credentials = pika.PlainCredentials('vehicle_app', 'secure_password') parameters = pika.ConnectionParameters( host='rabbitmq.factory.local', port=5672, virtual_host='/production', credentials=credentials, heartbeat=600, blocked_connection_timeout=300 )
connection = pika.BlockingConnection(parameters) channel = connection.channel()
# Declare exchange (idempotent) channel.exchange_declare( exchange='vehicle.config', exchange_type='topic', durable=True )
routing_key = f"vehicle.{config'model']}.{config'trim']}" message = json.dumps({ "vin": vin, "config": config, "timestamp": datetime.utcnow().isoformat() })
# Publish with persistence channel.basic_publish( exchange='vehicle.config', routing_key=routing_key, body=message, properties=pika.BasicProperties( delivery_mode=2, # Persistent content_type='application/json', correlation_id=str(uuid.uuid4()) ) )
connection.close()
python def process_message(ch, method, properties, body): try: data = json.loads(body) vin = data"vin"] config = data"config"]
# Process configuration apply_vehicle_config(vin, config)
# Manual acknowledgment after successful processing ch.basic_ack(delivery_tag=method.delivery_tag)
except Exception as e: print(f"Error processing message: {e}") # Reject and requeue (retry) ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
def start_consumer(): connection = pika.BlockingConnection(parameters) channel = connection.channel()
# Declare queue with DLX channel.queue_declare( queue='queue.paint_shop', durable=True, arguments={ 'x-dead-letter-exchange': 'dlx.vehicle.config', 'x-message-ttl': 86400000, # 24 hours 'x-max-length': 10000 } )
# Bind to exchange channel.queue_bind( exchange='vehicle.config', queue='queue.paint_shop', routing_key='vehicle.model_s.' )
# Set QoS: prefetch 10 messages channel.basic_qos(prefetch_count=10)
# Start consuming channel.basic_consume( queue='queue.paint_shop', on_message_callback=process_message, auto_ack=False # Manual ack )
print("Waiting for messages...") channel.start_consuming()
python def publish_with_confirm(channel, exchange, routing_key, message): # Enable publisher confirms channel.confirm_delivery()
try: channel.basic_publish( exchange=exchange, routing_key=routing_key, body=message, properties=pika.BasicProperties(delivery_mode=2), mandatory=True # Return if unroutable ) print("Message confirmed by broker") except pika.exceptions.UnroutableError: print("Message was returned (no queue bound)") except pika.exceptions.NackError: print("Message was nacked by broker")
python class VehicleRPCClient: def __init__(self): self.connection = pika.BlockingConnection(parameters) self.channel = self.connection.channel()
# Exclusive queue for responses result = self.channel.queue_declare(queue='', exclusive=True) self.callback_queue = result.method.queue self.channel.basic_consume( queue=self.callback_queue, on_message_callback=self.on_response, auto_ack=True )
self.response = None self.corr_id = None
def on_response(self, ch, method, props, body): if self.corr_id == props.correlation_id: self.response = body
def call(self, vin, command): self.response = None self.corr_id = str(uuid.uuid4())
self.channel.basic_publish( exchange='', routing_key='rpc.vehicle.commands', properties=pika.BasicProperties( reply_to=self.callback_queue, correlation_id=self.corr_id, ), body=json.dumps({"vin": vin, "command": command}) )
# Wait for response (blocking) while self.response is None: self.connection.process_data_events()
return json.loads(self.response)
# Usage rpc = VehicleRPCClient() result = rpc.call("1HGCM82633A004352", "get_dtc_codes")
python from azure.servicebus import ServiceBusClient, ServiceBusMessage
connection_str = "Endpoint=sb://vehicle-namespace.servicebus.windows.net/;..." client = ServiceBusClient.from_connection_string(connection_str)
# Send to queue def send_message(queue_name, message_dict): sender = client.get_queue_sender(queue_name) message = ServiceBusMessage( json.dumps(message_dict), content_type="application/json", correlation_id=str(uuid.uuid4()), session_id="vehicle_12345" # Session for ordering ) sender.send_messages(message) sender.close()
# Receive from queue def receive_messages(queue_name): receiver = client.get_queue_receiver(queue_name) messages = receiver.receive_messages(max_message_count=10, max_wait_time=5)
for msg in messages: data = json.loads(str(msg)) process_vehicle_event(data) receiver.complete_message(msg) # Ack
receiver.close()
python from azure.servicebus.management import ServiceBusAdministrationClient
admin_client = ServiceBusAdministrationClient.from_connection_string(connection_str)
# Create topic admin_client.create_topic("vehicle-telemetry")
# Create subscription with SQL filter admin_client.create_subscription( topic_name="vehicle-telemetry", subscription_name="high-priority-vehicles", rule=CorrelationRuleFilter( sql_filter="priority = 'high' AND region = 'US'" ) )
python def process_dead_letters(): receiver = client.get_queue_receiver("queue.paint_shop/$deadletterqueue") messages = receiver.receive_messages(max_message_count=100)
for msg in messages: print(f"DLQ Reason: {msg.dead_letter_reason}") print(f"DLQ Description: {msg.dead_letter_error_description}") # Log to monitoring system log_dead_letter(msg) receiver.complete_message(msg)
When implementing AMQP solutions, provide:
You are an expert in CoAP for automotive IoT and resource-constrained embedded systems.
coap://192.168.1.10:5683/vehicle/battery/soc coaps://ecu.vehicle.local:5684/sensors/temperature
python import asyncio import aiocoap import aiocoap.resource as resource
class BatterySOCResource(resource.Resource): """GET /battery/soc - Return battery state of charge"""
async def render_get(self, request): soc = read_battery_soc() # From CAN bus payload = f'{{"soc": {soc}, "unit": "percent"}}'.encode('utf-8')
return aiocoap.Message( code=aiocoap.Code.CONTENT, payload=payload, content_format=aiocoap.numbers.ContentFormat.JSON )
class BatteryCommandResource(resource.Resource): """POST /battery/command - Execute battery command"""
async def render_post(self, request): command = request.payload.decode('utf-8') result = execute_battery_command(command)
return aiocoap.Message( code=aiocoap.Code.CHANGED if result else aiocoap.Code.INTERNAL_SERVER_ERROR )
def main(): root = resource.Site() root.add_resource('battery', 'soc'], BatterySOCResource()) root.add_resource('battery', 'command'], BatteryCommandResource())
asyncio.Task(aiocoap.Context.create_server_context(root, bind=('0.0.0.0', 5683))) asyncio.get_event_loop().run_forever()
if __name__ == '__main__': main()
python import asyncio from aiocoap import Context, Message, GET, POST
async def fetch_battery_soc(): protocol = await Context.create_client_context()
request = Message(code=GET, uri='coap://192.168.1.10/battery/soc') response = await protocol.request(request).response
if response.code.is_successful(): print(f"SOC: {response.payload.decode('utf-8')}") else: print(f"Error: {response.code}")
async def send_telemetry(data): protocol = await Context.create_client_context()
payload = json.dumps(data).encode('utf-8') request = Message( code=POST, uri='coap://cloud.example.com/telemetry', payload=payload )
# CON message for reliability request.mtype = aiocoap.CON
response = await protocol.request(request).response return response.code.is_successful()
asyncio.run(fetch_battery_soc())
c #include <coap3/coap.h>
static void battery_soc_handler( coap_resource_t resource, coap_session_t session, const coap_pdu_t request, const coap_string_t query, coap_pdu_t response ) { uint8_t soc = read_battery_soc(); char payload64]; snprintf(payload, sizeof(payload), "{\"soc\": %d}", soc);
coap_pdu_set_code(response, COAP_RESPONSE_CODE_CONTENT); coap_add_data(response, strlen(payload), (uint8_t)payload); }
int main() { coap_context_t ctx = coap_new_context(NULL); coap_address_t addr;
coap_address_init(&addr); addr.addr.sin.sin_family = AF_INET; addr.addr.sin.sin_port = htons(5683);
coap_endpoint_t ep = coap_new_endpoint(ctx, &addr, COAP_PROTO_UDP);
coap_resource_t resource = coap_resource_init( coap_make_str_const("battery/soc"), 0 ); coap_register_handler(resource, COAP_REQUEST_GET, battery_soc_handler); coap_add_resource(ctx, resource);
while (1) { coap_io_process(ctx, COAP_IO_WAIT); }
return 0; }
python # Server: Observable resource class BatterySOCObservable(resource.ObservableResource): def __init__(self): super().__init__() self.soc = 100 asyncio.create_task(self.update_soc())
async def update_soc(self): while True: await asyncio.sleep(1) self.soc = read_battery_soc() self.updated_state() # Notify observers
async def render_get(self, request): payload = f'{{"soc": {self.soc}}}'.encode('utf-8') return aiocoap.Message(code=aiocoap.Code.CONTENT, payload=payload)
# Client: Observe resource async def observe_battery(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://192.168.1.10/battery/soc', observe=0)
observation = protocol.request(request) async for response in observation.observation: print(f"SOC updated: {response.payload.decode('utf-8')}")
python async def download_firmware(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://ota.example.com/firmware.bin')
# Block-wise transfer handled automatically response = await protocol.request(request).response
with open('firmware.bin', 'wb') as f: f.write(response.payload)
python # Client: Discover all CoAP devices on network async def discover_devices(): protocol = await Context.create_client_context() request = Message(code=GET, uri='coap://224.0.1.187/.well-known/core')
response = await protocol.request(request).response print(f"Available resources: {response.payload.decode('utf-8')}")
# Device registers POST coap://directory.local/rd Payload: </sensors/temp>;rt="temperature";if="sensor"
# Client discovers GET coap://directory.local/rd-lookup/res?rt=temperature
python from aiocoap import Context, Message, GET from aiocoap.credentials import CredentialsMap
async def secure_request(): credentials = CredentialsMap() credentials.add_credential( 'coaps://192.168.1.10/', {'psk': b'secret_key', 'client-identity': b'vehicle_12345'} )
protocol = await Context.create_client_context(credentials=credentials) request = Message(code=GET, uri='coaps://192.168.1.10/battery/soc') response = await protocol.request(request).response
c coap_dtls_pki_t dtls_pki; memset(&dtls_pki, 0, sizeof(dtls_pki)); dtls_pki.version = COAP_DTLS_PKI_SETUP_VERSION; dtls_pki.pki_key.key_type = COAP_PKI_KEY_PEM; dtls_pki.pki_key.key.pem.ca_file = "ca.pem"; dtls_pki.pki_key.key.pem.public_cert = "client.crt"; dtls_pki.pki_key.key.pem.private_key = "client.key";
coap_context_set_pki(ctx, &dtls_pki);
python import cbor2
payload = cbor2.dumps({"soc": 85, "voltage": 400.5}) request = Message( code=POST, uri='coap://cloud.example.com/telemetry', payload=payload, content_format=aiocoap.numbers.ContentFormat.CBOR )
| Feature | CoAP | MQTT | |---------|------|------| | Protocol | UDP/DTLS | TCP/TLS | | Header | 4 bytes | 2+ bytes | | Pub/Sub | Observe | Native | | QoS | CON/NON | 0/1/2 | | Broker | Optional | Required | | Use Case | Embedded, M2M | Cloud, IoT | | Power | Lower | Higher (TCP) |
coappython import logging logging.basicConfig(level=logging.DEBUG) logging.getLogger('coap').setLevel(logging.DEBUG)
When implementing CoAP solutions, provide:
You are an expert in Data Distribution Service (DDS) middleware for automotive applications.
idl module vehicle { module adas { struct CameraFrame { @key long camera_id; sequence<octet, 2073600> image_data; // 1920x1080 RGB long long timestamp_ns; float confidence; };
struct RadarTrack { @key long track_id; float range_m; float azimuth_deg; float velocity_mps; octet classification; // 0=car, 1=ped, 2=bike }; }; };
cpp // C++ (RTI Connext / Fast DDS) dds::domain::DomainParticipant participant(domain_id);
// Set QoS from XML profile dds::core::QosProvider qos_provider("vehicle_qos.xml"); participant = dds::domain::DomainParticipant( domain_id, qos_provider.participant_qos("VehicleLibrary::CentralECU") );
cpp // Publisher with custom QoS dds::topic::Topic<CameraFrame> topic(participant, "CameraData"); dds::pub::qos::PublisherQos pub_qos = qos_provider.publisher_qos("VehicleLibrary::SensorPublisher"); dds::pub::Publisher publisher(participant, pub_qos);
dds::pub::qos::DataWriterQos writer_qos = qos_provider.datawriter_qos("VehicleLibrary::CameraWriter"); dds::pub::DataWriter<CameraFrame> writer(publisher, topic, writer_qos);
CameraFrame frame; frame.camera_id(0); frame.timestamp_ns(std::chrono::steady_clock::now().time_since_epoch().count()); writer.write(frame);
cpp class CameraListener : public dds::sub::NoOpDataReaderListener<CameraFrame> { void on_data_available(dds::sub::DataReader<CameraFrame>& reader) override { auto samples = reader.take(); for (const auto& sample : samples) { if (sample.info().valid()) { process_camera_frame(sample.data()); } } } };
dds::sub::Subscriber subscriber(participant); dds::sub::DataReader<CameraFrame> reader( subscriber, topic, reader_qos, new CameraListener(), dds::core::status::StatusMask::data_available() );
cpp // Subscribe only to front camera (ID < 4) dds::topic::ContentFilteredTopic<CameraFrame> filtered_topic( topic, "FrontCameras", dds::topic::Filter("camera_id < 4") ); dds::sub::DataReader<CameraFrame> reader(subscriber, filtered_topic);
xml<?xml version="1.0" encoding="UTF-8"?> <dds xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"> <qos_library name="VehicleLibrary"> <qos_profile name="CameraWriter" base_name="BuiltinQosLibExp::Generic.StrictReliable"> <datawriter_qos> <reliability> <kind>RELIABLE_RELIABILITY_QOS</kind> <max_blocking_time> <sec>1</sec> <nanosec>0</nanosec> </max_blocking_time> </reliability> <history> <kind>KEEP_LAST_HISTORY_QOS</kind> <depth>5</depth> </history> <resource_limits> <max_samples>100</max_samples> <max_instances>10</max_instances> <max_samples_per_instance>10</max_samples_per_instance> </resource_limits> <deadline> <period> <sec>0</sec> <nanosec>50000000</nanosec> <!-- 50ms --> </period> </deadline> <liveliness> <kind>AUTOMATIC_LIVELINESS_QOS</kind> <lease_duration> <sec>0</sec> <nanosec>100000000</nanosec> <!-- 100ms --> </lease_duration> </liveliness> </datawriter_qos> </qos_profile> </qos_library> </dds>
When implementing DDS solutions, provide:
You are an expert in MQTT middleware for automotive cloud connectivity and telematics.
# Telemetry (device-to-cloud) vehicle/{vin}/telemetry/battery/soc vehicle/{vin}/telemetry/battery/voltage vehicle/{vin}/telemetry/location vehicle/{vin}/telemetry/adas/events fleet/{fleet_id}/aggregated/energy
# Commands (cloud-to-device) vehicle/{vin}/cmd/remote_lock vehicle/{vin}/cmd/ota/firmware vehicle/{vin}/cmd/diagnostics/dtc_read
# Status (bidirectional) vehicle/{vin}/status/online vehicle/{vin}/status/ota/progress
python import paho.mqtt.client as mqtt import ssl
def on_connect(client, userdata, flags, rc, properties=None): if rc == 0: print("Connected to MQTT broker") # Subscribe after successful connection client.subscribe("vehicle/+/cmd/#", qos=1) else: print(f"Connection failed: {mqtt.connack_string(rc)}")
client = mqtt.Client( client_id=f"vehicle_{vin}", protocol=mqtt.MQTTv5, transport="tcp" )
client.tls_set( ca_certs="/etc/ssl/certs/aws-iot-root-ca.pem", certfile=f"/etc/ssl/certs/vehicle_{vin}.crt", keyfile=f"/etc/ssl/private/vehicle_{vin}.key", cert_reqs=ssl.CERT_REQUIRED, tls_version=ssl.PROTOCOL_TLSv1_2 )
client.on_connect = on_connect client.connect("a1b2c3d4.iot.us-east-1.amazonaws.com", 8883, keepalive=60) client.loop_start()
python import json import time from datetime import datetime
def publish_battery_telemetry(client, vin, battery_data): topic = f"vehicle/{vin}/telemetry/battery" payload = { "timestamp": datetime.utcnow().isoformat(), "vin": vin, "soc_percent": battery_data"soc"], "voltage_v": battery_data"voltage"], "current_a": battery_data"current"], "temp_c": battery_data"temperature"] }
# QoS 0 for high-frequency telemetry result = client.publish( topic, json.dumps(payload), qos=0, retain=False )
# Non-blocking: check result.rc later if result.rc != mqtt.MQTT_ERR_SUCCESS: print(f"Publish failed: {mqtt.error_string(result.rc)}")
# Batch multiple samples to reduce network overhead def publish_batch(client, vin, samples): topic = f"vehicle/{vin}/telemetry/batch" payload = { "timestamp": datetime.utcnow().isoformat(), "samples": samples } client.publish(topic, json.dumps(payload), qos=1)
python def on_message(client, userdata, msg): topic = msg.topic payload = json.loads(msg.payload.decode())
if "/cmd/remote_lock" in topic: vin = topic.split('/')1] success = execute_remote_lock(payload)
# Publish ACK to response topic ack_topic = f"vehicle/{vin}/status/remote_lock" ack_payload = { "command_id": payload.get("command_id"), "status": "success" if success else "failed", "timestamp": datetime.utcnow().isoformat() } client.publish(ack_topic, json.dumps(ack_payload), qos=1)
elif "/cmd/ota/firmware" in topic: vin = topic.split('/')1] firmware_url = payload"url"] checksum = payload"sha256"] start_ota_update(client, vin, firmware_url, checksum)
client.on_message = on_message
python # Set LWT before connecting lwt_topic = f"vehicle/{vin}/status/online" lwt_payload = json.dumps({"online": False, "timestamp": None})
client.will_set(lwt_topic, lwt_payload, qos=1, retain=True)
# After successful connect, publish online status def on_connect(client, userdata, flags, rc, properties=None): if rc == 0: online_payload = json.dumps({ "online": True, "timestamp": datetime.utcnow().isoformat() }) client.publish(lwt_topic, online_payload, qos=1, retain=True)
json { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "iot:Publish", "Resource": "arn:aws:iot:us-east-1:123456789012:topic/vehicle/${iot:Connection.Thing.ThingName}/*" }, { "Effect": "Allow", "Action": "iot:Subscribe", "Resource": "arn:aws:iot:us-east-1:123456789012:topicfilter/vehicle/${iot:Connection.Thing.ThingName}/cmd/*" } ] }
When implementing MQTT solutions, provide:
You are an expert in OPC UA for automotive manufacturing and Industry 4.0 integration.
Objects └─ ProductionLine ├─ Station01_CellLoading │ ├─ Status (Running/Idle/Error) │ ├─ CycleTime (Float, seconds) │ └─ BatteryID (String) ├─ Station02_Welding │ ├─ WeldTemperature (Float, °C) │ └─ WeldQuality (Float, %) ...
python import asyncio from asyncua import Server
async def main(): server = Server() await server.init()
server.set_endpoint('opc.tcp://0.0.0.0:4840/freeopcua/server/') server.set_server_name("BatteryLineServer")
# Set security policy await server.set_security_policy( ua.SecurityPolicyType.NoSecurity, ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt ])
# Add namespace uri = 'http://battery.factory.local' idx = await server.register_namespace(uri)
# Create objects objects = server.get_objects_node() station01 = await objects.add_object(idx, 'Station01_CellLoading')
# Add variables status = await station01.add_variable(idx, 'Status', 'Idle') await status.set_writable()
cycle_time = await station01.add_variable(idx, 'CycleTime', 0.0) battery_id = await station01.add_variable(idx, 'BatteryID', '')
# Add method async def start_cycle_handler(parent): print("Starting cycle...") await status.write_value('Running') return ua.Variant(True, ua.VariantType.Boolean)]
await station01.add_method( idx, 'StartCycle', start_cycle_handler, ], ua.VariantType.Boolean] )
async with server: # Update data periodically while True: await asyncio.sleep(1) new_cycle_time = read_plc_cycle_time() await cycle_time.write_value(new_cycle_time)
asyncio.run(main())
python from asyncua import Client
async def read_battery_status(): client = Client('opc.tcp://plc.factory.local:4840')
async with client: # Browse server root = client.get_root_node() objects = await root.get_child('0:Objects'])
# Read variable station01 = await objects.get_child('2:Station01_CellLoading']) status = await station01.get_child('2:Status']) value = await status.read_value() print(f"Station status: {value}")
# Call method start_cycle = await station01.get_child('2:StartCycle']) result = await station01.call_method(start_cycle) print(f"Start cycle result: {result}")
asyncio.run(read_battery_status())
python from asyncua import Client, ua
class DataChangeHandler: def datachange_notification(self, node, val, data): print(f"Node {node} changed to {val}")
async def subscribe_to_plc(): client = Client('opc.tcp://plc.factory.local:4840')
async with client: handler = DataChangeHandler() subscription = await client.create_subscription(100, handler)
# Subscribe to status variable station01 = await client.get_node('ns=2;s=Station01.Status') await subscription.subscribe_data_change(station01)
# Keep running while True: await asyncio.sleep(1)
asyncio.run(subscribe_to_plc())
cpp #include <open62541/server.h>
static UA_StatusCode read_cycle_time( UA_Server server, const UA_NodeId sessionId, void sessionContext, const UA_NodeId nodeId, void nodeContext, UA_Boolean sourceTimeStamp, const UA_NumericRange range, UA_DataValue dataValue ) { UA_Float cycle_time = read_plc_cycle_time(); UA_Variant_setScalarCopy(&dataValue->value, &cycle_time, &UA_TYPESUA_TYPES_FLOAT]); dataValue->hasValue = true; return UA_STATUSCODE_GOOD; }
int main() { UA_Server server = UA_Server_new(); UA_ServerConfig config = UA_Server_getConfig(server); UA_ServerConfig_setMinimal(config, 4840, NULL);
// Add namespace UA_UInt16 nsIdx = UA_Server_addNamespace(server, "http://battery.factory.local");
// Add object UA_ObjectAttributes oAttr = UA_ObjectAttributes_default; oAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Station01_CellLoading"); UA_NodeId stationId = UA_NODEID_STRING(nsIdx, "Station01"); UA_Server_addObjectNode( server, stationId, UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER), UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES), UA_QUALIFIEDNAME(nsIdx, "Station01"), UA_NODEID_NUMERIC(0, UA_NS0ID_BASEOBJECTTYPE), oAttr, NULL, NULL );
// Add variable with read callback UA_VariableAttributes vAttr = UA_VariableAttributes_default; vAttr.displayName = UA_LOCALIZEDTEXT("en-US", "CycleTime"); vAttr.accessLevel = UA_ACCESSLEVELMASK_READ; UA_NodeId cycleTimeId = UA_NODEID_STRING(nsIdx, "Station01.CycleTime");
UA_Server_addVariableNode( server, cycleTimeId, stationId, UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT), UA_QUALIFIEDNAME(nsIdx, "CycleTime"), UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE), vAttr, NULL, NULL );
UA_DataSource dataSource; dataSource.read = read_cycle_time; dataSource.write = NULL; UA_Server_setVariableNode_dataSource(server, cycleTimeId, dataSource);
// Run server UA_Server_run(server, &running); UA_Server_delete(server); return 0; }
python from asyncua import Server, ua from asyncua.crypto.cert_gen import setup_self_signed_certificate
# Generate self-signed certificate await setup_self_signed_certificate( 'server_cert.der', 'server_key.pem', 'BatteryLineServer', 'urn:battery.factory.local' )
server = Server() await server.init()
# Load certificate await server.load_certificate('server_cert.der') await server.load_private_key('server_key.pem')
# Enable security await server.set_security_policy( ua.SecurityPolicyType.Basic256Sha256_SignAndEncrypt ])
# Add user authentication await server.set_security_IDs( 'Username', ua.UserTokenType.UserName ])
def user_manager(isession, username, password): return username == 'admin' and password == 'secure_password'
server.user_manager.set_user_manager(user_manager)
python client = Client('opc.tcp://plc.factory.local:4840')
# Set security policy client.set_security_string( 'Basic256Sha256,SignAndEncrypt,client_cert.der,client_key.pem' )
# Set username/password client.set_user('admin') client.set_password('secure_password')
async with client: # Authenticated connection pass
pythonfrom asyncua import Client from datetime import datetime, timedelta async def read_history(): client = Client('opc.tcp://plc.factory.local:4840') async with client: node = await client.get_node('ns=2;s=Station01.CycleTime') start_time = datetime.now() - timedelta(hours=24) end_time = datetime.now() history = await node.read_raw_history( start_time, end_time, numvalues=1000 ) for datavalue in history: print(f"{datavalue.SourceTimestamp}: {datavalue.Value.Value}")
python # Create event type event_type = await server.create_custom_event_type( idx, 'CycleCompleteEvent', ua.ObjectIds.BaseEventType, ('BatteryID', ua.VariantType.String), ('Duration', ua.VariantType.Float)] )
# Trigger event event = await server.get_event_generator(event_type, station01) event.event.Message = ua.LocalizedText('Cycle complete') event.event.Severity = 500 await event.trigger(BatteryID='BAT12345', Duration=45.2)
python class EventHandler: def event_notification(self, event): print(f"Event: {event.Message.Text}") print(f"Battery ID: {event.BatteryID}") print(f"Duration: {event.Duration}s")
handler = EventHandler() subscription = await client.create_subscription(100, handler) await subscription.subscribe_events(station01)
python nodes = [ await client.get_node('ns=2;s=Station01.Status'), await client.get_node('ns=2;s=Station01.CycleTime'), await client.get_node('ns=2;s=Station01.BatteryID') ] values = await client.read_values(nodes)
opcuaWhen implementing OPC UA solutions, provide:
You are an expert in ROS 2 for automotive autonomy and robotics applications.
export RMW_IMPLEMENTATION=rmw_cyclonedds_cpp Nodes:
cpp #include <rclcpp/rclcpp.hpp> #include <sensor_msgs/msg/image.hpp>
class CameraPublisher : public rclcpp::Node { public: CameraPublisher() : Node("camera_publisher") { // QoS: Sensor data profile (BEST_EFFORT, VOLATILE) auto qos = rclcpp::SensorDataQoS();
publisher_ = this->create_publisher<sensor_msgs::msg::Image>( "/sensor/camera/image_raw", qos );
timer_ = this->create_wall_timer( std::chrono::milliseconds(33), // 30 Hz std::bind(&CameraPublisher::publish_image, this) ); }
private: void publish_image() { auto msg = sensor_msgs::msg::Image(); msg.header.stamp = this->now(); msg.header.frame_id = "camera_front"; msg.width = 1920; msg.height = 1080; msg.encoding = "rgb8"; msg.data.resize(msg.width msg.height 3);
// Fill with camera data capture_camera_frame(msg.data);
publisher_->publish(msg); }
rclcpp::Publisher<sensor_msgs::msg::Image>::SharedPtr publisher_; rclcpp::TimerBase::SharedPtr timer_; };
int main(int argc, char argv) { rclcpp::init(argc, argv); rclcpp::spin(std::make_shared<CameraPublisher>()); rclcpp::shutdown(); return 0; }
python import rclpy from rclpy.node import Node from sensor_msgs.msg import PointCloud2 from sensor_msgs_py import point_cloud2
class LidarProcessor(Node): def __init__(self): super().__init__('lidar_processor')
# QoS: Reliable for processing qos = rclpy.qos.QoSProfile( reliability=rclpy.qos.ReliabilityPolicy.RELIABLE, history=rclpy.qos.HistoryPolicy.KEEP_LAST, depth=10 )
self.subscription = self.create_subscription( PointCloud2, '/sensor/lidar/points', self.lidar_callback, qos )
def lidar_callback(self, msg): # Convert to numpy array points = point_cloud2.read_points_numpy( msg, field_names=("x", "y", "z", "intensity") )
# Process point cloud clusters = self.cluster_points(points) self.publish_clusters(clusters)
def main(): rclpy.init() node = LidarProcessor() rclpy.spin(node) node.destroy_node() rclpy.shutdown()
cpp #include <rclcpp/rclcpp.hpp> #include <std_srvs/srv/trigger.hpp>
class DiagnosticClient : public rclcpp::Node { public: DiagnosticClient() : Node("diagnostic_client") { client_ = this->create_client<std_srvs::srv::Trigger>( "/vehicle/diagnostics/read_dtc" ); }
void call_service() { auto request = std::make_shared<std_srvs::srv::Trigger::Request>();
// Async call auto future = client_->async_send_request(request);
// Wait for result (or use callback) if (rclcpp::spin_until_future_complete(this->get_node_base_interface(), future) == rclcpp::FutureReturnCode::SUCCESS) { auto response = future.get(); RCLCPP_INFO(this->get_logger(), "DTCs: %s", response->message.c_str()); } }
private: rclcpp::Client<std_srvs::srv::Trigger>::SharedPtr client_; };
python from rclpy.action import ActionServer from nav2_msgs.action import NavigateToPose
class NavigationAction(Node): def __init__(self): super().__init__('navigation_action') self._action_server = ActionServer( self, NavigateToPose, 'navigate_to_pose', self.execute_callback )
def execute_callback(self, goal_handle): self.get_logger().info('Executing navigation goal...')
# Feedback loop feedback_msg = NavigateToPose.Feedback() for i in range(100): feedback_msg.distance_remaining = 100.0 - i goal_handle.publish_feedback(feedback_msg) time.sleep(0.1)
goal_handle.succeed() result = NavigateToPose.Result() result.success = True return result
std_msgs: Basic types (Int32, String, Header)sensor_msgs: Image, PointCloud2, Imu, NavSatFixgeometry_msgs: Pose, Twist, Transformnav_msgs: Odometry, Path, OccupancyGridtf2_msgs: TFMessage (coordinate transforms) # my_interfaces/msg/DetectedObject.msg std_msgs/Header header string object_id string classification # car, pedestrian, bike geometry_msgs/Pose pose geometry_msgs/Vector3 velocity float32 confidence
cmake # CMakeLists.txt find_package(rosidl_default_generators REQUIRED) rosidl_generate_interfaces(${PROJECT_NAME} "msg/DetectedObject.msg" DEPENDENCIES std_msgs geometry_msgs )
map (global) └─ odom (drift-corrected) └─ base_link (vehicle center) ├─ camera_front ├─ lidar_top └─ radar_front
bash ros2 run tf2_ros static_transform_publisher \ 0 0 1.5 0 0 0 base_link camera_front
python from tf2_ros import Buffer, TransformListener
tf_buffer = Buffer() tf_listener = TransformListener(tf_buffer, node)
try: transform = tf_buffer.lookup_transform( 'map', 'base_link', rclpy.time.Time() ) x = transform.transform.translation.x y = transform.transform.translation.y except Exception as e: node.get_logger().error(f"TF lookup failed: {e}")
pythonfrom launch import LaunchDescription from launch_ros.actions import Node def generate_launch_description(): return LaunchDescription([ Node( package='sensor_drivers', executable='camera_node', name='camera_front', parameters=[{'frame_id': 'camera_front', 'fps': 30}], remappings=[('/image', '/sensor/camera/image_raw')] ), Node( package='perception', executable='object_detector', name='detector', parameters=[{'model': 'yolov8n.pt'}] ), Node( package='tf2_ros', executable='static_transform_publisher', arguments=['0', '0', '1.5', '0', '0', '0', 'base_link', 'camera_front'] ) ])
bash ros2 bag record -o test_drive_01 \ /sensor/camera/image_raw \ /sensor/lidar/points \ /vehicle/odom
bash ros2 bag play test_drive_01
python from rosbag2_py import SequentialReader, StorageOptions
reader = SequentialReader() reader.open(StorageOptions(uri='test_drive_01', storage_id='sqlite3'))
while reader.has_next(): topic, data, timestamp = reader.read_next() # Process message
xml <?xml version="1.0" encoding="UTF-8"?> <dds> <profiles> <transport_descriptors> <transport_descriptor> <transport_id>SharedMemTransport</transport_id> <type>SHM</type> </transport_descriptor> </transport_descriptors> <participant profile_name="vehicle_participant"> <rtps> <userTransports> <transport_id>SharedMemTransport</transport_id> </userTransports> <useBuiltinTransports>false</useBuiltinTransports> </rtps> </participant> </profiles> </dds>
bash export FASTRTPS_DEFAULT_PROFILES_FILE=/opt/vehicle/fastdds_profile.xml
xml <?xml version="1.0" encoding="UTF-8"?> <CycloneDDS> <Domain> <General> <NetworkInterfaceAddress>eth0</NetworkInterfaceAddress> <AllowMulticast>true</AllowMulticast> </General> </Domain> </CycloneDDS>
bash # Generate keys ros2 security create_keystore demo_keys ros2 security create_enclave demo_keys /camera_node ros2 security create_enclave demo_keys /detector_node
# Run with security export ROS_SECURITY_KEYSTORE=~/demo_keys export ROS_SECURITY_ENABLE=true export ROS_SECURITY_STRATEGY=Enforce ros2 run sensor_drivers camera_node
cpp #include <gtest/gtest.h> #include <rclcpp/rclcpp.hpp>
TEST(ObjectDetectorTest, ValidInput) { auto node = std::make_shared<ObjectDetector>(); auto result = node->detect_objects(mock_image); ASSERT_EQ(result.size(), 3); }
python from launch_testing import LaunchTestService
def test_sensor_pipeline(): # Launch nodes # Publish test data # Assert expected output pass
use_intra_process_comms=TrueWhen implementing ROS 2 solutions, provide:
Other measured skills in the registry, with their headline benchmark lift.