add-device

Invocation: /add-device [manufacturer] [model] [component-type...]

Scaffold and implement a new device integration for the flowchem package. Creates the required file structure, then uses device documentation to implement the hardware communication layer. Request documentation from the user — do not leave the communication layer as stubs.

---

Step 1 — Collect required information

Parse slash-command arguments:

• Arg 1: manufacturer name

• Arg 2: device model name

• Arg 3+: component type tokens (optional)

If any required information is missing, use AskUserQuestion to collect everything in one pass:

1. Manufacturer name — company that makes the hardware (e.g. “Hamilton”, “New Era”)

2. Device model — exact model designation (e.g. “ML600”, “NE-1000”)

3. Approach — add directly to flowchem (src/flowchem/devices/) or as an external plugin. Use direct approach unless the device requires heavy new dependencies.

4. Component types — which functional interfaces does this device expose? One or more of:

   • syringe_pump — infuse / withdraw / stop (displacement pumps that can reverse)

   • hplc_pump — infuse / stop only (peristaltic, HPLC, unidirectional)

   • pump — generic pump, when neither above fits

   • valve — rotary multiport valve

   • pressure_sensor

   • photo_sensor

   • temperature — set/get temperature with configurable limits

   • stirring — set/get stir speed

   • mass_flow_controller

   • adc — analog-to-digital read

   • dac — digital-to-analog write

   • custom — define your own component from FlowchemComponent

5. Simulated variant? (yes/no) — creates a {DeviceClass}Sim stub for offline testing

---

Step 2 — Derive naming identifiers

Variable	Rule	Example
manufacturer_dir	manufacturer.lower(), spaces/hyphens → underscores	“New Era” → new_era
device_snake	model.lower(), spaces/hyphens → underscores, strip special chars	“NE-1000” → ne_1000
DeviceClass	PascalCase of model, strip hyphens/spaces	“NE-1000” → Ne1000

Component suffix table:

Component type	Class suffix	File suffix
syringe_pump / hplc_pump / pump	Pump	_pump
valve	Valve	_valve
pressure_sensor	PressureSensor	_pressure_sensor
photo_sensor	PhotoSensor	_photo_sensor
temperature	TemperatureControl	_temperature_control
stirring	StirringControl	_stirring_control
mass_flow_controller	MFC	_mfc
adc	ADC	_adc
dac	DAC	_dac
custom	(ask user)	(ask user)

If a device has multiple components of the same type, add a qualifier (e.g. Ne1000InletValve, Ne1000OutletValve).

---

Step 3 — Check if manufacturer directory exists

Read src/flowchem/devices/__init__.py. If from .{manufacturer_dir} import * already exists, only add new files inside the existing directory. Otherwise create the full directory structure and append the wildcard import.

---

Step 4 — Create scaffolding files

Create all files with method stubs (raise NotImplementedError). The communication layer will be filled in Step 5 once the documentation has been analysed.

4a. Device file — src/flowchem/devices/{manufacturer_dir}/{device_snake}.py

"""{ManufacturerStr} {ModelStr} device driver."""

from __future__ import annotations

from loguru import logger

from flowchem.components.device_info import DeviceInfo
from flowchem.devices.flowchem_device import FlowchemDevice
from flowchem.devices.{manufacturer_dir}.{device_snake}_{component_suffix} import {DeviceClass}{ComponentClassSuffix}
# Repeat the import line above for every component file.


class {DeviceClass}(FlowchemDevice):
    """{ManufacturerStr} {ModelStr}."""

    def __init__(self, name: str) -> None:
        super().__init__(name)
        self.device_info = DeviceInfo(
            manufacturer="{ManufacturerStr}",
            model="{ModelStr}",
        )
        # TODO: connection parameters will be added in Step 5

    async def initialize(self) -> None:
        """Open connection and register components."""
        # TODO: establish connection — Step 5
        logger.info(f"{self.device_info.manufacturer} {self.device_info.model} '{self.name}' initialized.")
        self.components.append({DeviceClass}{ComponentClassSuffix}("{component_name}", self))

    async def send_command(self, command: str) -> str:
        """Send a raw command and return the reply. All I/O lives here."""
        # TODO: implement in Step 5
        raise NotImplementedError

