API Reference

class pycmcontrol_mqtt.client.BrokerTLS(ca_certs=None, certfile=None, keyfile=None, insecure=False)

Base: object

Config opcional de TLS para MQTT. Se seu broker usa 8883/TLS, configure aqui.

• ca_certs: caminho do CA (opcional)

• certfile/keyfile: client cert (opcional)

• insecure: True = não valida hostname/cert (evite em produção)

ca_certs: str | None = None

certfile: str | None = None

insecure: bool = False

keyfile: str | None = None

class pycmcontrol_mqtt.client.CmControlClient(cfg, *, tls=None, connect_keepalive_s=60, request_timeout_s_default=10.0, strict_business_errors=True, business_error_prefixes=('ERRO',), business_error_contains=('FALHA', 'NOK'), business_ok_prefixes=('ERRO4',), debug=False)

Base: object

Cliente MQTT para CmControl Driver v1.00 e MQTT+REST, conforme doc.

Regras da doc: - REQUEST: br/com/cmcontrol/dispositivo/{device}/set/{endpoint} - RESPONSE: br/com/cmcontrol/dispositivo/{device}/get/{endpoint} - QoS = 0 - Retained = false - Dispositivo deve ficar inscrito em: …/get/+ - Se o sistema enviar /get/ping -> responder /set/pong {timestamp} - Se o sistema enviar /get/state -> responder /set/state {state:”1”}

Recursos: - request(): publica /set/{endpoint} e aguarda /get/{endpoint} com cache por tópico - login OAuth2 via MQTT+REST - setup.apontamento via MQTT+REST - erros claros (rede, mqtt auth, status != 200, timeout de resposta, etc)

apontar_lote_1porreq(seriais, *, timeout_s=None, delay_s=0.2, stop_on_error=False)

Faz N requests, 1 serial por payload (ideal para seu caso).

stop_on_error=True: para no primeiro erro de apontamento.

Tipo de retorno:

list[Dict[str, Any]]

apontar_serial(serial, *, timeout_s=None, evidencias=None)

Apontamento simples serializado (1 serial por request) — recomendado no seu cenário.

{

“enderecoDispositivo”:”device001”, “apontamentos”:[

{ “ok”: true, “seriais”:[{“codigo”:”…”}] }

]

}

Tipo de retorno:

Dict[str, Any]

base_topic()

Tipo de retorno:

str

connect()

Conecta ao broker MQTT e aguarda on_connect.

Converte erros de rede/socket/TLS em exceções próprias da lib.

Tipo de retorno:

None

disconnect()

Tipo de retorno:

None

ensure_login(timeout_s=None)

Tipo de retorno:

None

is_token_valid()

Tipo de retorno:

bool

last_exchange()

Tipo de retorno:

Dict[str, Any]

login_oauth2(timeout_s=None)

Doc (Autenticação / MQTT+REST): REQUEST: …/set/rest/oauth2/login RESPONSE: …/get/rest/oauth2/login

Tipo de retorno:

str

REQUEST payload: {

“request”: {

“headers”: { “Authorization”: “Basic <base64>” }, “type”: “GET”

}

}

RESPONSE: {

“status”:”200”, “log”:”…”, “access_token”:”…”, “token_type”:”Bearer”, “expires_in”:86400

}

logout_oauth2(timeout_s=None)

Tipo de retorno:

Dict[str, Any]

ordem_transporte(codigo, *, acao='APONTAR_TRANSPORTE', timeout_s=None, apontamentos=None)

Tipo de retorno:

Dict[str, Any]

ping(timeout_s=None)

Tipo de retorno:

Dict[str, Any]

Dispositivo -> Sistema:

SET …/set/ping { “timestamp”: <int> }

Sistema -> Dispositivo:

GET …/get/pong

publish_set(endpoint, payload)

Publica REQUEST em /set/{endpoint} com QoS=0 e retained=False (doc).

Tipo de retorno:

None

request(endpoint, payload, timeout_s=None)

Tipo de retorno:

Dict[str, Any]

setup_apontamento(setup, timeout_s=None)

Doc: /api/v1/setup.apontamento via MQTT+REST REQUEST: …/set/rest/api/v1/setup.apontamento RESPONSE: …/get/rest/api/v1/setup.apontamento

Tipo de retorno:

Dict[str, Any]

MQTT+REST payload: {

“request”: {

“headers”: {…}, “type”: “POST”, “params”: {…} # opcional

}, “data”: { … } # JSON Setup

}

token()

Tipo de retorno:

Optional[str]

