Instalação

Instale a biblioteca via PIP:

pip install cd3_connector

Caso já tenha a biblioteca instalada, atualize via PIP:

pip install --upgrade cd3_connector

Versões do Python suportadas: 3.8, 3.9, 3.10, 3.11, 3.12, 3.13.

Como usar

1. Inicialize o Connector

O primeiro passo é criar uma instância da classe CD3Connector, passando o nome de usuário, senha e funções de callback para lidar com eventos de conexão, desconexão e mensagens recebidas.

import cd3_connector
class Receiver():
	def __init__(self, user:str, password:str):
		self._conn = cd3_connector.CD3Connector(user, password, self._on_disconnect,self._on_message, self._on_connect)
		self._conn.start() # Inicia a conexão

Além dos parâmetros obrigatórios, o CD3Connector permite configurar a inicialização com os seguintes parâmetros opcionais:

active_conflated: (bool, padrão: False)
- Define se o modo conflated deve ser ativado, permitindo uma menor frequência de atualizações.
log_level: (int, padrão: logging.INFO)
- Define o nível de detalhe do log. Os valores podem ser:
  - logging.DEBUG para maior detalhe;
  - logging.INFO para informações gerais;
  - logging.WARNING ou logging.CRITICAL para menos detalhes
log_path: (str, padrão: None)
- Caminho para salvar os arquivos de log. Se não for fornecido, os logs serão criados em uma pasta padrão chamada cd3_connector no diretório atual.
active_realtime: (bool, padrão: False)
- Define se o modo realtime deve ser ativado, permitindo receber atualizações em tempo real dos servidores datafeed. Por padrão a informação é recebida dos servidores delay.

Exemplo de inicialização com active_conflated, log_level e log_path:

import cd3_connector
import logging

	class Receiver():
		def __init__(self, user:str, password:str):
			self._conn = cd3_connector.CD3Connector(user, password, self._on_disconnect, self._on_message, self._on_connect, active_conflated=True, log_level=logging.INFO, log_path="C:\\logs_cd3connector")
			self._conn.start() # Inicia a conexão

Exemplo de inicialização com active_conflated e log_path:

import cd3_connector

	class Receiver():
		def __init__(self, user:str, password:str):
			self._conn = cd3_connector.CD3Connector(user, password, self._on_disconnect, self._on_message, self._on_connect, active_conflated=True, log_path="C:\\logs_cd3connector")
			self._conn.start() # Inicia a conexão

Exemplo de inicialização com log_level e log_path:

import cd3_connector
import logging

	class Receiver():
		def __init__(self, user:str, password:str):
			self._conn = cd3_connector.CD3Connector(user, password, self._on_disconnect, self._on_message, self._on_connect, log_level=logging.INFO, log_path="C:\\logs_cd3connector")
			self._conn.start() # Inicia a conexão

Exemplo de inicialização com active_realtime:

import cd3_connector
import logging
 
	class Receiver():
		def __init__(self, user:str, password:str):
			self._conn = cd3_connector.CD3Connector(user, password, self._on_disconnect, self._on_message, self._on_connect, active_realtime=True)
			self._conn.start() # Inicia a conexão

2. Métodos principais

start(): Inicia a conexão com o servidor.
stop(): Finaliza a conexão com o servidor.
send_command(message:str): Envia um comando ao servidor

3. Enviar comandos

Com a conexão estabelecida, você pode enviar comandos ao servidor usando o método SendCommand. Cada comando deve ser enviado como uma string.

self._conn.send_command("SQT PETR4") # Solicita cotação de PETR4
self._conn.send_command("BQT PETR4") # Solicita book de ofertas de PETR4

4. Callbacks

Você deve definir funções de callback para tratar eventos de conexão, desconexão e para processar as mensagens recebidas. São elas:

on_connect: Invocada quando a conexão é estabelecida com sucesso.
on_disconnect: Invocada quando a conexão é perdida.
on_message: Invocada sempre que uma nova mensagem é recebida.

Exemplo de uma função de callback para receber mensagens:

def _on_message(self, msg: str):
	if msg.lower() == "invalid login.":
		print("Login inválido. Verifique suas credenciais.")
	else:
		print(f"Mensagem recebida: {msg}")

❗️
Informação importante
Durante o uso da biblioteca, pode ocorrer o seguinte erro:
RuntimeError: This CD3Connector instance has been destroyed and cannot be used.
Esse erro ocorre quando uma ação tenta interagir com uma instância do CD3Connector que foi destruída por meio de métodos internos, após o uso indevido de alguma funcionalidade. Uma instância destruída invalida todos os seus recursos e impede que ela seja reutilizada. Sempre que um RuntimeError desse tipo for levantado, é necessário criar uma nova instância do conector.
A biblioteca também registra uma mensagem de log quando a instância é
destruída.

Exemplos

Exemplo completo 1

import threading
from queue import Queue
import cd3_connector
class Receiver():
    def __init__(self, user:str, password:str):
        self._conn = cd3_connector.CD3Connector(user, password, self._on_disconnect, self._on_message, self._on_connect)
        self._queue = Queue()
        self._consumer = threading.Thread(target=self._process_messages)
        self._signal = threading.Event()
        self._restart_conn = True
        self._consumer.start()

    def start(self):
        self._conn.start()


    def join(self):
        self._signal.set()
        self._consumer.join()


    def _on_connect(self):
        self._conn.send_command("sqt petr4")
        self._conn.send_command("bqt petr4")


    def _on_disconnect(self):
        if self._restart_conn:
            self._conn.start()


    def _on_message(self, msg:str):
        self._queue.put(msg)


    def _process_messages(self):
        while not self._signal.is_set():
            msg = self._queue.get()
            print(msg)
            self._queue.task_done()
            if (msg.lower() == "invalid login.") or (msg.lower() == "software key not found.") or (msg.lower() == "you don't have any permission for this software."):
                self._restart_conn = False
                return


def main(): 
    cd3_conn = Receiver("username", "password")
    cd3_conn.start()
    cd3_conn.join()
    
if __name__ == "__main__":
    main()

Exemplo completo 2

import threading
from queue import Queue
import cd3_connector
import logging

class Receiver():
    def __init__(self, user:str, password:str):
        self._conn = cd3_connector.CD3Connector(user, password, self._on_disconnect, self._on_message, self._on_connect, active_conflated=True, log_level=logging.INFO, log_path="C:\\logs_cd3connector", active_realtime=True)
        self._queue = Queue()
        self._consumer = threading.Thread(target=self._process_messages)
        self._signal = threading.Event()
        self._restart_conn = True
        self._consumer.start()

    def start(self):
        self._conn.start()


    def join(self):
        self._signal.set()
        self._consumer.join()


    def _on_connect(self):
        self._conn.send_command("sqt petr4")
        self._conn.send_command("bqt petr4")


    def _on_disconnect(self):
        if self._restart_conn:
            self._conn.start()


    def _on_message(self, msg:str):
        self._queue.put(msg)


    def _process_messages(self):
        while not self._signal.is_set():
            msg = self._queue.get()
            print(msg)
            self._queue.task_done()
            if (msg.lower() == "invalid login.") or (msg.lower() == "software key not found.") or (msg.lower() == "you don't have any permission for this software."):
                self._restart_conn = False
                return


def main(): 
    cd3_conn = Receiver("username", "password")
    cd3_conn.start()
    cd3_conn.join()
    
if __name__ == "__main__":
    main()