ghpascon-rust 0.5.1

A personal Rust utility library
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

from datetime import datetime
from smartx_rfid.devices import R700_IOT
import asyncio
import logging

# Configure logging to show INFO level messages
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    datefmt="%H:%M:%S",
)


def on_tag_event(name: str, tag_data: dict):
    """Callback for when a tag is read"""
    print(f"🏷️  Tag Read: {tag_data}")
    print()


def on_r700_iot_event(name: str, event_type: str, event_data=None):
    """General event handler for R700_IOT events"""
    print("=" * 60)
    if event_type == "tag":
        on_tag_event(name, event_data)
        return
    print(f"{name} -> Event: {event_type}, Data: {event_data}")
    print()


async def main():
    # === SERIAL EXAMPLE ===
    print("=== R700 IOT Example ===")
    # config.pop("startTriggers")
    # config.pop("stopTriggers")
    r700_iot = R700_IOT(name="R700_IOT", ip="impinj-14-46-36", active_ant=[1], read_power=12)
    print(r700_iot.reading_config)
    r700_iot.on_event = on_r700_iot_event
    print("Starting R700 IOT connection...")
    asyncio.create_task(r700_iot.connect())

    # Keep the main loop running
    while True:
        await asyncio.sleep(1)
        if not r700_iot.is_connected:
            continue

        if r700_iot.is_reading:
            await r700_iot.stop_inventory()
        else:
            start_time = datetime.now()
            await r700_iot.start_inventory()
            end_time = datetime.now()
            elapsed_time = end_time - start_time
            logging.info(f" ===== Elapsed time: {elapsed_time} ===== ")


if __name__ == "__main__":
    asyncio.run(main())