Write a logic module

Your machine has resources – sensors, motors, cameras – that work individually. A logic module makes them work together. It runs as a service alongside viam-server, declares dependencies on the resources it needs, and implements whatever control, monitoring, or coordination logic your application requires.

Use a logic module when you need your machine to make decisions based on what it senses: trigger actions when readings cross a threshold, coordinate multiple components to accomplish a task, aggregate data from several sources, or run any continuous process that reads from some resources and acts on others.

Driver modules and logic modules

A driver module wraps hardware – it implements a component API like sensor or motor so that viam-server can talk to a specific piece of hardware.

A logic module (this page) orchestrates existing resources – it reads from sensors, commands motors, and makes decisions. It typically implements a service API.

Both are modules. The difference is what they do, not how they’re built. The lifecycle, config validation, dependency, and deployment patterns are the same.

The generic service API is a minimal service interface that exposes DoCommand and GetStatus. If you are writing custom control logic for a robotics application, this is the API you want. Pick a more specific typed API like vision or motion only when your module’s work fits one.

For background on module lifecycle, dependencies, and background tasks, see the overview.

Steps

When writing a logic module, follow the steps outlined below. To illustrate each step we’ll use a temperature alert monitor as a worked example. It watches one or more sensors, compares their readings against configurable thresholds, and maintains a list of active alerts that your application code can query.

1. Generate a generic service module

Before you run the generator, install the Viam CLI and log in with viam login.

The generator prompts for your organization’s public namespace. If you have not set one yet, click the organization dropdown at the upper right of the Viam app, select Settings, then Set a public namespace. You can also enter your Org ID at the prompt instead.

Run the Viam CLI generator:

viam module generate

The generator creates a new directory named after your module (for example, alert-monitor) in your current working directory. cd into that directory for the rest of the steps.

When run without flags, the generator prompts for each value below. If you pass these as --name, --language, --visibility, --public-namespace, --resource-subtype, --model-name, and --register flags instead, use the flag forms noted in the table (where different from the interactive labels).

The generator creates a complete project. The key files you will edit:

• Python
• Go

2. Define the config

Open the generated resource file. Define config attributes for the sensors to monitor and the alert thresholds. For this example, we use:

• sensor_names — names of the sensors to poll. Required.

• max_temp — temperature threshold above which to create an alert. Required; units match whatever your sensor reports.

• poll_interval_secs — seconds between polls. Optional; defaults to 10.

• Python

• Go

In src/models/temp_alert.py, find the generated class TempAlert(Generic, EasyResource): and add the following instance-variable declarations inside the class, after the existing MODEL declaration:

    sensor_names: list[str]
    max_temp: float
    poll_interval_secs: float
    alerts: list[dict]
    _monitor_task: Optional[asyncio.Task]
    _stop_event: asyncio.Event

In the generated .go file, add fields to the Config struct. Each field needs a json tag matching the attribute name users set in their config JSON.

Then update the Validate method. It returns three values: a list of required dependency names, a list of optional dependency names, and an error.

type Config struct {
    SensorNames  []string `json:"sensor_names"`
    MaxTemp      float64  `json:"max_temp"`
    PollInterval float64  `json:"poll_interval_secs"`
}

func (cfg *Config) Validate(path string) ([]string, []string, error) {
    if len(cfg.SensorNames) == 0 {
        return nil, nil, fmt.Errorf("sensor_names is required")
    }
    if cfg.MaxTemp == 0 {
        return nil, nil, fmt.Errorf("max_temp is required")
    }
    // 1. Declare: return all sensor names as required dependencies
    return cfg.SensorNames, nil, nil
}

3. Implement the constructor

The constructor receives the validated config and a dependencies map containing the resources you declared in the validation method. Look up each dependency by name, store it on your struct/instance, and start the background monitoring loop.

Update validate_config and new:

    @classmethod
    def validate_config(
        cls, config: ComponentConfig
    ) -> Tuple[Sequence[str], Sequence[str]]:
        fields = config.attributes.fields
        if "sensor_names" not in fields:
            raise Exception("sensor_names is required")
        if "max_temp" not in fields:
            raise Exception("max_temp is required")
        sensor_names = [\
            v.string_value\\
            for v in fields["sensor_names"].list_value.values\
        ]
        # 1. Declare: return sensor names as required dependencies
        return sensor_names, []

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

fields = config.attributes.fields
        instance.sensor_names = [\
            v.string_value\\
            for v in fields["sensor_names"].list_value.values\
        ]
        instance.max_temp = fields["max_temp"].number_value
        instance.poll_interval_secs = (
            fields["poll_interval_secs"].number_value
            if "poll_interval_secs" in fields
            else 10.0
        )

# 2. Resolve: find each sensor in the dependencies map
        instance.sensors = {}
        for name, dep in dependencies.items():
            if name.name in instance.sensor_names:
                instance.sensors[name.name] = dep

# Start the monitor loop
        instance._stop_event = asyncio.Event()
        instance._monitor_task = asyncio.create_task(instance._monitor_loop())
        return instance

The generator also emits do_command and get_status method stubs that raise NotImplementedError. You’ll replace do_command in Step 5; leave get_status alone unless your service has a meaningful status to report.

The generator emits a compound struct type (alertMonitorTempAlert) with resource.AlwaysRebuild embedded and two constructor functions; a private newAlertMonitorTempAlert that unpacks the raw config and delegates to a public NewTempAlert that takes a typed *Config. Keep that layout. resource.NativeConfig converts the raw config into your typed struct. sensor.FromProvider looks up a sensor dependency by name from the dependencies map.

4. Implement the background loop

The monitor loop polls sensors at a fixed interval and checks readings against thresholds. When a reading exceeds the threshold, it creates an alert.

    async def _monitor_loop(self):
        while not self._stop_event.is_set():
            for name, s in self.sensors.items():
                try:
                    # 3. Use: call methods on dependencies
                    readings = await s.get_readings()
                    temp = readings.get("temperature")
                    if temp is not None and temp > self.max_temp:
                        alert = {
                            "sensor": name,
                            "value": temp,
                            "threshold": self.max_temp,
                            "time": datetime.now().isoformat(),
                        }
                        self.alerts.append(alert)
                        self.logger.warning(
                            "Alert: %s reported %.1f (threshold: %.1f)",
                            name, temp, self.max_temp,
                        )
                except Exception as e:
                    self.logger.error("Failed to read %s: %s", name, e)

try:
                await asyncio.wait_for(
                    self._stop_event.wait(),
                    timeout=self.poll_interval_secs,
                )
                break  # Stop event was set
            except asyncio.TimeoutError:
                pass  # Continue polling

5. Implement DoCommand

DoCommand is the interface your application code uses to interact with the service. Define a command vocabulary that makes sense for your module.

    async def do_command(
        self,
        command: Mapping[str, ValueTypes],
        *,
        timeout: Optional[float] = None,
        **kwargs,
    ) -> Mapping[str, ValueTypes]:
        cmd = command.get("command", "")

if cmd == "get_alerts":
            # Snapshot so the list isn't mutated during serialization
            return {"alerts": list(self.alerts)}

if cmd == "get_alert_count":
            return {"count": len(self.alerts)}

if cmd == "acknowledge":
            self.alerts.clear()
            return {"status": "ok"}

if cmd == "set_threshold":
            self.max_temp = command["max_temp"]
            return {"status": "ok", "max_temp": self.max_temp}

raise Exception(f"Unknown command: {cmd}")

6. Handle shutdown

When viam-server stops the module or reconfigures it, your background loop must stop cleanly. Without this, goroutines or async tasks leak.

    async def close(self):
        self._stop_event.set()
        if self._monitor_task is not None:
            await self._monitor_task
            self._monitor_task = None
        self.logger.info("TempAlert monitor stopped")

7. Review the entry point

The generator also creates the entry point file that viam-server launches. You typically do not need to modify it.

src/main.py:

import asyncio
from viam.module.module import Module
from models.temp_alert import TempAlert as TempAlertModel

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

run_from_registry() automatically discovers all imported resource classes and registers them with viam-server. If you add more models to your module, import them here.

8. Test locally

Deploy with hot reloading:

Ensure you have at least one sensor configured on your machine (this is the resource your logic module will monitor).

Use the CLI to build and deploy your module. Replace <machine-part-id> with your machine’s part ID. At the top of your machine’s page, click the Live / Offline status dropdown, then click Part ID to copy it.

# Build in the cloud and deploy to the machine
viam module reload --part-id <machine-part-id>

If your development machine and target machine share the same architecture, you can build locally instead:

# Build locally and transfer to the machine
viam module reload-local --part-id <machine-part-id>

Use reload (cloud build) when developing on a different architecture than your target. Use reload-local when architectures match for faster iteration.

9. Schedule logic with jobs (optional)

Instead of running a continuous background loop, you can use jobs to have viam-server call your service’s DoCommand method on a schedule. This is useful for periodic tasks that don’t need sub-second polling.

1. In the Viam app, open your machine’s CONFIGURE tab. Click the + icon to add a resource and select Job.

2. Name the job and click Create.

3. Set the Schedule to one of:

   • Interval – a Go duration string like 5s, 1m, or 2h30m.
   • Cron – a 5- or 6-part cron expression (for example, 0 */5 * * *).

4. Select your service resource by name.

5. Select the DoCommand Method and specify the Command, for example:

{ "command": "get_alerts" }

6. Optionally, adjust the Log threshold to raise or lower the log verbosity for this job’s invocations compared to the module’s default.

7. Click Save.

viam-server calls DoCommand with the specified arguments on the configured schedule. You can view job history (last 10 successes and failures) in the machine’s configuration.