campbellcontrol.commands package

Submodules

campbellcontrol.commands.commands module

Module for defining MQTT commands, bulding payloads to send commands, and defining callbacks for handling responses from the logger.

class campbellcontrol.commands.commands.Command(group_id: str, client_id: str, options: dict | None = {})

Bases: ABC

Base class for defining a Campbell logger MQTT command. This class helps to define relevant topics used to interface with the command, generation of a payload that must be sent with the command, and provides a method to handle the reponse.

command_name: str: Name of the Campbell command used to build topics.

publish_topic: str: The topic used to send the command to a logger.

response_topic: str: The topic that response data is typically sent through

state_topic: str: The topic that supplimental data regarding the state is sent through, useful for logging the changes occuring on the logger rather than only the final response

abstractmethod payload(**kwargs) → Any: Return the payload used to send the command.

handle_state(*args, **kwargs) → Any: Return a payload only if it matches specific messages. Used for handling responses that are made on the state topic

json_payload(*args, **kwargs) → str

Jsonified payload string.

Returns:: A JSON formatted string payload.

handler(topic: str, message: str) → CommandResponse | None

Handler for messages that have either a ‘success’ or ‘error’ value.

Parameters:

topic – The topic that the message is received from.
message – The received message.

class campbellcontrol.commands.commands.OS(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to download and install an operating system.

command_name: str = 'OS': Name of the Campbell command used to build topics.

static payload(url: str) → URLPayload

Return the payload for the OS command.

Returns:: A payload dictionary

class campbellcontrol.commands.commands.Program(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to download a CRBasic Program. The downloaded file is set to the current program and reboots the logger.

If the download fails, the response is sent on state, we see two messages:

{“clientId”:”ABC”,
“state”:”online”, “fileTransfer”:”CRBasic file transfer started”}

{“clientId”:”ABC”,
“state”:”online”, “fileTransfer”:”CRBasic file transfer error”}

command_name: str = 'program': Name of the Campbell command used to build topics.

static payload(url: str, filename: str) → FileDownloadPayload

Return the payload for the Program command.

Parameters:

url – Url to a Campbell program.
filename – Name to assign to the download on the logger.

Returns:

A dictionary payload

handle_state(message: dict) → dict: Accepts the message on a state topic. If it matches specific message strings, return it as a response

class campbellcontrol.commands.commands.MQTTConfig(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to reconfigure MQTT settings. If the file is valid, the settings are applied and the logger reboots.

command_name: str = 'mqttConfig': Name of the Campbell command used to build topics.

static payload(url: str) → URLPayload

Return the payload for the MQTTConfig command.

Parameters:: url – Url to a Campbell proprietary binary formatted settings file.
Returns:: A dictionary payload

class campbellcontrol.commands.commands.EditConstants(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to edit constants in a CRBasic program. Values are converted to correct types by the logger.

command_name: str = 'editConst': Name of the Campbell command used to build topics.

static payload(*args, **kwargs) → Dict[str, str]

Return the payload for the EditConstants command.

Parameters:

*args – Any quantity of dictionaries.
**kwargs – Keyword arguments specifying constants and the new value.

Returns:

“value”} pairs.

Return type:

A dictionary of key value {“constant_name”

class campbellcontrol.commands.commands.Reboot(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to reboot the logger.

command_name: str = 'reboot': Name of the Campbell command used to build topics.

static payload() → ActionPayload

Return the payload for the Reboot command.

Returns:: A dictionary payload.

handle_state(message: dict) → dict: Accepts the message on a state topic. If it matches specific message strings, return it as a response

class campbellcontrol.commands.commands.ListFiles(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to list files in a drive.

command_name: str = 'fileControl': Name of the Campbell command used to build topics.

payload(drive: str | None = None) → FileListPayload

Build a payload for the file list command.

Parameters:: drive – An optional string specifying the drive to list.
Returns:: A dictionary payload.

static handler(topic: str, message: str) → CommandResponse | None

Handler for returning a list of files.

Parameters:

topic – The topic the message was received on.
message – The message contents in string format.

Returns:

A command reponse.

class campbellcontrol.commands.commands.DeleteFile(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to delete a file on the logger.

command_name: str = 'fileControl': Name of the Campbell command used to build topics.

payload(filename: str, drive: str | None = None) → FileActionPayload

Build a payload for the file deletion command.

Parameters:

filename – The file to delete.
drive – An optional string specifying the drive to target.

Returns:

A dictionary payload.

handle_state(message: dict) → dict: Accepts the message on a state topic. If it matches specific message strings, return it as a response

class campbellcontrol.commands.commands.StopProgram(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command that stops the currently running logger program.

command_name: str = 'fileControl': Name of the Campbell command used to build topics.

static payload() → ActionPayload

Build a payload for the StopProgram command.

Returns:: A dictionary payload.

class campbellcontrol.commands.commands.RunProgram(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to a run a logger program from a file located on the device.

command_name: str = 'fileControl': Name of the Campbell command used to build topics.

static payload(filename: str) → ActionPayload

Build a payload for the StopProgram command.

Parameters:: filename – Name of the file to run.
Returns:: A dictionary payload.

class campbellcontrol.commands.commands.SetSetting(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Set a setting in the logger

command_name: str = 'setting': Name of the Campbell command used to build topics.

payload(name: str, value: str, apply: bool = False) → SettingsSetPayload

Build a payload for the SetSetting command.

Parameters:

name – Name of the setting.
value – New value to set.
apply – Applies the change immediately (may cause a logger reboot).

Returns:

A dictionary payload.

handler(topic: str, message: str) → CommandResponse | None

Handler for messages in response to setting a setting

Parameters:

topic – The topic that the message is received from.
message – The received message.

class campbellcontrol.commands.commands.ApplySettings(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Apply changed logger settings

command_name: str = 'setting': Name of the Campbell command used to build topics.

payload() → SettingsApplyPayload

Build a payload for the ApplySettings command.

Returns:: A dictionary payload

class campbellcontrol.commands.commands.PublishSetting(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Publish the value of a setting.

command_name: str = 'setting': Name of the Campbell command used to build topics.

payload(name: str) → SettingsPublishPayload

Build a payload for publishing a setting.

Parameters:: name – Name of the setting to publish.
Returns:: A dictionary payload.

handler(topic: str, message: str) → CommandResponse | None

Handler for messages in response to setting, action is just to pass the response straight through

Parameters:

topic – The topic that the message is received from.
message – The received message.

class campbellcontrol.commands.commands.SetVar(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Set variables present in the logger script.

command_name: str = 'setVar': Name of the Campbell command used to build topics.

payload(name: str, value: str) → SetVarPayload

Build a ayload for setting a variable.

Parameters:

name – Name of the setting to change.
value – New value for the setting.

Returns:

A dictionary payload.

class campbellcontrol.commands.commands.GetVar(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Publish variables present in the logger script.

command_name: str = 'getVar': Name of the Campbell command used to build topics.

payload(name: str) → GetVarPayload

Build a payload for publishing a variable.

Parameters:: name – Name of the variable to publish.
Returns:: A dictionary payload.

handler(topic: str, message: str) → CommandResponse | None

Handler for messages that have either a ‘success’ or ‘error’ value.

Parameters:

topic – The topic that the message is received from.
message – The received message.

class campbellcontrol.commands.commands.HistoricData(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Retrieve historic data from a logger table

command_name: str = 'historicData': Name of the Campbell command used to build topics.

payload(table: str, start: str, end: str) → HistoricDataPayload

Payload for retrieving historic data from the logger.

Parameters:

table – Name of the table on the logger.
start – Start datetime to query.
end – End datetime to query.

Returns:

A dictionary payload

class campbellcontrol.commands.commands.TalkThru(group_id: str, client_id: str, options: dict | None = {})

Bases: Command

Command to send a TalkThru command and manage the TalkThru session.

command_name: str = 'talkThru': Name of the Campbell command used to build topics.

payload(com_port: str, out_string: str, num_tries: str | None = None, resp_delay: str | None = None, abort: bool | None = None) → TalkThruPayload

Build a payload for talking to a sensor.: This keeps a session open during which time the port doesn’t send any data and awaits further TalkThru commands. The session times out after 1 minute.

Parameters:

comPort – The COM port to talk through.
outString – String to send to the sensor.
numberTries – Number of attempts before failure.
respDelay – Time (milliseconds) to wait for response.
abort – Flag to end the TalkThru session.

Returns:

A dictionary payload.

static handler(topic: str, message: str) → CommandResponse | None

Handler for returning a talkThru response

Parameters:

topic – The message receive topic.
message – Value of the message.

Returns:

A command response or None.

campbellcontrol.commands.typing module

Module for storing payload types used in the Campbell MQTT API.

class campbellcontrol.commands.typing.CommandResponse

Bases: TypedDict

Response from the command handler.

payload: dict: The returned payload.

success: bool: Flag indicating success or failure.

error: str | None: The error message. Absent if not error.

class campbellcontrol.commands.typing.FileDownloadPayload

Bases: TypedDict

Payload for a file download.

url: str: A valid URL to the download.

filename: str: The file to download it to on the system.

class campbellcontrol.commands.typing.URLPayload

Bases: TypedDict

Payload for specifying a URL.

url: str: A valid URL.

class campbellcontrol.commands.typing.ActionPayload

Bases: TypedDict

Payload for specifying an action.

action: str: Name of an action.

class campbellcontrol.commands.typing.FileActionPayload

Bases: TypedDict

Payload for acting on a file.

action: str: Name of an action.

fileName: str: Name of the file to act upon.

drive: str | None: Optional key to the target drive.

class campbellcontrol.commands.typing.FileListPayload

Bases: TypedDict

Payload for listing files in a directory.

action: str: Name of the action taken.

drive: str | None: Optional key to the target drive.

class campbellcontrol.commands.typing.SettingsSetPayload

Bases: TypedDict

Payload for changing settings.

action: Literal['set']: The action to take (must be “set”).

name: str: Name of the setting to change.

value: str: New value for the setting.

apply: bool | None: Optional flag for applying the change immediately.

class campbellcontrol.commands.typing.SettingsPublishPayload

Bases: TypedDict

Payload for publishing setting values.

action: Literal['publish']: The action to take (must be “publish”).

name: str: Name of the setting to publish

class campbellcontrol.commands.typing.SettingsApplyPayload

Bases: TypedDict

Payload for applying Settings.

action: Literal['apply']: The action to take (must be “apply”).

apply: bool: Flag to apply settings.

class campbellcontrol.commands.typing.GetVarPayload

Bases: TypedDict

Payload for a getting a variable.

name: str: Name of the variable to publish.

class campbellcontrol.commands.typing.SetVarPayload

Bases: TypedDict

Payload for setting a variable.

name: str: Name of the variable to set.

value: str: New value of the variable.

class campbellcontrol.commands.typing.HistoricDataPayload

Bases: TypedDict

Payload for getting historic data.

table: str: The logger table to query.

start: str: Start date of the query.

end: str: End date of the query.

class campbellcontrol.commands.typing.TalkThruPayload

Bases: TypedDict

Payload for talking to a sensor.

comPort: str: The COM port to talk through.

outString: str: An ASCII string to send to the sensor.

numberTries: str | None: ASCII number string indicating number of attempts.

respDelay: str | None: ASCII number string (milliseconds) of time to wait for response.

abort: bool | None: Flag to abort TalkThru session.

campbellcontrol.commands package

Submodules

campbellcontrol.commands.commands module

campbellcontrol.commands.typing module

Module contents