cremalink package

Cremalink: A Python library for interacting with De’Longhi coffee machines.

This top-level package exposes the primary user-facing classes and functions for easy access, including the main Client, the Device model, and factory functions for creating device instances.

class cremalink.Client(token_path: str)[source]

Bases: object

Client for interacting with the Ayla IoT cloud platform. Manages authentication (access and refresh tokens) and device discovery.

get_device(dsn: str, device_map_path: dict | None = None)[source]

Retrieves a specific cloud device by its DSN.

Parameters:

dsn (str) – The Device Serial Number of the desired device.
device_map_path (dict | None) – Optional mapping for device properties.

Returns:

An instance of CloudDevice if found, otherwise None.

Return type:

CloudDevice | None

get_devices()[source]

Retrieves a list of Device Serial Numbers (DSNs) for all registered devices.

Returns:: A list of DSNs.
Return type:: list[str]

get_lan_config(dsn: str) → dict[source]

Fetch LAN connectivity details for a device (FR-006).

Ported from delonghi_coffee’s api.get_lan_config — same Ayla platform, same account. Never raises just because LAN is disabled; returns lan_enabled: False with the other fields None instead (FR-008).

get_serial_number(dsn: str) → str | None[source]

Fetch the raw d270_serialnumber property for a device.

Used as an input to model detection (FR-004); returns None if the property is absent rather than raising, so callers can safely fall through to the next detection tier.

list_account_devices() → list[dict][source]

Discover coffee-machine devices on the account (cloud-assisted onboarding).

Unlike get_devices() (kept unchanged for backward compatibility), this re-fetches /devices.json and returns one enriched dict per coffee-machine device only — non-coffee appliances (e.g. a Pinguino A/C on the same account) are excluded, not just flagged (FR-002).

Returns:: Each with dsn, product_name, oem_model, lan_enabled, connection_status.
Return type:: list[dict]

class cremalink.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

class cremalink.LocalServer(settings: ServerSettings)[source]

Bases: object

A wrapper class that encapsulates the Uvicorn server and the web application.

start() → None[source]: Starts the Uvicorn server to serve the application.

class cremalink.ServerSettings(_case_sensitive: bool | None = None, _nested_model_default_partial_update: bool | None = None, _env_prefix: str | None = None, _env_prefix_target: EnvPrefixTarget | None = None, _env_file: DotenvType | None = PosixPath('.'), _env_file_encoding: str | None = None, _env_ignore_empty: bool | None = None, _env_nested_delimiter: str | None = None, _env_nested_max_split: int | None = None, _env_parse_none_str: str | None = None, _env_parse_enums: bool | None = None, _cli_prog_name: str | None = None, _cli_parse_args: bool | list[str] | tuple[str, ...] | None = None, _cli_settings_source: CliSettingsSource[Any] | None = None, _cli_parse_none_str: str | None = None, _cli_hide_none_type: bool | None = None, _cli_avoid_json: bool | None = None, _cli_enforce_required: bool | None = None, _cli_use_class_docs_for_groups: bool | None = None, _cli_show_env_vars: bool | None = None, _cli_exit_on_error: bool | None = None, _cli_prefix: str | None = None, _cli_flag_prefix_char: str | None = None, _cli_implicit_flags: bool | Literal['dual', 'toggle'] | None = None, _cli_ignore_unknown_args: bool | None = None, _cli_kebab_case: bool | Literal['all', 'no_enums'] | None = None, _cli_shortcuts: Mapping[str, str | list[str]] | None = None, _secrets_dir: PathType | None = None, _build_sources: tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None = None, *, SERVER_SETTINGS_PATH: str = '', SERVER_IP: str = '127.0.0.1', SERVER_PORT: int = 10280, ADVERTISED_IP: str | None = None, NUDGER_POLL_INTERVAL: float = 1.0, MONITOR_POLL_INTERVAL: float = 5.0, REKEY_INTERVAL_SECONDS: float = 60.0, QUEUE_MAX_SIZE: int = 200, LOG_RING_SIZE: int = 200, DEVICE_REGISTER_VERIFY: bool = False, DEVICE_REGISTER_CA_PATH: str | None = None, DEVICE_REGISTER_TIMEOUT: float = 10.0, ENABLE_DEVICE_REGISTER: bool = True, ENABLE_NUDGER_JOB: bool = True, ENABLE_MONITOR_JOB: bool = True, ENABLE_REKEY_JOB: bool = True, FIXED_RANDOM_2: str | None = None, FIXED_TIME_2: str | None = None)[source]