4b. Component file(s) — src/flowchem/devices/{manufacturer_dir}/{device_snake}_{file_suffix}.py

"""{ManufacturerStr} {ModelStr} — {ComponentType} component."""

from __future__ import annotations

from typing import TYPE_CHECKING

from flowchem.components.pumps.syringe_pump import SyringePump  # change base class per guide below

if TYPE_CHECKING:
    from .{device_snake} import {DeviceClass}


class {DeviceClass}{ComponentClassSuffix}(SyringePump):
    hw_device: {DeviceClass}

    def __init__(self, name: str, hw_device: {DeviceClass}) -> None:
        super().__init__(name, hw_device)
        # Device-specific extra routes will be added in Step 5 if needed.

    # Stubs — implemented in Step 5:
    async def infuse(self, rate: str = "", volume: str = "") -> bool:
        raise NotImplementedError

    async def stop(self) -> bool:
        raise NotImplementedError

    async def is_pumping(self) -> bool:
        raise NotImplementedError

    async def withdraw(self, rate: str = "", volume: str = "") -> bool:
        raise NotImplementedError

See the component base class guide at the bottom of this file for the correct base class and method signatures per component type.

4c. Manufacturer __init__.py

"""{ManufacturerStr} devices."""
from .{device_snake} import {DeviceClass}
__all__ = ["{DeviceClass}"]

4d. Top-level registry (new manufacturers only)

Append to src/flowchem/devices/__init__.py:

from .{manufacturer_dir} import *

4e. Simulated variant (if requested)

Add at the bottom of the device file and export from __init__.py:

class {DeviceClass}Sim({DeviceClass}):
    """{ModelStr} simulator — no hardware required."""

    async def initialize(self) -> None:
        self.device_info.version = "sim"
        logger.info(f"[SIM] {self.name} initialized.")
        self.components.append({DeviceClass}{ComponentClassSuffix}("{component_name}", self))

    async def send_command(self, command: str) -> str:
        logger.debug(f"[SIM] {self.name} ← {command!r}")
        return "OK"

---

Step 5 — Implement the communication layer

Do not skip this step. Request documentation and implement the actual protocol before finishing.

5a. Request documentation

Ask the user to provide any of the following (the more the better):

• Programming/communication manual (PDF path, URL, or pasted text)

• Serial/network connection parameters if already known (baud rate, port, IP, etc.)

• Any existing working code in any language that talks to this device

• Known command examples (e.g. “sending S\r stops the pump”)

If the user provides a PDF path, read it with the Read tool. If they provide a URL, fetch it.

5b. Analyse the protocol

From the documentation, extract every item in this checklist. Flag any that are missing (see 5d).

Physical / transport layer

• Interface type: RS-232, RS-485, USB-CDC (appears as COM port), Ethernet TCP, USB-HID, other

• For serial: baud rate, data bits, parity (N/E/O), stop bits, flow control (None/RTS-CTS/XON-XOFF)

• For Ethernet: port number, connection mode (persistent / per-command)

• For vendor library: library name, installation method

Message framing

• Command terminator (e.g. \r, \n, \r\n, \x03, none)

• Response terminator or fixed response length

• Is there a prompt character the device sends before each reply (e.g. >)? Strip it.

• Multi-packet responses? (read until terminator vs. read N bytes)

• Maximum response wait time / timeout (typical: 1–10 s)

Command syntax

• Encoding: plain ASCII text, SCPI, Modbus RTU, binary protocol, JSON, other

• Command structure: prefix + verb + value, or register-based, or opcode + payload

• Case sensitivity

• Address/node prefix required (e.g. /1 in Hamilton, @01 in some pumps)

• Checksum or CRC required?

Device behaviour

• Does the device echo the command back before the response? (must be consumed and discarded)