topic_get(endpoint)

Tipo de retorno:

str

topic_set(endpoint)

Tipo de retorno:

str

validar_rota(serial, *, timeout_s=None)

Usa ciclo VALIDAR_ROTA conforme doc Setup.

Tipo de retorno:

Dict[str, Any]

class pycmcontrol_mqtt.config.CmControlConfig(device_addr, broker_host, broker_port=1883, mqtt_user='', mqtt_pass='', api_user='', api_pass='', connect_timeout_s=10, token_renew_margin_s=600)

Config de conexão com o broker e credenciais do CmControl (OAuth2 via MQTT+REST).

A doc diz: credenciais do broker vêm no CmControl em:

Menu Principal > Configurações > MQTT

classmethod from_env(env_path=None, prefix='CMC_')

Helper opcional (NÃO obrigatório). Para usar .env: pip install pycmcontrol_mqtt[env]

Tipo de retorno:

CmControlConfig

class pycmcontrol_mqtt.models.Apontamento(ok=True, serial=None, seriais_vinculados=None, evidencias=None)

Apontamento (doc ‘Apontamento’):

• ok: bool

• seriais: lista de Serial (quando serializado)

• evidencias: lista de Evidence (se obrigatório na operação)

class pycmcontrol_mqtt.models.Evidence(nome, extensao, conteudo, descricao='', observacao='')

Evidência (doc ‘Evidência’):

• nome, extensao, conteudo (base64)

• descricao (opcional)

• observacao (opcional)

class pycmcontrol_mqtt.models.OrdemTransporte(codigo, acao='APONTAR_TRANSPORTE')

OrdemTransporte (doc ‘OrdemTransporte’):

• codigo

• acao (default APONTAR_TRANSPORTE)

class pycmcontrol_mqtt.models.Serial(codigo)

class pycmcontrol_mqtt.models.SetupApontamento(enderecoDispositivo, apontamentos=None, ciclo='', ordemTransporte=None)

Setup (doc ‘Setup’):

• enderecoDispositivo (quando idSessao/idView forem nulos)

• ciclo (opcional): VALIDAR_ROTA etc.

• ordemTransporte (opcional)

• apontamentos: lista de Apontamento (obrigatório)

exception pycmcontrol_mqtt.errors.CmcApiError(status=None, log='', endpoint='', raw=None)

Erro genérico de chamadas REST via MQTT+REST.

exception pycmcontrol_mqtt.errors.CmcApontamentoError(status=None, log='', endpoint='', raw=None)

Erro de apontamento (setup.apontamento).

exception pycmcontrol_mqtt.errors.CmcConfigError

Configuração inválida (ex: faltou api_user/api_pass).

exception pycmcontrol_mqtt.errors.CmcConnectionError

Falha ao conectar no broker (DNS, timeout, firewall, rota, etc).

exception pycmcontrol_mqtt.errors.CmcConnectionTimeout

Timeout durante connect TCP/MQTT.

exception pycmcontrol_mqtt.errors.CmcDecodeError

Payload recebido não é JSON válido ou está corrompido.

exception pycmcontrol_mqtt.errors.CmcDisconnected

Conexão caiu durante uma operação.

exception pycmcontrol_mqtt.errors.CmcDnsError

Hostname não resolve (DNS).

exception pycmcontrol_mqtt.errors.CmcError

Erro base da biblioteca.

exception pycmcontrol_mqtt.errors.CmcInvalidArgument

Argumento inválido fornecido pelo usuário da biblioteca.

exception pycmcontrol_mqtt.errors.CmcLoginError(status=None, log='', endpoint='', raw=None)

Erro específico de login OAuth2.

exception pycmcontrol_mqtt.errors.CmcMqttAuthError(rc, message='')

Broker rejeitou usuário/senha (CONNACK rc != 0).

exception pycmcontrol_mqtt.errors.CmcMqttProtocolError

Erro de protocolo MQTT (versão, pacote inválido, etc).

exception pycmcontrol_mqtt.errors.CmcNotConnected

Operação requer conexão MQTT, mas connect() não foi chamado.

exception pycmcontrol_mqtt.errors.CmcResponseError(status=None, log='', endpoint='', raw=None)

Erro retornado pelo CmControl em uma resposta JSON. Ex.: status != 200, ou 200 com log indicando erro de negócio.

exception pycmcontrol_mqtt.errors.CmcTimeout

Timeout esperando resposta no tópico /get.

exception pycmcontrol_mqtt.errors.CmcTlsError

Erro de TLS/SSL (certificado, handshake, etc).