Module and environment variable validation with Pydantic

1. Overview

The Viam Python SDK is a great way to extend the platform with modules and automate machines with scripts. For each of these tasks, you may encounter the need to validate user-supplied configuration or settings from environment variables. While you can do this manually with simple type() and assert checks, there are more robust, type-safe libraries for handling this logic such as Pydantic.

Pydantic is the most widely used data validation library for Python; used by HuggingFace, FastAPI, Django, and more.

Building off the Working with Python environment variables codelab, you'll learn about using Pydantic to ensure your Viam machines are configured correctly.

Prerequisites

• Familiarity with Python

What You'll Learn

• How to validate modular resource configurations
• How to validate environment variable settings in a Python process script

What You'll Need

• A computer with MacOS, Windows, or Linux
• Python3 installed on your computer
• VS Code installed, or another similar code editor of your choice.

What You'll Build

• a sensor component module with type-safe configuration
• a Python script for automating a Viam machine that uses validated environment variables

2. Create a module project

In this step, you'll build upon the How-to Guide for creating a sensor module with Python to provide robust, type-safe validation to this configuration logic.

Set up the Python sensor module project

1. Create the project directory through the command line or using the code editor of your choice:

   mkdir open-meteo-module
   cd open-meteo-module
   

2. Create the necessary files for the project: requirements.txt, meta.json, run.sh, and main.py, using the command line or code editor of your choice:

   touch requirements.txt meta.json run.sh main.py
   

