first commit

basicdiscovery.py

+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+
+import time
+import json
+from awscrt import io, http
+from awscrt.mqtt import QoS
+from awsiot.greengrass_discovery import DiscoveryClient
+from awsiot import mqtt_connection_builder
+
+from utils.command_line_utils import CommandLineUtils
+
+allowed_actions = ['both', 'publish', 'subscribe']
+
+# cmdData is the arguments/input from the command line placed into a single struct for
+# use in this sample. This handles all of the command line parsing, validating, etc.
+# See the Utils/CommandLineUtils for more information.
+cmdData = CommandLineUtils.parse_sample_input_basic_discovery()
+
+tls_options = io.TlsContextOptions.create_client_with_mtls_from_path(cmdData.input_cert, cmdData.input_key)
+if (cmdData.input_ca is not None):
+    tls_options.override_default_trust_store_from_path(None, cmdData.input_ca)
+tls_context = io.ClientTlsContext(tls_options)
+
+socket_options = io.SocketOptions()
+
+proxy_options = None
+if cmdData.input_proxy_host is not None and cmdData.input_proxy_port != 0:
+    proxy_options = http.HttpProxyOptions(cmdData.input_proxy_host, cmdData.input_proxy_port)
+
+print('Performing greengrass discovery...')
+discovery_client = DiscoveryClient(
+    io.ClientBootstrap.get_or_create_static_default(),
+    socket_options,
+    tls_context,
+    cmdData.input_signing_region, None, proxy_options)
+resp_future = discovery_client.discover(cmdData.input_thing_name)
+discover_response = resp_future.result()
+
+if (cmdData.input_is_ci):
+    print("Received a greengrass discovery result! Not showing result in CI for possible data sensitivity.")
+else:
+    print(discover_response)
+
+if (cmdData.input_print_discovery_resp_only):
+    exit(0)
+
+
+def on_connection_interupted(connection, error, **kwargs):
+    print('connection interrupted with error {}'.format(error))
+
+
+def on_connection_resumed(connection, return_code, session_present, **kwargs):
+    print('connection resumed with return code {}, session present {}'.format(return_code, session_present))
+
+
+# Try IoT endpoints until we find one that works
+def try_iot_endpoints():
+    for gg_group in discover_response.gg_groups:
+        for gg_core in gg_group.cores:
+            for connectivity_info in gg_core.connectivity:
+                try:
+                    print(
+                        f"Trying core {gg_core.thing_arn} at host {connectivity_info.host_address} port {connectivity_info.port}")
+                    mqtt_connection = mqtt_connection_builder.mtls_from_path(
+                        endpoint=connectivity_info.host_address,
+                        port=connectivity_info.port,
+                        cert_filepath=cmdData.input_cert,
+                        pri_key_filepath=cmdData.input_key,
+                        ca_bytes=gg_group.certificate_authorities[0].encode('utf-8'),
+                        on_connection_interrupted=on_connection_interupted,
+                        on_connection_resumed=on_connection_resumed,
+                        client_id=cmdData.input_thing_name,
+                        clean_session=False,
+                        keep_alive_secs=30)
+
+                    connect_future = mqtt_connection.connect()
+                    connect_future.result()
+                    print('Connected!')
+                    return mqtt_connection
+
+                except Exception as e:
+                    print('Connection failed with exception {}'.format(e))
+                    continue
+
+    exit('All connection attempts failed')
+
+
+mqtt_connection = try_iot_endpoints()
+
+if cmdData.input_mode == 'both' or cmdData.input_mode == 'subscribe':
+    def on_publish(topic, payload, dup, qos, retain, **kwargs):
+        print('Publish received on topic {}'.format(topic))
+        print(payload)
+    subscribe_future, _ = mqtt_connection.subscribe(cmdData.input_topic, QoS.AT_MOST_ONCE, on_publish)
+    subscribe_result = subscribe_future.result()
+
+loop_count = 0
+while loop_count < cmdData.input_max_pub_ops:
+    if cmdData.input_mode == 'both' or cmdData.input_mode == 'publish':
+        message = {}
+        message['message'] = cmdData.input_message
+        message['sequence'] = loop_count
+        messageJson = json.dumps(message)
+        pub_future, _ = mqtt_connection.publish(cmdData.input_topic, messageJson, QoS.AT_LEAST_ONCE)
+        publish_completion_data = pub_future.result()
+        print('Published topic {}: {} (puback reason: {})\n'.format(cmdData.input_topic, messageJson, repr(publish_completion_data.puback.reason_code)))
+
+        loop_count += 1
+    time.sleep(1)
\ No newline at end of file

