cremalink.domain.device module

This module defines the core Device class, which represents a physical coffee machine. It serves as the primary high-level interface for interacting with a device, abstracting away the underlying transport and command details.

class cremalink.domain.device.Device(transport: DeviceTransport, dsn: str | None = None, model: str | None = None, nickname: str | None = None, ip: str | None = None, lan_key: str | None = None, scheme: str | None = None, is_online: bool | None = None, last_seen: str | None = None, firmware: str | None = None, serial: str | None = None, coffee_count: int | None = None, command_map: Dict[str, ~typing.Any]=<factory>, property_map: Dict[str, ~typing.Any]=<factory>, monitor_profile: MonitorProfile = <factory>, extra: dict[str, ~typing.Any]=<factory>)[source]

Bases: object

Represents a coffee machine, providing methods to control and monitor it.

This class holds the device’s state (e.g., IP, model) and uses a DeviceTransport object to handle the actual communication. Device-specific capabilities are loaded from a “device map” file.

transport

The transport object responsible for communication.

Type:: cremalink.transports.base.DeviceTransport

dsn

Device Serial Number.

Type:: str | None

model

The model identifier of the device.

Type:: str | None

nickname

A user-defined name for the device.

Type:: str | None

ip

The local IP address of the device.

Type:: str | None

lan_key

The key used for LAN-based authentication.

Type:: str | None

scheme

The communication scheme (e.g., ‘http’, ‘mqtt’).

Type:: str | None

is_online

Boolean indicating if the device is currently reachable.

Type:: bool | None

last_seen

Timestamp of the last communication.

Type:: str | None

firmware

The device’s firmware version.

Type:: str | None

serial

The device’s serial number.

Type:: str | None

coffee_count

The total number of coffees made.

Type:: int | None

command_map

A mapping of command aliases to their hex codes.

Type:: Dict[str, Any]

property_map

A mapping of property aliases to their technical names.

Type:: Dict[str, Any]

monitor_profile

Configuration for parsing monitor data.

Type:: cremalink.parsing.monitor.profile.MonitorProfile

extra

A dictionary for any other miscellaneous data.

Type:: dict[str, Any]

coffee_count: int | None = None

command_map: Dict[str, Any]

configure() → None[source]: Configures the underlying transport.

do(drink_name: str) → Any[source]

Executes a command by its friendly name (e.g., ‘espresso’).

Parameters:: drink_name – The name of the command to execute, as defined in the command_map.
Returns:: The response from the transport.
Raises:: ValueError – If the command name is not found in the device map.

dsn: str | None = None

extra: dict[str, Any]

firmware: str | None = None

classmethod from_map(transport: DeviceTransport, device_map_path: str | None = None, **kwargs) → Device[source]

Factory method to create a Device instance with a loaded device map.

If device_map_path is not provided, it attempts to find one using the device’s model.

Parameters:

transport – The communication transport to use.
device_map_path – Optional path to the device map JSON file.
**kwargs – Additional attributes to set on the Device instance.

Returns:

A configured Device instance.

get_commands() → list[str][source]: Returns a list of all available command names from the device map.

get_monitor() → MonitorView[source]

Retrieves and parses monitoring data into a structured, human-readable view.

Returns:: A MonitorView instance containing parsed status information.

get_monitor_frame() → MonitorFrame | None[source]

Decodes the raw monitor data into a MonitorFrame for low-level analysis.

Returns:: A MonitorFrame if decoding is successful, otherwise None.

get_monitor_snapshot() → MonitorSnapshot[source]: Retrieves the latest raw monitoring data from the transport.

get_properties() → Any[source]: Fetches all available properties from the device.

get_property(name: str) → Any[source]

Fetches a single property by its alias or technical name.

Parameters:: name – The alias or name of the property to fetch.
Returns:: The value of the requested property.

get_property_aliases() → list[str][source]: Returns a list of all available property aliases from the device map.

health() → Any[source]: Checks the health of the device connection.

ip: str | None = None

is_online: bool | None = None

lan_key: str | None = None

last_seen: str | None = None

model: str | None = None

monitor_profile: MonitorProfile

nickname: str | None = None

property_map: Dict[str, Any]

refresh_monitor() → Any[source]: Requests a refresh of the device’s monitoring data.

resolve_property(alias: str, default: str | None = None) → str[source]

Translates a property alias to its technical name using the property_map.

Parameters:

alias – The property alias to resolve.
default – A default value to return if the alias is not found.

Returns:

The resolved technical name, or the alias/default if not found.

scheme: str | None = None

send_command(command: str) → Any[source]

Encodes and sends a raw hex command to the device via the transport.

Parameters:: command – The hex command string to send.
Returns:: The response from the transport.

serial: str | None = None

transport: DeviceTransport