QuecPython
O QuecPython é um ambiente de tempo de execução para executar código MicroPython e representa uma nova abordagem para o desenvolvimento de IoT.
MicroPython é código aberto
MicroPython está disponível para uso geral sob licença MIT. Muitos de seus módulos e bibliotecas também estão disponíveis sob licença MIT ou licença de código aberto semelhante. O MicroPython é gratuito para uso educacional e comercial. Qualquer pessoa pode usar e modificar o firmware. É um projeto de código aberto com código-fonte disponível no Github. Por ser um projeto de código aberto, o MicroPython está aberto a contribuições da comunidade. Se você estiver procurando por uma versão de firmware voltada para uma plataforma de hardware, microcontrolador, porta ou recurso específico, verifique o link oficial de download do MicroPython.
Foram realizados testes com o Firmware abaixo*** e foi constatado que o mesmo deu boot no Módulo FIBOCOM L610-GL, pelo menos se conectou rede CLARO nos testes locais. Mas tem outros recursos habilitados.
Segundo documento com o zip, suporta:
1. aLiYun
2. app_fota
3. atcmd
4. dataCall
5. fota
6. net
7. ntptime
8. pm
9. uio
10. ujson
11. umqtt
12. usocket
13. utime
14. cellLocator
15. wifilocator
16. log
17. sysbus
18. queue
19. sim
20. Pin
21. ExtInt
22. UART
23. SPI
24. IIC
25. Timer
26. RTC
27. LCD
28. KeyPad
29. Power
30. ADC
31. PowerKey
32. PWM
33. quecgnss
34. audio
35. _thread
36. ussl
37. SecureData
38. uwebsocket
EG912U SERIES | QuecPython
***Resources Download | QuecPython
Download | QuecPython
EG912U-GL
QPY_OCPU_V0001_EG912U_GLAC_FW.ZIP
Gravando
Vendo se QuecPyhon responde
Verificando se tem sinal da operadora Claro
Mandando um Script que roda MQTT (client)
Script MQTT
'''
@Author: Baron
@Date: 2020-04-24
@LastEditTime: 2021-05-25 17:06:08
@Description: example for module umqtt
@FilePath: example_mqtt_file.py
'''
'''
The following two global variables are required. You can modify the values of the following two global variables according to your actual projects.
The values of these two variables are printed before the user code is executed.
'''
import utime
import log
import net
import _thread
import checkNet
import dataCall
from umqtt import MQTTClient
PROJECT_NAME = "QuecPython_MQTT_example"
PROJECT_VERSION = "1.0.0"
checknet = checkNet.CheckNetwork(PROJECT_NAME, PROJECT_VERSION)
# Reclaim the thread resource through the status after calling MQTTClient.disconnect().
TaskEnable = True
# Set the log output level.
log.basicConfig(level=log.INFO)
mqtt_log = log.getLogger("MQTT")
# Encapsulate MQTT so it can support more custom logic.
class MqttClient():
'''
mqtt init
'''
# Note: The parameter reconn enables or disables the internal reconnection mechanism. Default value: True (enable).
# If you need to test or use the external reconnection mechanism, please refer to this example code below. Before testing, set reconn to False, otherwise, the internal reconnection mechanism will be used by default.
def __init__(self, clientid, server, port, user=None, password=None, keepalive=0, ssl=False, ssl_params={},
reconn=True):
self.__clientid = clientid
self.__pw = password
self.__server = server
self.__port = port
self.__uasename = user
self.__keepalive = keepalive
self.__ssl = ssl
self.__ssl_params = ssl_params
self.topic = None
self.qos = None
# Network status flag.
self.__nw_flag = True
# Create a mutex.
self.mp_lock = _thread.allocate_lock()
# Create a class to initialize the MQTT object.
self.client = MQTTClient(self.__clientid, self.__server, self.__port, self.__uasename, self.__pw,
keepalive=self.__keepalive, ssl=self.__ssl, ssl_params=self.__ssl_params,
reconn=reconn)
def connect(self):
'''
Connect to the MQTT server.
'''
self.client.connect()
# Register the callback function of network status. When the network status changes, the function will be called.
flag = dataCall.setCallback(self.nw_cb)
if flag != 0:
# The network callback registration failed.
raise Exception("Network callback registration failed")
def set_callback(self, sub_cb):
'''
Set the callback function of receiving messages.
'''
self.client.set_callback(sub_cb)
def error_register_cb(self, func):
'''
Set the callback function of receiving MQTT thread error occurrence.
'''
self.client.error_register_cb(func)
def subscribe(self, topic, qos=0):
'''
Subscribe to topics.
'''
self.topic = topic # Save the topic. Multiple topics can be saved by a list.
self.qos = qos # Save the QoS.
self.client.subscribe(topic, qos)
def publish(self, topic, msg, qos=0):
'''
Publish a message.
'''
self.client.publish(topic, msg, qos)
def disconnect(self):
'''
Disconnect from the MQTT server.
'''
global TaskEnable
# Close the monitoring thread of wait_msg.
TaskEnable = False
# Disconnect from the MQTT server and release the resources.
self.client.disconnect()
def reconnect(self):
'''
MQTT reconnection mechanism (The following example is for your reference only and you can adjust based on actual needs.)
Note: 1. If other services need to be restarted after the client reconnects to the server, determine whether to release the resources of the previous services before restarting the services.
2. This section needs to be added based on the actual business logic, and this example only covers the process that the client resubscribes to topics after reconnecting to the MQTT server.
'''
# Determine whether the lock has been acquired.
if self.mp_lock.locked():
return
self.mp_lock.acquire()
# Close the previous connection before reconnecting to release resources. Please note the differences between *MQTTClient.disconnect()* and *MQTTClient.close()*, where MQTTClient.close() only releases socket resources but *MQTTClient.disconnect()* releases resources including threads.
self.client.close()
# Reconnect to the MQTT server.
while True:
net_sta = net.getState() # Get network registration information.
if net_sta != -1 and net_sta[1][0] == 1:
call_state = dataCall.getInfo(1, 0) # Get data call information.
if (call_state != -1) and (call_state[2][0] == 1):
try:
# The network is normal. Reconnect to the MQTT server.
self.connect()
except Exception as e:
# Reconnection to the MQTT server failed. Try again 5 s later.
self.client.close()
utime.sleep(5)
continue
else:
# The network is unrestored. Please wait.
utime.sleep(10)
continue
# Connect to the MQTT server successfully and subscribe to the topic.
try:
# Multiple topics can be saved by a list. Traverse the list to resubscribe the topic.
if self.topic is not None:
self.client.subscribe(self.topic, self.qos)
self.mp_lock.release()
except:
# Subscription failed. Reconnect to the MQTT server.
self.client.close()
utime.sleep(5)
continue
else:
utime.sleep(5)
continue
break # Stop loop.
# Exit and reconnect.
return True
def nw_cb(self, args):
'''
Call the callback function of data call.
'''
nw_sta = args[1]
if nw_sta == 1:
# Network connected.
mqtt_log.info("*** network connected! ***")
self.__nw_flag = True
else:
# Network disconnected.
mqtt_log.info("*** network not connected! ***")
self.__nw_flag = False
def __listen(self):
while True:
try:
if not TaskEnable:
break
self.client.wait_msg()
except OSError as e:
# Determine whether the network is disconnected.
if not self.__nw_flag:
# Reconnect after the network is restored from disconnection.
self.reconnect()
# Reconnect when the socket status is abnormal.
elif self.client.get_mqttsta() != 0 and TaskEnable:
self.reconnect()
else:
# You can call the raise method to return an exception or -1.
return -1
def loop_forever(self):
_thread.start_new_thread(self.__listen, ())
if __name__ == '__main__':
'''
When running this routine manually, you can remove this delay. If you change the file name of the routine to main.py, you need to add this delay when you want to start the routine automatically. Otherwise, you cannot see the information printed in poweron_print_once() below from the CDC interface.
'''
utime.sleep(5)
checknet.poweron_print_once()
'''
If the user program contains network-related codes, it must execute wait_network_connected() to wait for the network to be ready (successful data call); If it is a network-independent code, you can mask wait_network_connected().
【This routine must retain the following line.】
'''
checknet.wait_network_connected()
def sub_cb(topic, msg):
# global state
mqtt_log.info("Subscribe Recv: Topic={},Msg={}".format(topic.decode(), msg.decode()))
c = MqttClient("clientId-Y1ymSzvQcC", "broker.hivemq.com", 1883, reconn=False)
def err_cb(error):
'''
Set the callback function of receiving MQTT thread error occurrence.
'''
mqtt_log.info(error)
c.reconnect() # Reconnect to MQTT server after error occurrences.
# c = MqttClient("umqtt_client_753", "mq.tongxinmao.com", 18830, reconn=False)
# Set the callback function of receiving messages.
c.set_callback(sub_cb)
# Set the callback function of error occurrence.
c.error_register_cb(err_cb)
# Connect to the MQTT server.
c.connect()
# Subscribe to topics.
c.subscribe(b"/public/TEST/quecpython758")
mqtt_log.info("Connected to mq.tongxinmao.com, subscribed to /public/TEST/quecpython topic")
# Publish a message.
c.publish(b"/public/TEST/quecpython758", b"my name is Quecpython!")
mqtt_log.info("Publish topic: /public/TEST/quecpython758, msg: my name is Quecpython")
# Monitor MQTT messages.
c.loop_forever()
# Wait for 5 s to receive the message.
# Note: Comment c.disconnect () and utime.sleep(5) if you want to test the reconnection mechanism, including server disconnection.
# utime.sleep(5)
# Disconnect to the MQTT server.
# c.disconnect()
Execução
Ref:
O que é MicroPython? – COMPRACO - Compra coordenada por IA
QPYcom User Guide - QuecPython
HiveMQ – The Most Trusted MQTT platform to Transform Your Business
Index of /
珍藏，广和通L610全套标准指令AT手册下载 / NB-IoT/2G模块/4G模块/GPRS模块/GPS/北斗 / WhyCan Forum(哇酷开发者社区)
Sobre a SMARTCORE
A SmartCore fornece módulos para comunicação wireless, biometria, conectividade, rastreamento e automação.
Nosso portfólio inclui modem 2G/3G/4G/NB-IoT/Cat.M, satelital, módulos WiFi, Bluetooth, GNSS / GPS, Sigfox, LoRa, leitor de cartão, leitor QR code, mecanismo de impressão, mini-board PC, antena, pigtail, LCD, bateria, repetidor GPS e sensores.
