API Reference

This reference documentation is automatically generated from the source code docstrings.

CLI Application

configure(host=typer.Option(None, help='MQTT Broker Host'), port=typer.Option(None, help='MQTT Broker Port'), username=typer.Option(None, help='MQTT Username'), password=typer.Option(None, help='MQTT Password'), reset=typer.Option(False, '--reset', help='Reset configuration to defaults'), show=typer.Option(False, '--show', help='Show current configuration'))

Configure the MQTT broker connection settings.

Interactive prompts are used if no arguments are provided. Saved configuration is stored in ~/.config/ha-cli/config.json.

Parameters:

Name	Type	Description	Default
host	Optional[str]	MQTT Broker Host.	Option(None, help='MQTT Broker Host')
port	Optional[int]	MQTT Broker Port.	Option(None, help='MQTT Broker Port')
username	Optional[str]	MQTT Username.	Option(None, help='MQTT Username')
password	Optional[str]	MQTT Password.	Option(None, help='MQTT Password')
reset	bool	Reset configuration to defaults.	Option(False, '--reset', help='Reset configuration to defaults')
show	bool	Show current configuration.	Option(False, '--show', help='Show current configuration')

Source code in src/hamqtt/main.py

@app.command()
def configure(
    host: Optional[str] = typer.Option(None, help="MQTT Broker Host"),
    port: Optional[int] = typer.Option(None, help="MQTT Broker Port"),
    username: Optional[str] = typer.Option(None, help="MQTT Username"),
    password: Optional[str] = typer.Option(None, help="MQTT Password"),
    reset: bool = typer.Option(False, "--reset", help="Reset configuration to defaults"),
    show: bool = typer.Option(False, "--show", help="Show current configuration"),
) -> None:
    """
    Configure the MQTT broker connection settings.

    Interactive prompts are used if no arguments are provided.
    Saved configuration is stored in `~/.config/ha-cli/config.json`.

    Args:
        host (Optional[str]): MQTT Broker Host.
        port (Optional[int]): MQTT Broker Port.
        username (Optional[str]): MQTT Username.
        password (Optional[str]): MQTT Password.
        reset (bool): Reset configuration to defaults.
        show (bool): Show current configuration.
    """
    if reset:
        delete_config()
        console.print("[yellow]Configuration reset to defaults.[/yellow]")
        return

    current_config = load_config()

    if show:
        # manual dump to avoid showing secrets
        config_dict = current_config.model_dump()
        if config_dict["mqtt"]["password"]:
            config_dict["mqtt"]["password"] = "********"

        console.print("[bold blue]Current Configuration:[/bold blue]")
        console.print_json(data=config_dict)
        return

    if not (host or port or username or password):
        # Interactive mode if no args provided
        console.print("[bold blue]Interactive Configuration[/bold blue]")
        host = Prompt.ask("MQTT Broker Host", default=current_config.mqtt.host)
        port = IntPrompt.ask("MQTT Broker Port", default=current_config.mqtt.port)
        username = Prompt.ask("MQTT Username", default=current_config.mqtt.username or "") or None
        password = (
            Prompt.ask(
                "MQTT Password",
                password=True,
                default=current_config.mqtt.password or "",
            )
            or None
        )

        # Ask to verify/test connection?
        # For now just save
    else:
        # Update only provided values
        host = host or current_config.mqtt.host
        port = port or current_config.mqtt.port
        if username is None:
            username = current_config.mqtt.username
        if password is None:
            password = current_config.mqtt.password

    new_config = Config(mqtt=MqttConfig(host=host, port=port, username=username, password=password))
    save_config(new_config)
    console.print("[green]Configuration saved successfully![/green]")

register(unique_id=typer.Option(..., help='Unique ID for the entity'), name=typer.Option(..., help='Readable name for the entity'), device_class=typer.Option(None, help='Device class (e.g., temperature, humidity)'), unit=typer.Option(None, help='Unit of measurement'), component=typer.Option('sensor', help='HA Component (sensor, binary_sensor, etc.)'), state_topic=typer.Option(None, help='Custom state topic. Defaults to ha-cli/<unique_id>/state'), device_name=typer.Option(None, help='Device Name for grouping'), dry_run=typer.Option(False, help='Print payload instead of sending'))

Register a new entity with Home Assistant via MQTT Discovery.

Publishes a discovery configuration payload to the configured discovery topic.