create.py

+import boto3
+import json
+import random
+import string
+import os
+
+# Initialize clients
+thingClient = boto3.client('iot')
+defaultPolicyName = 'GGTest_Group_Core-policy'  # Update with your policy name
+thingGroupName = 'VehicleEmissionsGroup'       # Custom group name
+
+# Directory structure configuration
+root_dir = "iot_resources"
+certs_dir = os.path.join(root_dir, "certificates")
+keys_dir = os.path.join(root_dir, "keys")
+data_dir = os.path.join(root_dir, "data")
+
+# Create directories if they don't exist
+os.makedirs(certs_dir, exist_ok=True)
+os.makedirs(keys_dir, exist_ok=True)
+os.makedirs(data_dir, exist_ok=True)
+
+def randomThingName(prefix="vehicle_"):
+    """Generate a consistent Thing name with numeric suffix."""
+    return f"{prefix}{random.randint(0, 99999):05d}"
+
+def createThingGroup(groupName):
+    """Create a Thing Group in AWS IoT if it doesn't exist."""
+    try:
+        thingClient.create_thing_group(thingGroupName=groupName)
+        print(f"✅ Thing Group '{groupName}' created.")
+    except thingClient.exceptions.ResourceAlreadyExistsException:
+        print(f"⚠️ Thing Group '{groupName}' already exists.")
+
+def createThingWithCertAndAddToGroup(device_id):
+    """Create a Thing with consistent naming for MQTT client integration."""
+    thingName = f"device_{device_id}"
+    print(f"🚀 Creating thing: {thingName}")
+
+    # Create Thing in AWS IoT
+    thing = thingClient.create_thing(thingName=thingName)
+    thingArn = thing['thingArn']
+
+    # Create keys and certificates for the Thing
+    cert = thingClient.create_keys_and_certificate(setAsActive=True)
+    certArn = cert['certificateArn']
+    certId = cert['certificateId']
+
+    # Save certs with consistent naming pattern
+    cert_files = {
+        'cert': os.path.join(certs_dir, f"device_{device_id}.pem"),
+        'pubkey': os.path.join(keys_dir, f"device_{device_id}.public.pem"),
+        'privkey': os.path.join(keys_dir, f"device_{device_id}.private.pem")
+    }
+    
+    with open(cert_files['cert'], 'w') as f:
+        f.write(cert['certificatePem'])
+    with open(cert_files['pubkey'], 'w') as f:
+        f.write(cert['keyPair']['PublicKey'])
+    with open(cert_files['privkey'], 'w') as f:
+        f.write(cert['keyPair']['PrivateKey'])
+
+    # Attach policy to the certificate
+    thingClient.attach_policy(
+        policyName=defaultPolicyName,
+        target=certArn
+    )
+
+    # Attach the certificate to the Thing
+    thingClient.attach_thing_principal(
+        thingName=thingName,
+        principal=certArn
+    )
+
+    # Add the Thing to the Thing Group
+    thingClient.add_thing_to_thing_group(
+        thingGroupName=thingGroupName,
+        thingName=thingName
+    )
+
+    print(f"✅ {thingName} registered with certificate {certId[:6]}...")
+    print(f"   Certificates saved to:\n   - {cert_files['cert']}\n   - {cert_files['privkey']}\n")
+
+    return cert_files
+
+# Create the Thing Group
+createThingGroup(thingGroupName)
+
+# Create multiple Things with sequential IDs
+num_devices = 10  # Adjust as needed
+for device_id in range(num_devices):
+    createThingWithCertAndAddToGroup(device_id)
+
+print(f"✅ All {num_devices} devices created and added to group '{thingGroupName}'")
+print(f"Certificate files are in: {certs_dir}")
+print(f"Key files are in: {keys_dir}")
\ No newline at end of file

gdk-config.json

+{
+  "component" :{
+    "greengrasscomp": {
+      "author": "cozykace",
+      "version": "1.0.6",
+      "build": {
+        "build_system" :"zip"
+      },
+      "publish": {
+        "bucket": "greengrasscomp-artefacts",
+        "region": "us-east-1"
+      }
+    }
+  },
+  "gdk_version": "1.0.0"
+}
\ No newline at end of file

lab4em.py

+# Import SDK packages
+from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
+import time
+import random
+import json
+import pandas as pd
+import numpy as np
+import os
+
+
+#TODO 1: modify the following parameters
+#Starting and end index, modify this
+device_st = 0
+device_end = 5
+
+#Path to the dataset, modify this
+data_path = os.path.join("iot_resources", "data", "vehicle{}.csv")
+certificate_formatter = os.path.join("iot_resources", "certificates", "device_{}.pem")
+key_formatter = os.path.join("iot_resources", "keys", "device_{}.private.pem")
+root_ca_path = "AmazonRootCA1.pem"
+
+
+iot_endpoint = "a1u83sicayovh8-ats.iot.us-east-1.amazonaws.com"  
+topic = "vehicles/device7/data"
+
+
+    
+class MQTTClient:
+    def __init__(self, device_id, cert, key):
+        # For certificate based connection
+        self.device_id = str(device_id)
+        self.client = AWSIoTMQTTClient(self.device_id)
+        self.client.configureEndpoint(iot_endpoint, 8883)
+        self.client.configureCredentials(root_ca_path, key, cert)
+        #TODO 2: modify your broker address
+        self.client.configureOfflinePublishQueueing(-1)
+        self.client.configureDrainingFrequency(2)
+        self.client.configureConnectDisconnectTimeout(10)
+        self.client.configureMQTTOperationTimeout(5)
+        self.client.onMessage = self.customOnMessage
+        
+
+    def customOnMessage(self,message):
+        #TODO 3: fill in the function to show your received message
+        print(f"[Device {self.device_id}] Received message: {message.payload} on topic {message.topic}")
+
+
+    # Suback callback
+    def customSubackCallback(self,mid, data):
+        #You don't need to write anything here
+        pass
+
+
+    # Puback callback
+    def customPubackCallback(self,mid):
+        #You don't need to write anything here
+        pass
+
+
+    def publish(self, topic="vehicle/emission/data"):
+    # Load the vehicle's emission data
+        df = pd.read_csv(data_path.format(self.device_id))
+        for index, row in df.iterrows():
+            # Create a JSON payload from the row data
+            payload = json.dumps(row.to_dict())
+            
+            # Publish the payload to the specified topic
+            print(f"Publishing: {payload} to {topic}")
+            self.client.publishAsync(topic, payload, 0, ackCallback=self.customPubackCallback)
+            
+            # Sleep to simulate real-time data publishing
+            time.sleep(1)
+
+    def publish_custom(self, topic, message_dict):
+        payload = json.dumps(message_dict)
+        print(f"[Device {self.device_id}] Publishing custom message: {payload} to {topic}")
+        self.client.publishAsync(topic, payload, 0, ackCallback=self.customPubackCallback)
+
+            
+
+
+
+print("Loading vehicle data...")
+data = []
+for i in range(device_st, device_end):
+    # Assuming each device has a corresponding CSV file
+    try:
+        a = pd.read_csv(data_path.format(i))
+        data.append(a)
+    except FileNotFoundError:
+        print(f"File {data_path.format(i)} not found.")
+        data.append(None) 
+
+print("Initializing MQTTClients...")
+clients = []
+for device_id in range(device_st, device_end):
+    cert_path = certificate_formatter.format(device_id)
+    key_path = key_formatter.format(device_id)
+    if not os.path.exists(cert_path) or not os.path.exists(key_path):
+        print(f"missing cert/key.")
+        continue
+
+    client = MQTTClient(device_id, cert_path, key_path)
+    try:
+        client.client.connect()
+        print(f"Connected client {device_id}")
+        clients.append(client)
+    except Exception as e:
+        print(f"Failed to connect {device_id}: {e}")
+
+ 
+while True:
+    print("Choose an action:\n  s - send data from CSV\n  c - send custom message\n  d - disconnect\n  any other key - exit")
+    x = input().strip().lower()
+
+    if x == "s":
+        for client in clients:
+            client.publish()
+
+    elif x == "c":
+        topic_input = input("Enter topic (or press enter to use default): ").strip()
+        topic_to_use = topic_input if topic_input else topic
+
+        key = input("Enter key: ")
+        value = input("Enter value: ")
+
+        custom_message = {key: value}
+
+        for client in clients:
+            client.publish_custom(topic_to_use, custom_message)
+
+    elif x == "d":
+        for client in clients:
+            client.client.disconnect()
+        print("clients disconnected.")
+        break
+
+    else:
+        print("Exiting.")
+        break
+
+    time.sleep(3)
+
+
+
+
+

