cremalink.domain package

This package defines the core domain models for the cremalink library, representing the main entities like the coffee machine itself.

It exposes the primary Device class and factory functions for creating device instances, abstracting away the underlying implementation details.

class cremalink.domain.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
cremalink.domain.create_cloud_device(dsn: str, access_token: str, device_map_path: str | None = None) → Device[source]

Creates a Device instance configured for cloud-based communication.

This factory sets up a CloudTransport, which communicates with the device via the manufacturer’s cloud services, requiring an access token.

Parameters:

  • dsn – The Device Serial Number.

  • access_token – The authentication token for the cloud service.

  • device_map_path – Optional path to a device-specific command map file.

Returns:

A Device instance configured with a CloudTransport.

cremalink.domain.create_local_device(dsn: str, lan_key: str, device_ip: str | None, server_host: str, server_port: int = 10280, device_scheme: str = 'http', auto_configure: bool = True, device_map_path: str | None = None) → Device[source]

Creates a Device instance configured for local network communication.

This factory sets up a LocalTransport which requires details about the local network environment, including the device’s IP and the local server’s address.

Parameters:

  • dsn – The Device Serial Number.

  • lan_key – The key for authenticating on the local network.

  • device_ip – The IP address of the coffee machine.

  • server_host – The hostname or IP address of the local proxy server.

  • server_port – The port of the local proxy server.

  • device_scheme – The communication protocol to use with the device (e.g., ‘http’).

  • auto_configure – If True, the transport will be configured automatically.

  • device_map_path – Optional path to a device-specific command map file.

Returns:

A Device instance configured with a LocalTransport.

Submodules