Parameters:

Name	Type	Description	Default
unique_id	str	Unique ID for the entity.	Option(..., help='Unique ID for the entity')
name	str	Readable name for the entity.	Option(..., help='Readable name for the entity')
device_class	Optional[str]	Device class (e.g., temperature, humidity).	Option(None, help='Device class (e.g., temperature, humidity)')
unit	Optional[str]	Unit of measurement.	Option(None, help='Unit of measurement')
component	str	HA Component (sensor, binary_sensor, etc.).	Option('sensor', help='HA Component (sensor, binary_sensor, etc.)')
state_topic	Optional[str]	Custom state topic. Defaults to ha-cli//state.	Option(None, help='Custom state topic. Defaults to ha-cli/<unique_id>/state')
device_name	Optional[str]	Device Name for grouping entities.	Option(None, help='Device Name for grouping')
dry_run	bool	If True, print payload instead of sending.	Option(False, help='Print payload instead of sending')

Source code in src/hamqtt/main.py

@app.command()
def register(
    unique_id: str = typer.Option(..., help="Unique ID for the entity"),
    name: str = typer.Option(..., help="Readable name for the entity"),
    device_class: Optional[str] = typer.Option(None, help="Device class (e.g., temperature, humidity)"),
    unit: Optional[str] = typer.Option(None, help="Unit of measurement"),
    component: str = typer.Option("sensor", help="HA Component (sensor, binary_sensor, etc.)"),
    state_topic: Optional[str] = typer.Option(None, help="Custom state topic. Defaults to ha-cli/<unique_id>/state"),
    device_name: Optional[str] = typer.Option(None, help="Device Name for grouping"),
    dry_run: bool = typer.Option(False, help="Print payload instead of sending"),
) -> None:
    """
    Register a new entity with Home Assistant via MQTT Discovery.

    Publishes a discovery configuration payload to the configured discovery topic.

    Args:
        unique_id (str): Unique ID for the entity.
        name (str): Readable name for the entity.
        device_class (Optional[str]): Device class (e.g., temperature, humidity).
        unit (Optional[str]): Unit of measurement.
        component (str): HA Component (sensor, binary_sensor, etc.).
        state_topic (Optional[str]): Custom state topic. Defaults to ha-cli/<unique_id>/state.
        device_name (Optional[str]): Device Name for grouping entities.
        dry_run (bool): If True, print payload instead of sending.
    """
    config = load_config()

    # Default state topic convention
    if not state_topic:
        state_topic = f"ha-cli/{unique_id}/state"

    device = None
    if device_name:
        device = DeviceInfo(
            identifiers=[f"ha-cli-{device_name}"],
            name=device_name,
            manufacturer="HA CLI",
            model="CLI Client",
        )

    entity = DiscoveryEntity(
        unique_id=unique_id,
        name=name,
        device_class=device_class,
        unit_of_measurement=unit,
        state_topic=state_topic,
        device=device,
    )

    discovery_topic = entity.get_topic(prefix=config.mqtt.discovery_prefix, component=component)
    payload = entity.model_dump_json(exclude_none=True)

    if dry_run:
        console.print(f"[bold]Topic:[/bold] {discovery_topic}")
        console.print_json(payload)
    else:
        try:
            with MqttClient(config.mqtt) as client:
                client.publish(discovery_topic, payload, retain=True)
            console.print(f"[green]Registered entity {name} ({unique_id})[/green]")
            console.print(f"State Topic: {state_topic}")
        except Exception as e:
            console.print(f"[red]Failed to register:[/red] {e}")

send(unique_id=typer.Option(..., help='Unique ID of the entity'), state=typer.Option(..., help='State/Value to send'), topic=typer.Option(None, help='Override state topic. Defaults to ha-cli/<unique_id>/state'), dry_run=typer.Option(False, help='Print payload instead of sending'))

Send a state update for a registered entity.

Parameters:

Name	Type	Description	Default
unique_id	str	Unique ID of the entity.	Option(..., help='Unique ID of the entity')
state	str	State/Value to send.	Option(..., help='State/Value to send')
topic	Optional[str]	Override state topic. Defaults to ha-cli//state.	Option(None, help='Override state topic. Defaults to ha-cli/<unique_id>/state')
dry_run	bool	If True, print payload instead of sending.	Option(False, help='Print payload instead of sending')