main.py

+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: MIT-0.
+
+'''
+main.py:
+
+AWS Greengrass component demo for the AWS Greengrass PubSub SDK.
+
+Simple Time publish and IPC / MQTT message logger
+'''
+
+__version__ = "0.0.1"
+__status__ = "Development"
+__copyright__ = "Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved."
+__author__ = "Dean Colcott <https://www.linkedin.com/in/deancolcott/>"
+
+import sys
+import json
+import time
+import logging
+from datetime import datetime
+
+# AWS Greengrass PubSub Componnht SDK Imports.
+from awsgreengrasspubsubsdk.message_formatter import PubSubMessageFormatter
+from awsgreengrasspubsubsdk.pubsub_client import AwsGreengrassPubSubSdkClient
+
+# Example user defined message handling classes
+from pubsub_message_handlers.my_system_message_handler import MySystemMessageHandler
+from pubsub_message_handlers.my_sensor_message_handler import MySensorMessageHandler
+
+# Config the logger.
+log = logging.getLogger(__name__)
+logging.basicConfig(format="[%(name)s.%(funcName)s():%(lineno)d] - [%(levelname)s] - %(message)s", 
+                    stream=sys.stdout, 
+                    level=logging.DEBUG)
+
+class MyAwsGreengrassV2Component():
+
+    def __init__(self, ggv2_component_config):
+        '''
+            Initialises the AWS Greengrass V2 custom component including IPC and MQTT PubSub SDK Client.
+        '''
+        
+        log.info('Initialising AwsGreengrassV2 PubSub SDK Component Example.....')
+        
+        pubsub_base_topic = ggv2_component_config['base-pubsub-topic']
+        log.info('PubSub Base Topic: {}'.format(pubsub_base_topic))
+        
+        ipc_subscribe_topics = ggv2_component_config['ipc-subscribe-topics']
+        log.info('IPC Custom Subscribe Topics: {}'.format(ipc_subscribe_topics))
+        
+        mqtt_subscribe_topics = ggv2_component_config['mqtt-subscribe-topics']
+        log.info('MQTT Custom Subscribe Topics: {}'.format(mqtt_subscribe_topics))
+        
+        # Initilise the PubSub Message Formatter and PubSub Client SDK. 
+        self.message_formatter = PubSubMessageFormatter()
+        self.pubsub_client = AwsGreengrassPubSubSdkClient(pubsub_base_topic, self.default_message_handler )
+
+        # Take handles to SDK publish message and error just for easier access.
+        self.publish_message = self.pubsub_client.publish_message
+        self.publish_error = self.pubsub_client.publish_error
+
+        ##################################################################################
+        # Initilise and register example user defined message handler classes for message routing.
+        ##################################################################################
+        # Message handler for System type requests
+        log.info('Initialising and registering MySystemMessageHandler Class')
+        self.my_system_message_handler = MySystemMessageHandler(self.publish_message, self.publish_error, self.message_formatter)
+        self.pubsub_client.register_message_handler(self.my_system_message_handler)
+        log.info('Initialising and registering MySystemMessageHandlerClass - Complete')
+
+         # Message handler for connected sensor type requests
+        log.info('Initialising and registering MySensorMessageHandler Class')
+        self.my_sensor_message_handler = MySensorMessageHandler(self.publish_message, self.publish_error, self.message_formatter)
+        self.pubsub_client.register_message_handler(self.my_sensor_message_handler)
+        log.info('Initialising and registering MySensorMessageHandler Class - Complete')
+
+        # Activate the MQTT and IPC PubSub Channels, can active one, either or both.
+        log.info('Activating AWS Greengrass PubSub SDK IPC and MQTT Protocols')
+        self.pubsub_client.activate_ipc_pubsub()
+        self.pubsub_client.activate_mqtt_pubsub()
+        log.info('Activating AWS Greengrass PubSub SDK IPC and MQTT Protocols - Complete')
+        
+        # Subscribe to any user defined IPC topics
+        log.info('Subscribing to user defined IPC Protocols')
+        for topic in ipc_subscribe_topics:
+            self.pubsub_client.subscribe_to_topic('ipc', topic)
+        log.info('Subscribing to user defined IPC Protocols - Complete')
+
+        # Subscribe to any user defined MQTT topics
+        log.info('Subscribing to user defined MQTT Protocols')
+        for topic in mqtt_subscribe_topics:
+            self.pubsub_client.subscribe_to_topic('mqtt', topic)
+        log.info('Subscribing to user defined MQTT Protocols - Complete')
+        
+        log.info('Initilising AwsGreengrassV2 PubSub SDK Component Example Complete.')
+
+    ##################################################
+    # Main service / process application logic
+    ##################################################
+    def service_loop(self):
+        '''
+        Holds the process up while handling event-driven PubSub triggers.
+        Put synchronous application logic here or have the component completely event driven.
+        
+        This example periodically publishes the local time to IPC and MQTT in a well formatted message.
+        '''
+        
+        while True:
+            try:
+                # Build and publish a well formatted message displaying local time to IPC and MQTT
+                receive_route = "local_time_update"
+                my_message = { "local-time" : datetime.now().strftime("%d-%b-%Y %H:%M:%S")  }
+                sdk_format_msg = self.message_formatter.get_message(route=receive_route, message=my_message)
+                log.info('Publishing message: {}'.format(sdk_format_msg))
+                self.publish_message('ipc_mqtt', sdk_format_msg)
+            
+            except Exception as err:
+                # Publish error to IPC and MQTT on default error topic. 
+                protocol = 'ipc_mqtt'
+                err_msg = 'Exception in main process loop: {}'.format(err)
+                self.publish_error(protocol, err_msg)
+                
+            finally:
+                time.sleep(10)
+
+    def default_message_handler(self, protocol, topic, message_id, status, route, message):
+        '''
+            This default message handler function is a route of last resort to handle 
+            PubSub messages received by the SDK with a route value that does not 
+            match any functions in the registered message_handler classes.
+
+            In this example, we generate and publish an error message to both IPC and MQTT.
+            You could instead handle this locally or just sliently discard messages here depending 
+            on the application.
+        '''
+
+        # Publish error to IPC and MQTT on default error topic, will log locally as an error as well. 
+        err_msg = 'Received message to unknown route / message handler: {} - message: {}'.format(route, message)
+        self.publish_error(protocol, err_msg)
+
+if __name__ == "__main__":
+
+    try:
+
+        # Parse config from component recipe into sys.argv[1] 
+        ggv2_component_config = sys.argv[1]
+        ggv2_component_config = json.loads(ggv2_component_config)
+        
+        log.info('GG PubSub SDK Config: {}'.format(ggv2_component_config))
+
+        # Create the component class with the parsed Greengrass recipe config.
+        ggv2_component = MyAwsGreengrassV2Component(ggv2_component_config)
+        
+        # Start the main process loop to hold up the process.
+        ggv2_component.service_loop()
+
+    except Exception as err:
+        log.error('Exception occurred initialising component. ERROR MESSAGE: {}'.format(err))