3. In the meta.json, add the JSON metadata for the module (replacing the "" with the public namespace for your Viam organization or something random if you don't plan on publishing this):

   {
       "$schema": "https://dl.viam.dev/module.schema.json",
       "module_id": "<namespace>:open-meteo",
       "visibility": "public",
       "url": "",
       "description": "Modular sensor component: meteo_pm",
       "models": [\
         {\
           "api": "rdk:component:sensor",\
           "model": "<namespace>:open-meteo:meteo_pm"\
         }\
       ],
       "entrypoint": "./run.sh"
   }
   

4. In the run.sh, add the following shell scripting code for running the module script:

   #!/bin/sh
   cd `dirname $0`
   

Create a virtual environment to run our code

VENV_NAME="venv" PYTHON="$VENV_NAME/bin/python"

ENV_ERROR="This module requires Python >=3.8, pip, and virtualenv to be installed."

if ! python3 -m venv $VENV_NAME >/dev/null 2>&1; then echo "Failed to create virtualenv." if command -v apt-get >/dev/null; then echo "Detected Debian/Ubuntu, attempting to install python3-venv automatically." SUDO="sudo" if ! command -v $SUDO >/dev/null; then SUDO="" fi if ! apt info python3-venv >/dev/null 2>&1; then echo "Package info not found, trying apt update" $SUDO apt -qq update >/dev/null fi $SUDO apt install -qqy python3-venv >/dev/null 2>&1 if ! python3 -m venv $VENV_NAME >/dev/null 2>&1; then echo $ENV_ERROR >&2 exit 1 fi else echo $ENV_ERROR >&2 exit 1 fi fi

remove -U if viam-sdk should not be upgraded whenever possible

-qq suppresses extraneous output from pip

echo "Virtualenv found/created. Installing/upgrading Python packages..." if ! $PYTHON -m pip install -r requirements.txt -Uqq; then exit 1 fi

Be sure to use exec so that termination signals reach the python process,

or handle forwarding termination signals manually

echo "Starting module..." exec $PYTHON main.py $@

5. In the `requirements.txt`, add the dependencies for the project:

```txt
openmeteo-requests
requests-cache
retry-requests
viam-sdk
pydantic

6. In the main.py, add the initial sensor module implementation code (replacing with the same value used in the meta.json):

   import asyncio
   from typing import Any, ClassVar, Mapping, Optional, Sequence
   from typing_extensions import Self
   

from viam.components.sensor import Sensor from viam.logging import getLogger from viam.module.module import Module from viam.proto.app.robot import ComponentConfig from viam.proto.common import ResourceName from viam.resource.base import ResourceBase from viam.resource.easy_resource import EasyResource from viam.resource.types import Model, ModelFamily from viam.utils import SensorReading, struct_to_dict

import openmeteo_requests import requests_cache from retry_requests import retry

class MeteoPm(Sensor, EasyResource): MODEL: ClassVar[Model] = Model( ModelFamily("", "open-meteo"), "meteo_pm" )

latitude: float longitude: float

@classmethod def new( cls, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase] ) -> Self: return super().new(config, dependencies)

@classmethod def validate_config(cls, config: ComponentConfig) -> Sequence[str]: fields = config.attributes.fields if "latitude" in fields: if not fields["latitude"].HasField("number_value"): raise Exception("Latitude must be a float.") if "longitude" in fields: if not fields["longitude"].HasField("number_value"): raise Exception("Longitude must be a float.") return []

def reconfigure( self, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase] ): attrs = struct_to_dict(config.attributes) self.latitude = float(attrs.get("latitude", 45)) self.longitude = float(attrs.get("longitude", -121))

async def get_readings( self, *, extra: Optional[Mapping[str, Any]] = None, timeout: Optional[float] = None, **kwargs ) -> Mapping[ str, SensorReading, ]: cache_session = requests_cache.CachedSession( '.cache', expire_after=3600) retry_session = retry(cache_session, retries=5, backoff_factor=0.2) openmeteo = openmeteo_requests.Client(session=retry_session)

url = "https://air-quality-api.open-meteo.com/v1/air-quality" params = { "latitude": self.latitude, "longitude": self.longitude, "current": ["pm10", "pm2_5"], "timezone": "America/Los_Angeles" } responses = openmeteo.weather_api(url, params=params) response = responses[0] current = response.Current() current_pm10 = current.Variables(0).Value() current_pm2_5 = current.Variables(1).Value() return { "pm2_5": current_pm2_5, "pm10": current_pm10 }

if name == "main": asyncio.run(Module.run_from_registry())

7. Create the Python virtual environment and install the project dependencies:

python3 -m venv .venv source .venv/bin/activate pip install -r requirements.txt

## 3. Add module configuration validation

Now that there is a practical project in place, you'll explore how to improve the validation logic and configuration settings with [Pydantic](https://pydantic.dev/).

Focusing on the `validate_config` class method first, the `config.attributes.fields` are a mapping of [attributes](https://docs.viam.com/configure/#components) set as JSON in the configuration for each component or service. These fields can be checked for existence and verifying the values for those fields are of a specific type. However, as more configuration attributes are added, this will become unwieldy to maintain.

```python
@classmethod
def validate_config(cls, config: ComponentConfig) -> Sequence[str]:
 fields = config.attributes.fields
 if "latitude" in fields:
     if not fields["latitude"].HasField("number_value"):
         raise Exception("Latitude must be a float.")

if "longitude" in fields:
     if not fields["longitude"].HasField("number_value"):
         raise Exception("Longitude must be a float.")
 return []

Pydantic provides a library of base classes and helper methods for creating intuitive validation models that make use of Python's type hints. The same logic in the current validate_config class method can be implemented with the following Pydantic model:

from viam.utils import struct_to_dict
from pydantic import BaseModel

class Config(BaseModel):
    latitude: float = 45
    longitude: float = -121

If there are any invalid values in the config.attributes passed to the Config class, then it will raise a custom ValidationError. The Config class also includes the default value logic included in the reconfigure method of the module, making it immediately useful for that use case as well.

self.config = Config(**struct_to_dict(config.attributes))

Now if there's additional validation logic required for the module configuration, you can add it to the Config class:

from pydantic import BaseModel, Field

class Config(BaseModel):
    latitude: float = Field(ge=-90, le=90, default=45)
    longitude: float = Field(ge=-180, le=180, default=-121)

Now the validation uses numeric constraints to check that the latitude is between -90 and 90 and longitude is between -180 and 180.

The final module class should look like this:

import asyncio
from typing import Any, ClassVar, Mapping, Optional, Sequence
from typing_extensions import Self

import openmeteo_requests
import requests_cache
from retry_requests import retry

from pydantic import BaseModel, Field

class Config(BaseModel):
    latitude: float = Field(ge=-90, le=90, default=45)
    longitude: float = Field(ge=-180, le=180, default=-121)

class MeteoPm(Sensor, EasyResource):
    MODEL: ClassVar[Model] = Model(
        ModelFamily("<namespace>", "open-meteo"), "meteo_pm"
    )

config: Config

@classmethod
    def new(
        cls, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase]
    ) -> Self:
        return super().new(config, dependencies)

@classmethod
    def validate_config(cls, config: ComponentConfig) -> Sequence[str]:
        Config(**struct_to_dict(config.attributes))
        return []