Bases: BaseSettings

Defines the application’s settings, which can be loaded from environment variables or a .env file. This provides a centralized and type-safe way to configure the server’s behavior.

advertised_ip: str | None

device_register_ca_path: str | None

device_register_timeout: float

device_register_verify: bool

enable_device_register: bool

enable_monitor_job: bool

enable_nudger_job: bool

enable_rekey_job: bool

fixed_random_2: str | None

fixed_time_2: str | None

log_ring_size: int

model_config = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_show_env_vars': False, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': '', 'env_prefix_target': 'variable', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'populate_by_name': True, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_by_alias': True, 'validate_by_name': True, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

monitor_poll_interval: float

nudger_poll_interval: float

queue_max_size: int

rekey_interval_seconds: float

server_ip: str

server_port: int

server_settings_path: str

cremalink.authenticate_cloud(email: str, password: str, language: str = 'en') → CloudToken[source]

Performs the multi-step authentication flow to obtain a refresh token for the Ayla IoT cloud platform.

This function executes the following steps: 1. Authorize: Initiates the authorization process to obtain a context parameter. 2. Get IDs: Retrieves necessary IDs (ucid, gmid, gmid_ticket) for the login process. 3. Login: Authenticates the user with their email and password, along with risk context information, to obtain a login token. 4. Get User Info: Fetches user information using the login token to obtain UID, UIDSignature, and signatureTimestamp. 5. Consent: Simulates the user consent process to obtain a signature for the authorization continue step. 6. Authorization Continue: Continues the authorization process using the context, login token, and consent signature to obtain an authorization code. 7. IDP Token: Exchanges the authorization code for an IDP token. 8. Exchange for Ayla: Uses the IDP token to sign in to the Ayla cloud and obtain a refresh token.

Parameters:

email (str) – The user’s email address for authentication.
password (str) – The user’s password for authentication.
language (str) – The preferred language for the authentication process (default is “en”).

Returns:

An instance of CloudToken containing the obtained refresh token.

Return type:

CloudToken

cremalink.create_app(settings: ServerSettings | None = None, device_adapter: DeviceAdapter | None = None, logger=None) → FastAPI[source]

Application factory for the FastAPI server.

Initializes all components (state, settings, adapter, jobs) and wires up the API routes, startup/shutdown events, and dependencies.

Returns:: A configured FastAPI application instance.

cremalink.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.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.

cremalink.detect_model_id(raw_serial: str | None, cloud_metadata: dict[str, Any] | None, oem_model: str | None) → str | None[source]

Resolve a discovered device to an existing cremalink device-map id.

Parameters:

raw_serial – The device’s raw serial number string, if known.
cloud_metadata – Additional cloud-reported fields (e.g. product_name, model), if any.
oem_model – The Ayla oem_model/model identifier, if known.

Returns:

A valid cremalink device-map id, or None if no detection tier yielded a recognized model. Never returns a guessed or default id.

cremalink.device_map(model_id: str) → str[source]

Finds the absolute path to a device map file for a given model ID.

This function handles packaged resources, extracting them to a temporary directory if they are not directly accessible on the filesystem.

Parameters:: model_id – The model ID of the device.
Returns:: The absolute path to the device map JSON file as a string.
Raises:: DeviceMapNotFoundError – If the map for the given model ID doesn’t exist.

Subpackages

Submodules

cremalink.local_server module
- LocalServer
  - LocalServer.start()
- main()