Source code in src/hamqtt/main.py

@app.command()
def send(
    unique_id: str = typer.Option(..., help="Unique ID of the entity"),
    state: str = typer.Option(..., help="State/Value to send"),
    topic: Optional[str] = typer.Option(None, help="Override state topic. Defaults to ha-cli/<unique_id>/state"),
    dry_run: bool = typer.Option(False, help="Print payload instead of sending"),
) -> None:
    """
    Send a state update for a registered entity.

    Args:
        unique_id (str): Unique ID of the entity.
        state (str): State/Value to send.
        topic (Optional[str]): Override state topic. Defaults to ha-cli/<unique_id>/state.
        dry_run (bool): If True, print payload instead of sending.
    """
    config = load_config()

    if not topic:
        topic = f"ha-cli/{unique_id}/state"

    if dry_run:
        console.print(f"[bold]Topic:[/bold] {topic}")
        console.print(f"[bold]Payload:[/bold] {state}")
    else:
        try:
            with MqttClient(config.mqtt) as client:
                client.publish(topic, state)
            console.print(f"[green]Sent state '{state}' to '{topic}'[/green]")
        except Exception as e:
            console.print(f"[red]Failed to send state:[/red] {e}")

Configuration

Config

Bases: BaseModel

Global application configuration.

Attributes:

Name	Type	Description
mqtt	MqttConfig	The MQTT connection configuration.

Source code in src/hamqtt/config.py

class Config(BaseModel):
    """
    Global application configuration.

    Attributes:
        mqtt (MqttConfig): The MQTT connection configuration.
    """

    mqtt: MqttConfig

MqttConfig

Bases: BaseModel

Configuration for the MQTT connection.

Attributes:

Name	Type	Description
host	str	The hostname or IP address of the MQTT broker.
port	int	The port number of the MQTT broker. Defaults to 1883.
username	Optional[str]	The username for authentication. Defaults to None.
password	Optional[str]	The password for authentication. Defaults to None.
discovery_prefix	str	The prefix used for Home Assistant discovery topics. Defaults to "homeassistant".

Source code in src/hamqtt/config.py

class MqttConfig(BaseModel):
    """
    Configuration for the MQTT connection.

    Attributes:
        host (str): The hostname or IP address of the MQTT broker.
        port (int): The port number of the MQTT broker. Defaults to 1883.
        username (Optional[str]): The username for authentication. Defaults to None.
        password (Optional[str]): The password for authentication. Defaults to None.
        discovery_prefix (str): The prefix used for Home Assistant discovery topics. Defaults to "homeassistant".
    """

    host: str
    port: int = 1883
    username: Optional[str] = None
    password: Optional[str] = None
    discovery_prefix: str = "homeassistant"

delete_config()

Delete the configuration file.

Source code in src/hamqtt/config.py

def delete_config() -> None:
    """
    Delete the configuration file.
    """
    if CONFIG_FILE.exists():
        CONFIG_FILE.unlink()

load_config()

Load the application configuration from the config file.

If the configuration file does not exist, it attempts to migrate from the legacy location. If migration is not possible or file doesn't exist, a default configuration is returned with 'localhost' as the MQTT host.

Returns:

Name	Type	Description
Config	Config	The loaded or default configuration.

Source code in src/hamqtt/config.py

def load_config() -> Config:
    """
    Load the application configuration from the config file.

    If the configuration file does not exist, it attempts to migrate from the legacy location.
    If migration is not possible or file doesn't exist, a default configuration is returned
    with 'localhost' as the MQTT host.

    Returns:
        Config: The loaded or default configuration.
    """
    _migrate_config()

    if not CONFIG_FILE.exists():
        # Return default config if file doesn't exist
        return Config(mqtt=MqttConfig(host="localhost"))

    try:
        data = json.loads(CONFIG_FILE.read_text())
        return Config(**data)
    except Exception as e:
        # Fallback or error handling
        print(f"Error loading config: {e}")
        return Config(mqtt=MqttConfig(host="localhost"))

save_config(config)

Save the application configuration to the config file.

Creates the configuration directory if it does not exist.

Parameters:

Name	Type	Description	Default
config	Config	The configuration object to save.	required

Source code in src/hamqtt/config.py

def save_config(config: Config) -> None:
    """
    Save the application configuration to the config file.

    Creates the configuration directory if it does not exist.

    Args:
        config (Config): The configuration object to save.
    """

    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    CONFIG_FILE.write_text(config.model_dump_json(indent=2))

