Compare commits
3 commits
2e1742279d
...
279a45a9e8
| Author | SHA1 | Date | |
|---|---|---|---|
|
279a45a9e8 |
|||
|
64ab73e9c3 |
|||
|
b752888812 |
3 changed files with 26 additions and 19 deletions
|
|
@ -81,7 +81,6 @@ def main():
|
||||||
default=[i for i in os.environ.get("DOORIS_STATIC_API_TOKENS", "").split(",") if bool(i)],
|
default=[i for i in os.environ.get("DOORIS_STATIC_API_TOKENS", "").split(",") if bool(i)],
|
||||||
)
|
)
|
||||||
args = argp.parse_args()
|
args = argp.parse_args()
|
||||||
print(args.static_api_tokens)
|
|
||||||
|
|
||||||
# setup logging
|
# setup logging
|
||||||
logging.basicConfig(
|
logging.basicConfig(
|
||||||
|
|
|
||||||
|
|
@ -90,15 +90,12 @@ class CCUJackClient:
|
||||||
self.data_updated = asyncio.Event()
|
self.data_updated = asyncio.Event()
|
||||||
|
|
||||||
async def start(self):
|
async def start(self):
|
||||||
await self.mqtt.connect()
|
|
||||||
await self.find_locks()
|
|
||||||
|
|
||||||
self.task_cron= asyncio.get_running_loop().create_task(
|
|
||||||
self.cron(), name="ccujack-cron"
|
|
||||||
)
|
|
||||||
self.task_process_messages = asyncio.get_running_loop().create_task(
|
self.task_process_messages = asyncio.get_running_loop().create_task(
|
||||||
self.process_mqt_messages(), name="process-mqtt-messages"
|
self.process_mqt_messages(), name="process-mqtt-messages"
|
||||||
)
|
)
|
||||||
|
self.task_cron= asyncio.get_running_loop().create_task(
|
||||||
|
self.cron(), name="ccujack-cron"
|
||||||
|
)
|
||||||
|
|
||||||
async def close_connections(self):
|
async def close_connections(self):
|
||||||
await asyncio.gather(self.mqtt.disconnect(), self.http.close())
|
await asyncio.gather(self.mqtt.disconnect(), self.http.close())
|
||||||
|
|
@ -162,15 +159,18 @@ class CCUJackClient:
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
|
|
||||||
await asyncio.sleep(15 * 60) # 15 minutes
|
|
||||||
logger.info("Running CCUJack cron")
|
logger.info("Running CCUJack cron")
|
||||||
await self.find_locks()
|
if not self.mqtt.is_connected:
|
||||||
if not self.mqtt.is_connected():
|
|
||||||
logger.warning("MQTT client was discovered to be disconnected; reconnecting now")
|
logger.warning("MQTT client was discovered to be disconnected; reconnecting now")
|
||||||
await self.mqtt.connect()
|
await self.mqtt.connect()
|
||||||
|
|
||||||
|
await self.find_locks()
|
||||||
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.exception(f"Error in CCUJack cron task: {e}")
|
logger.exception(f"Error in CCUJack cron task: {e}")
|
||||||
|
finally:
|
||||||
|
await asyncio.sleep(15 * 60) # 15 minutes
|
||||||
|
|
||||||
|
|
||||||
async def query_param_value(self, address: str) -> CCUValue:
|
async def query_param_value(self, address: str) -> CCUValue:
|
||||||
if address in self.param_values:
|
if address in self.param_values:
|
||||||
|
|
|
||||||
|
|
@ -128,19 +128,21 @@ class AsyncMqttClient:
|
||||||
|
|
||||||
to_add = topics.difference(self.active_subscriptions)
|
to_add = topics.difference(self.active_subscriptions)
|
||||||
if to_add:
|
if to_add:
|
||||||
|
self.active_subscriptions.update(to_add)
|
||||||
|
if self.is_connected:
|
||||||
logger.info(f"mqtt client subscribing to topics {', '.join(to_add)}")
|
logger.info(f"mqtt client subscribing to topics {', '.join(to_add)}")
|
||||||
self.fut_subscribe = asyncio.get_running_loop().create_future()
|
self.fut_subscribe = asyncio.get_running_loop().create_future()
|
||||||
self.client.subscribe([(i, qos) for i in to_add])
|
self.client.subscribe([(i, qos) for i in to_add])
|
||||||
await self.fut_subscribe
|
await self.fut_subscribe
|
||||||
self.active_subscriptions.update(to_add)
|
|
||||||
|
|
||||||
to_remove = self.active_subscriptions.difference(topics)
|
to_remove = self.active_subscriptions.difference(topics)
|
||||||
if to_remove:
|
if to_remove:
|
||||||
|
self.active_subscriptions.difference_update(to_remove)
|
||||||
|
if self.is_connected:
|
||||||
logger.info(f"mqtt client unsubscribing from topics {','.join(to_remove)}")
|
logger.info(f"mqtt client unsubscribing from topics {','.join(to_remove)}")
|
||||||
self.fut_unsubscribe = asyncio.get_running_loop().create_future()
|
self.fut_unsubscribe = asyncio.get_running_loop().create_future()
|
||||||
self.client.unsubscribe(list(to_remove))
|
self.client.unsubscribe(list(to_remove))
|
||||||
await self.fut_unsubscribe
|
await self.fut_unsubscribe
|
||||||
self.active_subscriptions.difference_update(to_remove)
|
|
||||||
|
|
||||||
async def connect(self):
|
async def connect(self):
|
||||||
server_host, server_port = self.connection_string.rsplit(":", maxsplit=1)
|
server_host, server_port = self.connection_string.rsplit(":", maxsplit=1)
|
||||||
|
|
@ -157,13 +159,19 @@ class AsyncMqttClient:
|
||||||
# re-establish all supposed mqtt subscriptions
|
# re-establish all supposed mqtt subscriptions
|
||||||
if len(self.active_subscriptions) > 0:
|
if len(self.active_subscriptions) > 0:
|
||||||
qos = 1
|
qos = 1
|
||||||
await self.client.subscribe((i, qos) for i in self.active_subscriptions)
|
self.fut_subscribe = asyncio.get_running_loop().create_future()
|
||||||
|
self.client.subscribe([(i, qos) for i in self.active_subscriptions])
|
||||||
|
await self.fut_subscribe
|
||||||
|
|
||||||
async def disconnect(self):
|
async def disconnect(self):
|
||||||
|
if not self.is_connected:
|
||||||
|
return
|
||||||
|
|
||||||
logger.info("Disconnecting mqtt client from broker")
|
logger.info("Disconnecting mqtt client from broker")
|
||||||
self.fut_disconnect = asyncio.get_running_loop().create_future()
|
self.fut_disconnect = asyncio.get_running_loop().create_future()
|
||||||
self.client.disconnect()
|
self.client.disconnect()
|
||||||
await self.fut_disconnect
|
await self.fut_disconnect
|
||||||
|
|
||||||
|
@property
|
||||||
def is_connected(self) -> bool:
|
def is_connected(self) -> bool:
|
||||||
return self.client.is_connected()
|
return self.client.is_connected()
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue