# 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](https://docs.viam.com/build-modules/write-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](https://docs.viam.com/build-modules/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](https://docs.viam.com/cli/overview/#install) 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:

```bash
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).

| Prompt | What to enter | Why |
| --- | --- | --- |
| Set a module name: | `alert-monitor` | A short, descriptive name |
| Specify the language for the module: | `python` or `go` | Your implementation language |
| Visibility: | `private` | `private`: visible only within your org. `public`: visible to everyone. `public_unlisted`: usable by anyone who knows the module ID, but hidden from the registry page. You can change visibility later. |
| Namespace/Organization ID | Your organization namespace | Scopes the module to your org |
| Select a resource to be added to the module: | `Generic Service` (flag: `generic-service`) | Flexible service API |
| Set a model name of the resource: | `temp-alert` | The model name for your service |
| Register module | `yes` | Registers the module with Viam |

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

- [Python](https://docs.viam.com/build-modules/write-a-logic-module/#tabset-build-moduleswrite-a-logic-module-1-0)
- [Go](https://docs.viam.com/build-modules/write-a-logic-module/#tabset-build-moduleswrite-a-logic-module-1-1)

| File | Purpose |
| --- | --- |
| `src/main.py` | Entry point – starts the module server |
| `src/models/temp_alert.py` | Service class skeleton – you will edit this |
| `requirements.txt` | Python dependencies |
| `meta.json` | Module metadata for the registry |
| `setup.sh` | Installs dependencies into a virtualenv |
| `build.sh` | Packages the module for upload |
| `.github/workflows/deploy.yml` | CI workflow for cloud builds |

| File | Purpose |
| --- | --- |
| `cmd/module/main.go` | Entry point – starts the module server |
| `module.go` | Service implementation skeleton – you will edit this |
| `go.mod` | Go module definition |
| `Makefile` | Build targets |
| `meta.json` | Module metadata for the registry |
| `.github/workflows/deploy.yml` | CI workflow for cloud builds |

### 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](https://docs.viam.com/build-modules/write-a-logic-module/#tabset-build-moduleswrite-a-logic-module-2-0)
- [Go](https://docs.viam.com/build-modules/write-a-logic-module/#tabset-build-moduleswrite-a-logic-module-2-1)

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:

```python
    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.

```go
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`:

```python
    @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.

```python
    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.

```python
    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.

```python
    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`:

```python
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.

```bash
# 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:

```bash
# 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](https://app.viam.com/), 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:

```json
{ "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.