Discovery Models

DeviceInfo

Bases: BaseModel

Information about the device associated with the entity.

Attributes:

Name	Type	Description
identifiers	list[str]	A list of IDs that uniquely identify the device.
name	Optional[str]	The name of the device.
model	Optional[str]	The model of the device.
manufacturer	Optional[str]	The manufacturer of the device.
sw_version	Optional[str]	The software version of the device.

Source code in src/hamqtt/discovery.py

class DeviceInfo(BaseModel):
    """
    Information about the device associated with the entity.

    Attributes:
        identifiers (list[str]): A list of IDs that uniquely identify the device.
        name (Optional[str]): The name of the device.
        model (Optional[str]): The model of the device.
        manufacturer (Optional[str]): The manufacturer of the device.
        sw_version (Optional[str]): The software version of the device.
    """

    identifiers: list[str]
    name: Optional[str] = None
    model: Optional[str] = None
    manufacturer: Optional[str] = None
    sw_version: Optional[str] = None

DiscoveryEntity

Bases: BaseModel

Represents an entity for Home Assistant MQTT Discovery.

Attributes:

Name	Type	Description
name	Optional[str]	The name of the entity.
unique_id	str	A unique identifier for the entity.
object_id	Optional[str]	Used to generate the entity ID in Home Assistant.
device_class	Optional[str]	The class of the device (e.g., 'temperature').
state_topic	str	The MQTT topic where the entity's state is published.
unit_of_measurement	Optional[str]	The unit of measurement (e.g., '°C').
device	Optional[DeviceInfo]	Information about the device the entity belongs to.
icon	Optional[str]	The icon to display in the frontend.
value_template	Optional[str]	Template to extract the value from the payload.
payload_on	Optional[str]	The payload that represents the 'on' state.
payload_off	Optional[str]	The payload that represents the 'off' state.

Source code in src/hamqtt/discovery.py

class DiscoveryEntity(BaseModel):
    """
    Represents an entity for Home Assistant MQTT Discovery.

    Attributes:
        name (Optional[str]): The name of the entity.
        unique_id (str): A unique identifier for the entity.
        object_id (Optional[str]): Used to generate the entity ID in Home Assistant.
        device_class (Optional[str]): The class of the device (e.g., 'temperature').
        state_topic (str): The MQTT topic where the entity's state is published.
        unit_of_measurement (Optional[str]): The unit of measurement (e.g., '°C').
        device (Optional[DeviceInfo]): Information about the device the entity belongs to.
        icon (Optional[str]): The icon to display in the frontend.
        value_template (Optional[str]): Template to extract the value from the payload.
        payload_on (Optional[str]): The payload that represents the 'on' state.
        payload_off (Optional[str]): The payload that represents the 'off' state.
    """

    name: Optional[str] = None
    unique_id: str
    object_id: Optional[str] = None
    device_class: Optional[str] = None
    state_topic: str
    unit_of_measurement: Optional[str] = None
    device: Optional[DeviceInfo] = None
    icon: Optional[str] = None
    value_template: Optional[str] = None
    payload_on: Optional[str] = None
    payload_off: Optional[str] = None
    # Add other common fields as needed

    def get_topic(self, prefix: str = "homeassistant", component: str = "sensor") -> str:
        """
        Generates the discovery topic.
        Format: <discovery_prefix>/<component>/[<node_id>/]<object_id>/config
        """
        # Using unique_id as object_id for simplicity if not provided,
        # but topic construction usually relies on object_id.
        oid = self.object_id or self.unique_id
        return f"{prefix}/{component}/{oid}/config"

get_topic(prefix='homeassistant', component='sensor')

Generates the discovery topic. Format: //[/]/config

Source code in src/hamqtt/discovery.py

def get_topic(self, prefix: str = "homeassistant", component: str = "sensor") -> str:
    """
    Generates the discovery topic.
    Format: <discovery_prefix>/<component>/[<node_id>/]<object_id>/config
    """
    # Using unique_id as object_id for simplicity if not provided,
    # but topic construction usually relies on object_id.
    oid = self.object_id or self.unique_id
    return f"{prefix}/{component}/{oid}/config"

MQTT Client

MqttClient

A wrapper around paho.mqtt.client for handling MQTT connections and publishing.