• Is there an initialisation handshake on connect?

• Error response format — how does the device signal a command failure?

• Status / busy-idle query command (needed for is_pumping(), is_target_reached(), etc.)

Per-component commands — for each component method, identify the exact command string:

• Map infuse(rate, volume) → command

• Map stop() → command

• Map is_pumping() → command + response parsing

• Map set_temperature(temp) → command + value encoding

• etc.

5c. Implement the transport layer in the device file

Choose the pattern that matches the protocol. Replace the send_command stub and add connection setup to __init__ / initialize().

Pattern A — ASCII serial (most common)

import asyncio
import aioserial
from flowchem.utils.exceptions import InvalidConfigurationError

class {DeviceClass}(FlowchemDevice):

    def __init__(self, name: str, port: str, baudrate: int = 9600,
                 parity: str = "N", stopbits: int = 1) -> None:
        super().__init__(name)
        self.device_info = DeviceInfo(manufacturer="...", model="...")
        self._port = port
        self._baudrate = baudrate
        self._parity = parity
        self._stopbits = stopbits
        self._serial: aioserial.AioSerial | None = None
        self._lock = asyncio.Lock()

    async def initialize(self) -> None:
        try:
            self._serial = aioserial.AioSerial(
                port=self._port,
                baudrate=self._baudrate,
                parity=self._parity,
                stopbits=self._stopbits,
                timeout=5,
            )
        except Exception as e:
            raise InvalidConfigurationError(
                f"Cannot open {self._port}: {e}"
            ) from e
        # Optional: send an identification query and store result in device_info.version
        self.device_info.version = await self._query_version()
        logger.info(f"Connected to {self.device_info.manufacturer} '{self.name}' on {self._port}")
        self.components.append(...)

    async def send_command(self, command: str) -> str:
        """Send ASCII command and return stripped reply."""
        async with self._lock:
            raw = (command + "\r").encode()   # adjust terminator per manual
            await self._serial.write_async(raw)
            reply = await self._serial.readline_async()
        return reply.decode().strip()

    async def _query_version(self) -> str:
        # TODO: replace with real version command from the manual
        return "unknown"

Pattern B — SCPI (instruments: oscilloscopes, power supplies, analysers)

import asyncio
import aioserial

class {DeviceClass}(FlowchemDevice):

    CMD_TERMINATOR = "\n"

    async def send_command(self, command: str) -> str:
        async with self._lock:
            await self._serial.write_async((command + self.CMD_TERMINATOR).encode())
            if "?" in command:          # query — expect a response
                reply = await self._serial.readline_async()
                return reply.decode().strip()
            return ""

    async def query_errors(self) -> str:
        return await self.send_command("SYST:ERR?")

Pattern C — Modbus RTU

from pymodbus.client import AsyncModbusSerialClient

class {DeviceClass}(FlowchemDevice):

    def __init__(self, name: str, port: str, slave_id: int = 1) -> None:
        super().__init__(name)
        self._port = port
        self._slave_id = slave_id
        self._client: AsyncModbusSerialClient | None = None

    async def initialize(self) -> None:
        self._client = AsyncModbusSerialClient(
            port=self._port, baudrate=9600, parity="N", stopbits=1
        )
        await self._client.connect()
        ...

    async def read_register(self, address: int) -> int:
        result = await self._client.read_holding_registers(address, count=1, slave=self._slave_id)
        return result.registers[0]

    async def write_register(self, address: int, value: int) -> None:
        await self._client.write_register(address, value, slave=self._slave_id)

    # send_command is not used in Modbus — components call read_register / write_register directly

Pattern D — Ethernet TCP

import asyncio