process_emission.py

+import os
+import csv
+import json
+import logging
+import sys
+import time
+from typing import Dict, List, Union, Optional
+
+# Get absolute path to data folder - CORRECTED PATH
+current_dir = os.path.dirname(os.path.abspath(__file__))
+DATA_PATH = os.path.join(current_dir, '.aws_iot_resources', 'data')
+
+try:
+    import greengrasssdk
+    client = greengrasssdk.client("iot-data")
+    GG_ENV = True
+except ImportError:
+    GG_ENV = False
+    print("Running in local test mode (no Greengrass SDK)")
+
+logger = logging.getLogger(__name__)
+logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
+
+def load_vehicle_data(vehicle_id: str) -> List[Dict]:
+    """Load and process vehicle data from CSV files"""
+    vehicle_data = []
+    file_path = os.path.join(DATA_PATH, f"{vehicle_id}.csv")
+    
+    if not os.path.exists(file_path):
+        logger.error(f"Data file not found: {file_path}")
+        return vehicle_data
+
+    try:
+        with open(file_path, mode='r', encoding='utf-8') as csvfile:
+            reader = csv.DictReader(csvfile)
+            for row in reader:
+                # Convert numeric values and clean the data
+                processed = {}
+                for key, value in row.items():
+                    try:
+                        # Handle empty strings and convert numbers
+                        if value.strip() == '':
+                            processed[key] = 0.0
+                        else:
+                            processed[key] = float(value)
+                    except (ValueError, AttributeError):
+                        processed[key] = value.strip()
+                vehicle_data.append(processed)
+                
+        logger.info(f"Loaded {len(vehicle_data)} records from {vehicle_id}.csv")
+        
+    except Exception as e:
+        logger.error(f"Error processing {file_path}: {str(e)}")
+    
+    return vehicle_data
+
+def process_emission_data(records: Union[str, Dict, List]) -> Optional[Dict]:
+    """Process vehicle data to find maximum CO2 emission"""
+    max_co2 = 0.0
+    vehicle_id = None
+    valid_records = 0
+
+    # Handle vehicle_id string input
+    if isinstance(records, str):
+        records = load_vehicle_data(records)
+
+    # Normalize input format
+    if not isinstance(records, list):
+        records = [records] if isinstance(records, dict) else []
+
+    if not records:
+        logger.error("No valid records found")
+        return None
+
+    for record in records:
+        try:
+            # Get CO2 value - using the exact field name from your CSV
+            co2_val = record['vehicle_CO2']  # This matches your CSV header
+            
+            # Get vehicle ID - also matches your CSV
+            current_vehicle_id = record['vehicle_id']
+            
+            # Initialize or verify consistent vehicle_id
+            if vehicle_id is None:
+                vehicle_id = current_vehicle_id
+            elif vehicle_id != current_vehicle_id:
+                logger.warning(f"Ignoring record for different vehicle: {current_vehicle_id}")
+                continue
+
+            # Track maximum CO2 value
+            if co2_val > max_co2:
+                max_co2 = co2_val
+                
+            valid_records += 1
+                
+        except (KeyError, TypeError, ValueError) as e:
+            logger.warning(f"Skipping invalid record: {str(e)}")
+            continue
+
+    if not valid_records:
+        logger.error("No valid CO2 measurements found")
+        return None
+
+    logger.info(f"Processed {valid_records} records for {vehicle_id}, max CO2: {max_co2}")
+    
+    return {
+        "vehicle_id": vehicle_id,
+        "max_co2": max_co2,
+        "timestamp": int(time.time()),
+        "unit": "ppm",
+        "records_processed": valid_records
+    }
+
+def lambda_handler(event, context=None):
+    """Handler for AWS Lambda/Greengrass"""
+    result = process_emission_data(event)
+    if result and GG_ENV:
+        publish_results(result)
+    return result
+
+def publish_results(result: Dict) -> bool:
+    """Publish results to IoT Core"""
+    try:
+        topic = f"vehicles/{result['vehicle_id']}/emission/results"
+        client.publish(
+            topic=topic,
+            payload=json.dumps(result)
+        )
+        logger.info(f"Published to {topic}: {result}")
+        return True
+    except Exception as e:
+        logger.error(f"Publish failed: {str(e)}")
+        return False
+
+if __name__ == "__main__":
+    print("\nVehicle Emission Processor")
+    print(f"Data directory: {DATA_PATH}")
+    
+    try:
+        # Check if data directory exists
+        if not os.path.exists(DATA_PATH):
+            raise FileNotFoundError(f"Directory not found: {DATA_PATH}")
+        
+        # Find all vehicle CSV files
+        vehicle_files = [
+            f for f in os.listdir(DATA_PATH) 
+            if f.startswith('vehicle') and f.endswith('.csv')
+        ]
+        
+        if not vehicle_files:
+            raise FileNotFoundError("No vehicle CSV files found")
+        
+        print(f"\nFound {len(vehicle_files)} vehicle files:")
+        for f in sorted(vehicle_files):
+            print(f"- {f}")
+        
+        # Process each vehicle file
+        for csv_file in sorted(vehicle_files):
+            vehicle_id = os.path.splitext(csv_file)[0]
+            print(f"\nProcessing {vehicle_id}...")
+            
+            result = process_emission_data(vehicle_id)
+            if result:
+                print(f"Results:")
+                print(f"- Vehicle ID: {result['vehicle_id']}")
+                print(f"- Max CO2: {result['max_co2']} ppm")
+                print(f"- Records processed: {result['records_processed']}")
+                
+                if GG_ENV:
+                    if publish_results(result):
+                        print("Published to IoT Core")
+            else:
+                print(f"Failed to process {vehicle_id}")
+                
+    except Exception as e:
+        print(f"\nERROR: {str(e)}")
+        print("\nTroubleshooting steps:")
+        print(f"1. Verify the folder exists: {DATA_PATH}")
+        print("2. Ensure it contains vehicle CSV files (vehicle0.csv, vehicle1.csv, etc.)")
+        print("3. Check file permissions")
+        print("4. Confirm CSV files have 'vehicle_CO2' and 'vehicle_id' columns")
+        sys.exit(1)
\ No newline at end of file