This class supports usage as a context manager for automatic connection and disconnection.

Source code in src/hamqtt/mqtt_client.py

class MqttClient:
    """
    A wrapper around paho.mqtt.client for handling MQTT connections and publishing.

    This class supports usage as a context manager for automatic connection and disconnection.
    """

    def __init__(self, config: MqttConfig) -> None:
        """
        Initialize the MQTT client.

        Args:
            config (MqttConfig): The MQTT configuration.
        """
        self.config = config
        self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)

        if self.config.username:
            self.client.username_pw_set(self.config.username, self.config.password)

    def connect(self) -> None:
        """
        Connect to the MQTT broker and start the network loop.

        Raises:
             socket.error: If the connection fails.
             OSError: If the connection fails.
        """
        try:
            self.client.connect(self.config.host, self.config.port, 60)
            self.client.loop_start()
        except (socket.error, OSError) as e:
            console.print(f"[bold red]Connection failed:[/bold red] {e}")
            raise

    def disconnect(self) -> None:
        """
        Stop the network loop and disconnect from the MQTT broker.
        """
        self.client.loop_stop()
        self.client.disconnect()

    def publish(self, topic: str, payload: Any, retain: bool = False) -> Any:
        """
        Publish a message to an MQTT topic.

        If the payload is not a string, it is serialized to JSON.
        This method blocks until the message is published.

        Args:
            topic (str): The MQTT topic to publish to.
            payload (Any): The message payload.
            retain (bool): Whether to retain the message. Defaults to False.

        Returns:
            mqtt.MQTTMessageInfo: Information about the published message.
        """
        if not isinstance(payload, str):
            payload = json.dumps(payload)

        info = self.client.publish(topic, payload, retain=retain)
        info.wait_for_publish()
        return info

    def __enter__(self) -> "MqttClient":
        self.connect()
        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        self.disconnect()

__init__(config)

Initialize the MQTT client.

Parameters:

Name	Type	Description	Default
config	MqttConfig	The MQTT configuration.	required

Source code in src/hamqtt/mqtt_client.py

def __init__(self, config: MqttConfig) -> None:
    """
    Initialize the MQTT client.

    Args:
        config (MqttConfig): The MQTT configuration.
    """
    self.config = config
    self.client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)

    if self.config.username:
        self.client.username_pw_set(self.config.username, self.config.password)

connect()

Connect to the MQTT broker and start the network loop.

Raises:

Type	Description
error	If the connection fails.
OSError	If the connection fails.

Source code in src/hamqtt/mqtt_client.py

def connect(self) -> None:
    """
    Connect to the MQTT broker and start the network loop.

    Raises:
         socket.error: If the connection fails.
         OSError: If the connection fails.
    """
    try:
        self.client.connect(self.config.host, self.config.port, 60)
        self.client.loop_start()
    except (socket.error, OSError) as e:
        console.print(f"[bold red]Connection failed:[/bold red] {e}")
        raise

disconnect()

Stop the network loop and disconnect from the MQTT broker.

Source code in src/hamqtt/mqtt_client.py

def disconnect(self) -> None:
    """
    Stop the network loop and disconnect from the MQTT broker.
    """
    self.client.loop_stop()
    self.client.disconnect()

publish(topic, payload, retain=False)

Publish a message to an MQTT topic.

If the payload is not a string, it is serialized to JSON. This method blocks until the message is published.

Parameters:

Name	Type	Description	Default
topic	str	The MQTT topic to publish to.	required
payload	Any	The message payload.	required
retain	bool	Whether to retain the message. Defaults to False.	False

Returns:

Type	Description
Any	mqtt.MQTTMessageInfo: Information about the published message.

Source code in src/hamqtt/mqtt_client.py

def publish(self, topic: str, payload: Any, retain: bool = False) -> Any:
    """
    Publish a message to an MQTT topic.

    If the payload is not a string, it is serialized to JSON.
    This method blocks until the message is published.

    Args:
        topic (str): The MQTT topic to publish to.
        payload (Any): The message payload.
        retain (bool): Whether to retain the message. Defaults to False.

    Returns:
        mqtt.MQTTMessageInfo: Information about the published message.
    """
    if not isinstance(payload, str):
        payload = json.dumps(payload)

    info = self.client.publish(topic, payload, retain=retain)
    info.wait_for_publish()
    return info