def reconfigure(
        self, config: ComponentConfig, dependencies: Mapping[ResourceName, ResourceBase]
    ):
        self.config = Config(**struct_to_dict(config.attributes))

url = "https://air-quality-api.open-meteo.com/v1/air-quality"
        params = {
            "latitude": self.config.latitude,
            "longitude": self.config.longitude,
            "current": ["pm10", "pm2_5"],
            "timezone": "America/Los_Angeles"
        }
        responses = openmeteo.weather_api(url, params=params)

response = responses[0]
        current = response.Current()
        current_pm10 = current.Variables(0).Value()
        current_pm2_5 = current.Variables(1).Value()

return {
            "pm2_5": current_pm2_5,
            "pm10": current_pm10
        }

if __name__ == "__main__":
    asyncio.run(Module.run_from_registry())

4. Create Python control script

In this section, you'll create a Python project that includes a script based on the code sample displayed in the "Connect" tab of the Viam app. In that script, you'll get the required API key and API key ID to connect to a machine from an environment variable.

1. Create the project directory through the command line or using the code editor of your choice:

   mkdir viam-env-validation && cd viam-env-validation
   

2. Create a Python virtual environment and install the script dependencies:

   python3 -m venv .venv && source .venv/bin/activate && pip install viam-sdk
   

3. Create the main.py script file and add the initial code sample from the Viam app:

   import asyncio
   

from viam.robot.client import RobotClient

async def connect(): opts = RobotClient.Options.with_api_key( api_key='', api_key_id='' ) return await RobotClient.at_address('my-machine.aabahtjw04.viam.cloud', opts)

async def main(): machine = await connect()

print('Resources:') print(machine.resource_names)

await machine.close()

if name == 'main': asyncio.run(main())

4. Update the code sample to use environment variables to set the API key and API key ID, as demonstrated in the [Working with Python environment variables codelab](https://codelabs.viam.com/guide/environment-variables/index.html):

```python
import asyncio
import os

from viam.robot.client import RobotClient

ROBOT_API_KEY = os.getenv('ROBOT_API_KEY')
ROBOT_API_KEY_ID = os.getenv('ROBOT_API_KEY_ID')

async def connect():
       opts = RobotClient.Options.with_api_key(
           api_key=ROBOT_API_KEY,
           api_key_id=ROBOT_API_KEY_ID,
       )
       return await RobotClient.at_address('my-machine.aabahtjw04.viam.cloud', opts)

5. Add environment variable validation

Pydantic provides a separate package for handling settings management called pydantic-settings, which can load and validate configuration values from environment variables, using the python-dotenv package internally.

1. Start by installing the pydantic-settings dependency in the virtual environment:

   pip install pydantic-settings
   

2. In the main.py script, create the Settings class to validate the expected environment variables:

   from pydantic_settings import BaseSettings, SettingsConfigDict
   

class Settings(BaseSettings): api_key: str = "" api_key_id: str = ""

model_config = SettingsConfigDict(env_prefix="robot_", env_file=".env")

Similar to the Pydantic `BaseModel`, the `BaseSettings` class references the Python type hints to automatically validate the values set for the environment variables. The `model_config` property allows you to configure the default behavior for managing environment variable parsing and validation, including [case-sensitivity](https://docs.pydantic.dev/latest/concepts/pydantic_settings/#case-sensitivity), referencing [.env files](https://docs.pydantic.dev/latest/concepts/pydantic_settings/#dotenv-env-support), and a [shared prefix for variable names](https://docs.pydantic.dev/latest/concepts/pydantic_settings/#environment-variable-names). If there is a `.env` file available, the class will use those values after checking for global values in the runtime environment.

3. In the script, you can use the `Settings` class in the `connect()` method:

```python
async def connect():
       settings = Settings()
       opts = RobotClient.Options.with_api_key(
           api_key=settings.api_key,
           api_key_id=settings.api_key_id,
       )
       return await RobotClient.at_address('my-machine.aabahtjw04.viam.cloud', opts)

If the script is run in an environment without the proper settings available, it will raise a helpful ValidationError to inform you or whoever is using it!

6. Conclusion And Resources

Validation is not always the first thing that comes to mind when creating hardware automations, however they are a useful part of building robust and scalable systems for you and your team as you maintain your projects in the long term. Now you're equipped with the knowledge to make this happen!

What You Learned