class {DeviceClass}(FlowchemDevice):

    def __init__(self, name: str, ip_address: str, port: int = 5000) -> None:
        super().__init__(name)
        self._ip = ip_address
        self._tcp_port = port
        self._reader: asyncio.StreamReader | None = None
        self._writer: asyncio.StreamWriter | None = None
        self._lock = asyncio.Lock()

    async def initialize(self) -> None:
        self._reader, self._writer = await asyncio.open_connection(self._ip, self._tcp_port)
        logger.info(f"TCP connection to {self._ip}:{self._tcp_port} established.")
        self.components.append(...)

    async def send_command(self, command: str) -> str:
        async with self._lock:
            self._writer.write((command + "\r\n").encode())
            await self._writer.drain()
            reply = await self._reader.readline()
        return reply.decode().strip()

Pattern E — Vendor library wrapping

from vendor_package import VendorDevice   # the third-party library

class {DeviceClass}(FlowchemDevice):

    def __init__(self, name: str, port: str) -> None:
        super().__init__(name)
        self._port = port
        self._device: VendorDevice | None = None

    async def initialize(self) -> None:
        loop = asyncio.get_event_loop()
        # Wrap synchronous vendor calls with run_in_executor to avoid blocking the event loop
        self._device = await loop.run_in_executor(None, VendorDevice, self._port)
        self.components.append(...)

    async def send_command(self, command: str) -> str:
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, self._device.send, command)

5d. Flag missing information — stop and ask the user

If any of the following cannot be answered from the documentation, stop and explicitly list what is missing before writing any code. Do not guess or leave silent stubs.

Critical gaps that block implementation:

• Connection parameters (baud rate, port, IP) not documented

• Command terminator / response format not specified

• No command exists in the manual for a required component method (e.g. no way to query pump status)

• Response parsing for numeric values is ambiguous (units? scaling factor? byte order?)

• Handshake or initialisation sequence not described

• Error response indistinguishable from a normal reply

Example flag message to the user:

The manual describes how to set the flow rate (command F{value}) but does not document a command to query whether the pump is currently running. is_pumping() cannot be implemented without this. Please check:

• Is there a status query command? (e.g. a ? or STATUS command)

• Is there a way to read back the current flow rate, which could serve as a proxy?

• Or should is_pumping() always return True while a command is in progress (requires tracking state internally)?

5e. Implement component methods

For each method stub in the component files, implement the actual command based on the analysed protocol. Use ureg (pint) for all physical quantities.

Key conventions:

• Parse rate/volume strings via ureg.Quantity(value).m_as("ml/min") to get a float in the expected unit

• Use from flowchem import ureg at the top of the device file

• For base classes that do range validation in set_temperature() / set_speed(), call await super().method(arg) first and use the returned pint.Quantity — then translate to the device command

Example — implementing infuse for an ASCII serial pump:

async def infuse(self, rate: str = "", volume: str = "") -> bool:
    from flowchem import ureg
    rate_val = ureg.Quantity(rate).m_as("ml/min") if rate else 1.0
    vol_val  = ureg.Quantity(volume).m_as("ml")   if volume else 0.0
    reply = await self.hw_device.send_command(f"FR{rate_val:.3f}")
    if volume:
        await self.hw_device.send_command(f"VOL{vol_val:.3f}")
    await self.hw_device.send_command("RUN")
    return reply != "ERR"

---

Step 6 — Verification checklist

Device class

• Subclasses FlowchemDevice

• __init__ calls super().__init__(name) first

• self.device_info = DeviceInfo(manufacturer=..., model=...) set in __init__

• initialize() appends at least one component to self.components

• All I/O (serial/TCP/library) is in the device class; no component imports transport libraries

Component class(es)

• Correct base class from the guide below

• hw_device: {DeviceClass} class attribute declared

• __init__ calls super().__init__(name, hw_device) — TemperatureControl passes TempRange too

• No direct I/O; all calls go through self.hw_device

• SyringePump subclasses declare is_withdrawing_capable() as @staticmethod

• from __future__ import annotations at top of each component file

• Device class imported inside if TYPE_CHECKING: block

• No method still raises NotImplementedError unless flagged as genuinely not supported

Package wiring

• Manufacturer __init__.py has __all__ listing all exported classes

• src/flowchem/devices/__init__.py contains from .{manufacturer_dir} import * (new manufacturer only)