recipe.json

+{
+  "RecipeFormatVersion": "2020-01-25",
+  "ComponentName": "greengrasscomp",
+  "ComponentVersion": "1.0.6",
+  "ComponentDescription": "AWS Greengrass component.",
+  "ComponentPublisher": "cozykace",
+  "ComponentConfiguration": {
+    "DefaultConfiguration": {
+      "GGV2PubSubSdkConfig": {
+        "base-pubsub-topic" : "greengrasscomp",
+        "ipc-subscribe-topics" : ["vehicles/+/emission/data"],
+        "mqtt-subscribe-topics" : ["vehicles/+/emission/results"]
+      },
+      "accessControl": {
+        "aws.greengrass.ipc.pubsub": {
+          "greengrasscomp:publish:1": {
+            "policyDescription": "Allows access to publish to the component IPC topics.",
+            "operations": [
+              "aws.greengrass#PublishToTopic",
+              "aws.greengrass#SubscribeToTopic"
+            ],
+            "resources": [
+              "vehicles/+/emission/data",
+              "vehicles/+/emission/results"
+            ]
+          },
+          "greengrasscomp:subscribe:1": {
+            "policyDescription": "Allows access to subscribe to the component IPC topics.",
+            "operations": [
+              "aws.greengrass#SubscribeToTopic"
+            ],
+            "resources": [
+              "*"
+            ]
+          }
+        },
+        "aws.greengrass.ipc.mqttproxy": {
+          "greengrasscomp:publish:1": {
+            "policyDescription": "Allows access to publish to the component MQTT topics.",
+            "operations": [
+              "aws.greengrass#PublishToIoTCore",
+              "aws.greengrass#SubscribeToIoTCore"
+            ],
+            "resources": [
+              "vehicles/+/emission/data",
+              "vehicles/+/emission/results"
+            ]
+          },
+          "greengrasscomp:subscribe:1": {
+            "policyDescription": "Allows access to subscribe to the component MQTT topics.",
+            "operations": [
+              "aws.greengrass#SubscribeToIoTCore"
+            ],
+            "resources": [
+              "*"
+            ]
+          }
+        }
+      }
+    }
+  },
+  "Manifests": [
+    {
+      "Name": "Linux",
+      "Platform": {
+        "os": "linux"
+      },
+      "Artifacts": [
+        {
+          "URI": "s3://aws-greengrass-components/src.zip",
+          "Unarchive": "ZIP"
+        }
+      ],
+      "Lifecycle": {
+        "Install" : {
+          "Timeout" : 300,
+          "Script" : "python3 -m pip install awsgreengrasspubsubsdk"
+        },
+        "Run": {
+          "Script": "python3 -u {artifacts:decompressedPath}/src/main.py '{configuration:/GGV2PubSubSdkConfig}'",
+          "RequiresPrivilege": "false"
+        }
+      }
+    }
+  ]
+}