# Design Document: Digital Loggers (DLI) Power Switch Control System
## 1. Overview
This system provides an MCP (Model Context Protocol) interface for an AI agent to control [Digital Loggers (DLI) Web Power Switches](https://www.digital-loggers.com/LPC9.html). It allows the agent to discover available hardware, query the status of specific outlets or groups, and safely control power states (On/Off/Cycle).
## 2. Architecture & Technology Stack
* **Language:** Python 3.10+
* **Protocol:** Command-line interface and MCP server using `argparse` and `fastmcp`.
* **Execution Model:** Asynchronous (using `asyncio`).
* **Concurrency:**
* `asyncio.Lock` is used to ensure atomic read-modify-write operations on the configuration file.
* **Threading:** `asyncio.to_thread` is used to offload blocking I/O operations (HTTP requests to hardware via `power-switch-pro`) to a thread pool, preventing the asyncio event loop from blocking.
* **Parallelism:** Inventory discovery is parallelized across switches using `asyncio.gather`.
* **Dependencies:**
* [`power-switch-pro`](https://pypi.org/project/power-switch-pro/) (BSD-3-Clause): Python client for DLI REST API.
* **Logging:** Uses Python's built-in `logging` module for structured error reporting.
* **Authentication:** Handled by `power-switch-pro` (Digest Auth).
* **Mocking:** The server includes a conditional mock capability. If the environment variable `DLI_MCP_ENV` is set to `TEST`, `get_client` dynamically adds the `tests/` directory to the search path and returns a `MockDLIPowerSwitch` (defined in `tests/mock_device.py`) regardless of the IP address. This ensures that test artifacts remain separate from production code and normal operation is unaffected.
## 3. Command-Line Interface
The `server.py` script is a command-line tool that uses `argparse` to handle different commands. This functionality is encapsulated in a `main()` function to improve testability and module import behavior.
The available commands are:
* `inventory`: Lists all switches and their outlet statuses.
* `power_action`: Performs a power action (on, off, cycle) on a specific outlet.
* `group_power_action`: Performs a power action on a group of outlets.
* `sync_config_from_hardware`: Synchronizes outlet names from the hardware.
* `list_outlets`: Lists all outlets on a given switch.
* `update_outlet`: Updates the definition of an outlet.
## 4. Configuration Data Model
The system state is defined in a local file named `switches_config.json`. This file acts as the source of truth for device credentials, aliases, logical groupings, and descriptive context for the agent.
### Schema Definition (`switches_config.json`)
```json
{
"switches": [
{
"alias": "garage_rack",
"description": "Primary network rack located in the detached garage.",
"ip_address": "192.168.1.50",
"username": "admin",
"password": "password",
"controller_name": "DLI Power Switch 8.0",
"outlets": {
"1": {
"name": "Modem",
"description": "Fiber ONT. Cycle this to fix internet connectivity.",
"type": "critical"
},
"2": {
"name": "Router",
"type": "critical"
},
"3": {
"name": "Monitor",
"description": "Video wall display.",
"type": "standard"
},
"4": {
"name": "Security_DVR",
"description": "Do not reboot. Recording 24/7.",
"type": "prohibited"
}
}
},
{
"alias": "mock_switch",
"description": "A mock switch for testing.",
"ip_address": "127.0.0.1",
"username": "mock_user",
"password": "mock_password",
"outlets": {
"1": { "name": "Mock Outlet 1", "type": "standard" }
}
}
],
"groups": {
"network_stack": {
"members": ["garage_rack:Modem", "garage_rack:Router"],
"description": "Core internet infrastructure. Cycle only during outages."
},
"monitors": {
"members": ["garage_rack:3", "office_switch:Monitor"],
"description": "All display screens."
}
}
}
```
## 5. Shared Logic & Helpers
These helper functions must be implemented to standardize how switches and outlets are resolved from user input.
### 5.1. Helper: `get_client(switch_config)`
* **Purpose:** Get a power switch client, either real or mock.
* **Input:** `switch_config` (dict) - The configuration dictionary for a single switch.
* **Logic:**
1. If `os.environ["DLI_MCP_ENV"] == "TEST"`, dynamically import and return an instance of `MockDLIPowerSwitch` from `tests/mock_device.py`.
2. Otherwise, return an instance of the real `DLIPowerSwitch` from the `power-switch-pro` library.
### 5.2. Helper: `resolve_switch(identifier)`
* **Purpose:** Find a switch object using its Alias OR IP Address.
* **Input:** `identifier` (string) - e.g., "garage_rack" or "192.168.1.50".
* **Logic:**
1. Load `switches_config.json`.
2. Iterate through the `switches` list.
3. Return the switch object where `switch.alias == identifier` OR `switch.ip_address == identifier`.
4. **Error Handling:** If no match is found, raise `ValueError: "Device not found"`.
### 5.3. Helper: `resolve_outlet(switch, identifier)`
* **Purpose:** Find the physical outlet index using its Name OR Index.
* **Input:**
* `switch`: The switch object (from `resolve_switch`).
* `identifier`: (string) - e.g., "Modem" or "1".
* **Logic:**
1. **Index Match:** Check if `identifier` exists as a key in `switch.outlets` (e.g., is "1" a key?). If yes, return `identifier`.
2. **Name Lookup:** Iterate through all items in `switch.outlets`.
* If `outlet.name` (case-insensitive) matches `identifier`, return the corresponding **index** key.
3. **Failure:** If no match is found, raise `ValueError: "Outlet not found"`.
### 5.4. Helper: `save_config(config)`
* **Purpose:** Safely writes the configuration dictionary to `switches_config.json`.
* **Implementation:** Uses an atomic write pattern to prevent data corruption.
1. Write to a temporary file in the same directory.
2. Flush and fsync to ensure data is on disk.
3. Use `os.replace` to atomically swap the temporary file with the target file.
### 5.5. Helper: `_sync_switch_config_sync(switch_config)`
* **Purpose:** Connects to hardware and updates the `switch_config` dictionary **in-place** in memory.
* **Input:** `switch_config` (dict) - A dictionary representing a single switch configuration.
* **Logic:**
1. Initialize client using `get_client(switch_config)`.
2. Attempt to fetch controller name and update `switch_config["controller_name"]`.
3. Iterate through outlets (1-8):
* Fetch hardware outlet name.
* Update `switch_config["outlets"][index]["name"]`.
* If outlet entry doesn't exist in config, create a default one.
4. **Error Handling:** Catches exceptions per outlet (to allow partial sync) and prints tracebacks to stderr.
* **Note:** This function is blocking and should be run in a thread.
### 5.6. Helper: `validate_ip(ip)`
* **Purpose:** Validates that a string is a valid IPv4 or IPv6 address.
* **Input:** `ip` (string).
* **Logic:** Uses Python's `ipaddress` module. Raises `ValueError` if invalid.
### 5.7. Helper: `load_config()`
* **Purpose:** loads the JSON configuration from `switches_config.json`.
* **Logic:**
1. Check if file exists. If not, raise `FileNotFoundError`.
2. Open file with `utf-8` encoding.
3. `json.load(f)`.
4. **Error Handling:** Catch `json.JSONDecodeError` and raise `ValueError` with a descriptive message to inform the user of invalid JSON syntax.
## 6. Tools (MCP Interface)
### Tool 1: `get_inventory`
* **Description:** Returns the full system configuration, including all available switches, defined outlets, and their **current live power status**.
* **Input:** None.
* **Implementation Steps:**
1. Load `switches_config.json`.
2. **Parallel Execution:** Create tasks to fetch status for each switch concurrently using `asyncio.to_thread(_fetch_switch_status_sync, switch_config)`.
* `_fetch_switch_status_sync` initializes `client` and fetches status for all outlets (blocking).
3. **Gather Results:** `await asyncio.gather(*tasks)`.
4. **Merge Data:** Map the live state (`true`/`false`) from the results back to the static config structure.
5. **Group Status:** (Optional) Calculate aggregate status for groups.
6. Return the enriched JSON object.
### Tool 2: `power_action`
* **Description:** Controls a single outlet.
* **Input:**
* `switch_id`: (string) The Alias or IP (e.g., "garage_rack").
* `outlet_id`: (string) The Name or Index (e.g., "Modem" or "1").
* `action`: (string) "on", "off", or "cycle".
* **Implementation Steps:**
1. `switch = resolve_switch(switch_id)`
2. `index = resolve_outlet(switch, outlet_id)`
3. **Hard Block:** If `switch.outlets[index].type == "prohibited"`, raise `PermissionError`.
4. **Safety Check:** If `switch.outlets[index].type == "critical"` AND `action` is "off" or "cycle":
* Require explicit user confirmation.
5. **Execute:** Run blocking code in a thread: `await asyncio.to_thread(_power_action_sync, ...)`
* Instantiate client.
* Perform action: `outlet.on()`, `outlet.off()`, or `outlet.cycle()`.
### Tool 3: `group_power_action`
* **Description:** Controls a logical group of outlets.
* **Input:**
* `target`: (string) The name of a group (e.g., "network_stack").
* `action`: (string) "on", "off", or "cycle".
* **Implementation Steps:**
1. **Resolve Group:** Look up `target` in the `groups` section of the config.
2. **Parse Members:** The group `members` list contains strings like `"garage_rack:Modem"`.
3. **Validate:** Ensure all switches and outlets exist.
4. **Hard Block Check:** Scan all members of the group. If *any* member is of type `"prohibited"`, abort.
5. **Execute:** Iterate through the list sequentially.
* Call `power_action` (which now handles threading internally) for each member.
* If an individual action fails, record the error and continue.
6. **Report:** Return a summary string.
### Tool 4: `sync_config_from_hardware`
* **Description:** Updates the local `switches_config.json` names to match the physical device settings.
* **Input:** `switch_id` (string).
* **Implementation Steps:**
1. Acquire `CONFIG_LOCK`.
2. Load `switches_config.json`.
3. Find the switch object matching `switch_id`.
4. Offload sync logic to thread: `await asyncio.to_thread(_sync_switch_config_sync, switch_config)`.
5. Save the updated configuration to file (using atomic save).
6. Release `CONFIG_LOCK`.
7. Return success or error message.
### Tool 5: `add_switch`
* **Description:** Adds a new DLI power switch to the configuration.
* **Input:**
* `ip_address`: (string) The IP address of the new switch.
* `username`: (string) The username for the new switch.
* `password`: (string) The password for the new switch.
* **Implementation Steps:**
1. Validate `ip_address` using `validate_ip`.
2. Acquire `CONFIG_LOCK` (Async Lock).
3. Check if configuration file exists.
* If yes, load `switches_config.json`.
* If no, create directory (if needed) and initialize an empty configuration structure.
4. Generate a unique alias for the new switch.
5. Create a new switch configuration dictionary in memory.
6. Offload sync to thread: `await asyncio.to_thread(_sync_switch_config_sync, new_switch_config)`.
7. If sync is successful (or partially successful), append the new switch to the config list.
8. Save the updated configuration to file (Atomic write).
9. Release `CONFIG_LOCK`.
10. Return success message.
### Tool 6: `remove_switch`
* **Description:** Removes a DLI power switch from the configuration.
* **Input:**
* `switch_id`: (string) The Alias or IP address of the switch to remove.
* **Implementation Steps:**
1. Acquire `CONFIG_LOCK`.
2. Load `switches_config.json`.
3. Filter out the switch with the matching `alias` or `ip_address`.
4. If no switch was removed (count remains the same), raise `ValueError`.
5. Save the updated configuration.
6. Release `CONFIG_LOCK`.
### Tool 7: `list_outlets`
* **Description:** Lists all outlets and their status for a given switch.
* **Input:** `switch_id` (string).
* **Implementation Steps:**
1. `switch_config = resolve_switch(switch_id)`
2. Offload logic to thread: `await asyncio.to_thread(_list_outlets_sync, switch_config)`.
* Initialize client.
* Iterate from 1 to 8, fetching status.
3. Return a JSON array of outlet information.
### Tool 8: `update_outlet`
* **Description:** Updates the definition of an outlet in the configuration file and writes the new name to the hardware.
* **Input:**
* `switch_id`: (string) The Alias or IP (e.g., "garage_rack").
* `outlet_id`: (string) The Name or Index (e.g., "Modem" or "1").
* `new_name`: (string, optional) The new name for the outlet. This is written to both the config file and the hardware.
* `new_description`: (string, optional) The new description for the outlet. This is only written to the config file.
* `new_type`: (string, optional) The new type for the outlet (`standard`, `critical`, or `prohibited`). This is only written to the config file.
* **Implementation Steps:**
1. Acquire `CONFIG_LOCK`.
2. `config = load_config()`
3. `switch = resolve_switch(switch_id)`
4. `index = resolve_outlet(switch, outlet_id)`
5. If `new_name` is provided:
* Offload write to thread: `await asyncio.to_thread(_update_outlet_name_sync, ...)`
6. Update the fields in `config` for the given outlet.
7. `save_config(config)`
8. Release `CONFIG_LOCK`.
9. Return a success message.
## 7. Implementation Notes for Junior Engineer
### Code Style
* Use Python type hints for all function arguments and return values.
* Use `TypedDict` (e.g., `SwitchConfig`, `OutletConfig`) to define the structure of configuration dictionaries, ensuring type safety and readability over generic `Dict[str, Any]`.
### Testing Strategy
The project includes a comprehensive test suite using Python's built-in `unittest` framework. The tests are located in the `tests/` directory and aim for high code coverage to ensure reliability.
* **Mocking:** The system uses a conditional mocking strategy triggered by the `DLI_MCP_ENV` environment variable. When set to `TEST`, `tests/mock_device.py` is loaded. This decouples test logic from specific IP addresses and production code. The `unittest.mock` library is used extensively to patch modules and simulate exceptions.
* **Command-Line Interface Testing:** The CLI is tested using the `subprocess` module to run `server.py` as a separate process. The `DLI_MCP_ENV` variable is passed to these subprocesses to ensure they run in test mode. A separate test configuration file is used for these tests.
* **Critical/Prohibited Logic:** Specific unit tests are written to verify:
* "Prohibited" outlets raise a `PermissionError`.
* "Critical" outlets trigger the confirmation logic.
* **CI Pipeline:** A GitHub Actions workflow (`.github/workflows/test.yml`) is configured to automatically run the full test suite on Windows, Linux, and macOS across Python versions 3.10, 3.11, and 3.12. This ensures no regressions are introduced and maintains cross-platform compatibility.
* **Coverage:** The test suite is designed to achieve a high level of test coverage for the core server logic. The `coverage.py` package can be used to generate a coverage report. Unit tests themselves should not be counted in the coverage metrics.
To run coverage excluding the tests themselves:
```bash
python -m coverage run --omit="tests/*" tests/test_server.py
python -m coverage report -m
```
## References
* <https://pypi.org/project/power-switch-pro/>
* <https://github.com/bryankemp/power_switch_pro>
