cremalink.parsing.monitor package

This package handles the parsing and decoding of the device’s ‘monitor’ data.

The monitor data is a compact binary payload that represents the real-time status of the coffee machine, including its current state, any active alarms, and the progress of ongoing actions. This package provides the tools to decode this binary data into a structured and human-readable format.

class cremalink.parsing.monitor.MonitorFrame(direction: int, request_id: int, answer_required: int, accessory: int, switches: bytes, alarms: bytes, status: int, action: int, progress: int, timestamp: bytes, extra: bytes, raw: bytes, raw_b64: str)[source]

Bases: object

Represents the decoded, low-level structure of a monitor data frame.

This class takes the raw bytes from a monitor update and parses them into their fundamental components according to the device’s binary protocol. This includes separating headers, payload, checksums, and timestamps.

accessory: int

action: int

alarms: bytes

answer_required: int

as_dict() → dict[source]: Returns a dictionary representation of the frame’s contents. This is useful for serialization or debugging.

direction: int

extra: bytes

classmethod from_b64(raw_b64: str) → MonitorFrame[source]

Decodes a base64 string into a structured MonitorFrame.

This factory method performs the primary byte-level parsing, including: - Base64 decoding. - Length validation. - CRC-16 checksum verification. - Splitting the raw bytes into their respective fields (header, payload, etc.).

Parameters:: raw_b64 – The base64-encoded monitor data string.
Returns:: A populated MonitorFrame instance.
Raises:: ValueError – If the data is malformed, too short, or fails the CRC check.

progress: int

raw: bytes

raw_b64: str

request_id: int

status: int

switches: bytes

timestamp: bytes

class cremalink.parsing.monitor.MonitorProfile(flags: Dict[str, ~cremalink.parsing.monitor.profile.FlagDefinition]=<factory>, enums: Dict[str, ~typing.Dict[int, str]]=<factory>, predicates: Dict[str, ~cremalink.parsing.monitor.profile.PredicateDefinition]=<factory>)[source]

Bases: object

A complete profile for parsing a device’s monitor frame.

This class holds all the definitions needed to translate the raw bytes of a monitor frame into meaningful, human-readable data. It is typically loaded from a device-specific JSON file.

available_fields() → list[str][source]: Returns a sorted list of all defined flag and predicate names.

enums: Dict[str, Dict[int, str]]

flags: Dict[str, FlagDefinition]

classmethod from_dict(data: dict | None) → MonitorProfile[source]: Constructs a MonitorProfile from a dictionary (e.g., from a JSON file).

predicates: Dict[str, PredicateDefinition]

summary() → dict[str, Any][source]: Provides a brief summary of the profile’s contents.

class cremalink.parsing.monitor.MonitorSnapshot(raw: bytes, raw_b64: str, received_at: datetime, parsed: dict[str, ~typing.Any]=<factory>, warnings: list[str] = <factory>, errors: list[str] = <factory>, source: str = 'local', device_id: str | None = None, frame: MonitorFrame | None = None)[source]

Bases: object

Represents a single snapshot of the device’s monitoring status.

This dataclass acts as a container for all information related to a single monitor update from the device. It holds the raw data, timestamps, any parsed values, and metadata about the decoding process.

raw

The raw bytes of the monitor data payload.

Type:: bytes

raw_b64

The base64-encoded string representation of the raw data.

Type:: str

received_at

The timestamp when this snapshot was received.

Type:: datetime.datetime

parsed

A dictionary to hold the successfully parsed key-value data.

Type:: dict[str, Any]

warnings

A list of any warnings generated during parsing.

Type:: list[str]

errors

A list of any errors encountered during parsing.

Type:: list[str]

source

The origin of the data (e.g., ‘local’ or ‘cloud’).

Type:: str

device_id

The identifier of the device that sent the data.

Type:: str | None

frame

A MonitorFrame instance if the raw bytes were successfully decoded into a low-level frame structure.

Type:: cremalink.parsing.monitor.frame.MonitorFrame | None

device_id: str | None = None

errors: list[str]

frame: MonitorFrame | None = None

parsed: dict[str, Any]

raw: bytes

raw_b64: str

received_at: datetime

source: str = 'local'

warnings: list[str]

class cremalink.parsing.monitor.MonitorView(snapshot: MonitorSnapshot, profile: MonitorProfile | dict[str, Any] | None = None)[source]

Bases: object

A user-friendly view of a MonitorSnapshot, powered by a MonitorProfile.

This class acts as a wrapper around a MonitorSnapshot. It uses a given MonitorProfile to translate raw, low-level data (like integer codes and bit flags) into human-readable values (like enum names and boolean predicates).

It provides dynamic attribute access (__getattr__) to resolve flags and predicates from the profile on the fly. For example, view.is_on or view.has_descaling_alarm.

property accessory_code: int | None: The raw integer code for the currently detected accessory.

property accessory_name: str | None: The human-readable name of the accessory (e.g., ‘Milk Carafe’).

property action_code: int | None: The raw integer code for the device’s current action.

property action_name: str | None: The human-readable name of the device’s action (e.g., ‘Brewing’).

property available_fields: list[str]: A list of all available dynamic fields (flags and predicates).

property parsed: dict[str, Any]: The dictionary of initially parsed, low-level fields.

property profile_summary: dict[str, Any]: A summary of the loaded profile.

property progress_percent: int | None: The progress percentage (0-100) of the current action.

property raw: bytes: The raw bytes of the monitor payload.

property raw_b64: str: The base64-encoded string of the monitor payload.

property received_at: The timestamp when the snapshot was received.

property status_code: int | None: The raw integer code for the device’s main status.

property status_name: str | None: The human-readable name of the device’s status (e.g., ‘Standby’).

cremalink.parsing.monitor.build_monitor_snapshot(payload: dict[str, Any], source: str = 'local', device_id: str | None = None) → MonitorSnapshot[source]

Constructs a MonitorSnapshot from a raw payload dictionary.

This function orchestrates the decoding process: 1. It extracts the base64-encoded monitor string from the input payload. 2. It decodes the base64 string into raw bytes. 3. It calls extract_fields_from_b64 to parse the raw bytes into a

low-level dictionary and a MonitorFrame.

It bundles all this information into a MonitorSnapshot object.

Parameters:

payload – The raw dictionary payload, typically from the local server or cloud API.
source – The origin of the data (e.g., ‘local’, ‘cloud’).
device_id – The identifier of the device.

Returns:

A populated MonitorSnapshot instance.

cremalink.parsing.monitor.decode_monitor_b64(raw_b64: str) → bytes[source]: A simple wrapper for base64 decoding that provides a more specific error message.

cremalink.parsing.monitor package

Submodules