Suggest this test snippet to verify the server starts:

# {device_snake}_test.toml
[device.my-{device_snake}]
type = "{DeviceClass}"
port = "COM1"   # adjust to actual port / ip_address

Run: flowchem {device_snake}_test.toml

---

Appendix — Component base class guide

Choose the base class that genuinely matches the device interface. If fitting the interface requires unnatural workarounds, inherit from FlowchemComponent directly.

syringe_pump → from flowchem.components.pumps.syringe_pump import SyringePump

@staticmethod
def is_withdrawing_capable() -> bool: return True  # False if no reverse

async def infuse(self, rate: str = "", volume: str = "") -> bool: ...
async def stop(self) -> bool: ...
async def is_pumping(self) -> bool: ...
async def withdraw(self, rate: str = "", volume: str = "") -> bool: ...

hplc_pump → from flowchem.components.pumps.hplc_pump import HPLCPump

# is_withdrawing_capable() is False by default
async def infuse(self, rate: str = "", volume: str = "") -> bool: ...
async def stop(self) -> bool: ...
async def is_pumping(self) -> bool: ...

valve — ask user for port layout before choosing:

• Pre-built: from flowchem.components.valves.distribution_valves import SixPortDistributionValve (also Two/Four/Eight/Ten/Twelve/Sixteen)

• Injection: from flowchem.components.valves.injection_valves import SixPortTwoPositionValve

• Custom: from flowchem.components.valves.valve import Valve — supply stator_ports, rotor_ports; implement _change_connections(); hw_device must have get_raw_position() / set_raw_position()

pressure_sensor → from flowchem.components.sensors.pressure_sensor import PressureSensor

async def read_pressure(self, units: str = "bar") -> float: ...

photo_sensor → from flowchem.components.sensors.photo_sensor import PhotoSensor

async def acquire_signal(self): ...
async def calibrate_zero(self): ...

temperature → from flowchem.components.technical.temperature import TemperatureControl, TempRange

# __init__ must pass TempRange with real hardware limits:
# super().__init__(name, hw_device, TempRange(min=ureg.Quantity("-40 degC"), max=ureg.Quantity("200 degC")))
# set_temperature() in the base validates range → call super() first, then translate to command

async def get_temperature(self) -> float: ...
async def is_target_reached(self) -> bool: ...
async def power_on(self): ...
async def power_off(self): ...

stirring → from flowchem.components.technical.stirring import StirringControl

# super().__init__(name, hw_device, min_speed=100, max_speed=1400)  # set real limits
# set_speed() in the base validates range → call super() first, then translate to command

async def get_speed(self) -> float: ...
async def get_speed_setpoint(self) -> float: ...
async def power_on(self): ...
async def power_off(self): ...
async def is_on(self) -> bool: ...

mass_flow_controller → from flowchem.components.technical.flow import MassFlowController

async def set_flow_setpoint(self, flowrate: str = "0 ml/min") -> bool: ...
async def get_flow_setpoint(self) -> float: ...
async def stop(self) -> bool: ...

adc → from flowchem.components.technical.ADC import AnalogDigitalConverter

async def read(self) -> float: ...

dac → from flowchem.components.technical.DAC import DigitalAnalogConverter

async def read(self) -> float: ...
async def set(self, value: str = "0 V") -> bool: ...

custom → from flowchem.components.flowchem_component import FlowchemComponent Define methods and register all routes manually via self.add_api_route(path, fn, methods=["GET"|"PUT"]).

---

Reference: key files

Purpose	Path
Base device class	src/flowchem/devices/flowchem_device.py
Device registry	src/flowchem/devices/__init__.py
Canonical minimal example	src/flowchem/devices/fakedevice/
All component base classes	src/flowchem/components/
Add-device guide	docs/development/additional/add_device/add_to_flowchem.md
Plugin guide	docs/development/additional/add_device/add_as_plugin.md
Real-device walkthrough	docs/development/additional/add_device/example_explained.md