SmartX RFID
Python library for SmartX RFID integrations: device communication, webhook delivery, database utilities, parsing, and support modules for production flows.
Installation
pip install smartx-rfid
For development:
poetry install
Main Modules
- smartx_rfid.devices
- RFID: X714, R700_IOT, ACUPAD
- Generic: SERIAL, TCP
- Printers: SatoPrinter, SatoWs4Printer
- DeviceManager
- smartx_rfid.webhook
- WebhookManager
- WebhookXtrack
- smartx_rfid.db
- DatabaseManager
- smartx_rfid.auth
- token and password helpers
- smartx_rfid.api
- Omie and Xtrack API clients
- smartx_rfid.utils
- TagList, AlertsManager, delayed_function, hash, regex helpers
Quick Start (Device Events)
import asyncio
from smartx_rfid.devices import X714
async def handle_tag(device_name: str, event_data: dict):
print(f"[{device_name}] EPC={event_data.get('epc')} RSSI={event_data.get('rssi')}")
async def main():
reader = X714(name="X714-Reader", start_reading=True)
def on_event(device_name: str, event_type: str, event_data: dict):
if event_type == "tag":
asyncio.create_task(handle_tag(device_name, event_data))
reader.on_event = on_event
await reader.connect()
while True:
await asyncio.sleep(1)
asyncio.run(main())
Webhook Manager
The webhook API now has two explicit sending methods:
- post_event: standardized event envelope
- post: custom payload POST
It is designed for burst traffic and includes:
- async client reuse (keep-alive)
- request concurrency limiter
- retry with exponential backoff for network errors, HTTP 429, and HTTP 5xx
- safe payload serialization for datetime/date, Decimal, bytes, custom objects, dict/list/tuple/set
Constructor
from smartx_rfid.webhook import WebhookManager
webhook = WebhookManager(
url="https://api.example.com/webhook", # base URL
timeout=5.0,
max_retries=2,
max_concurrent_requests=50,
)
post_event (standard payload)
success = await webhook.post_event(
device="reader-01",
event_type="tag_read",
event_data={"epc": "E200001175000001", "rssi": -45.2},
)
Payload format sent by post_event:
{
"device": "reader-01",
"event_type": "tag_read",
"event_data": {
"epc": "E200001175000001",
"rssi": -45.2
}
}
post (custom payload + optional URL override)
from datetime import datetime
from decimal import Decimal
payload = {
"type": "heartbeat",
"status": "ok",
"timestamp": datetime.utcnow(),
"load": Decimal("0.73"),
}
success = await webhook.post(
payload=payload,
headers={"X-Source": "rfid-core"},
url="https://api.example.com/custom-endpoint", # optional
)
If url is not provided, WebhookManager uses the base URL configured in the constructor.
High-Throughput Example
import asyncio
webhook = WebhookManager(
url="https://api.example.com/events",
max_concurrent_requests=20,
max_retries=2,
)
events = [{"index": i, "ok": True} for i in range(200)]
results = await asyncio.gather(*(webhook.post(payload=e) for e in events))
print(f"sent={sum(results)} failed={len(results) - sum(results)}")
await webhook.aclose()
Recommended pattern:
async with WebhookManager(url="https://api.example.com/events") as webhook:
await webhook.post_event("reader-01", "connected", {"status": "ok"})
Migration Note
Previous usage:
await webhook.post("device", "event_type", {"data": 1})
Current usage:
await webhook.post_event("device", "event_type", {"data": 1})
For arbitrary payloads, use:
await webhook.post(payload={"anything": "you need"})
Database Example
from datetime import datetime
from sqlalchemy import Column, DateTime, Float, Integer, String
from sqlalchemy.orm import DeclarativeBase
from smartx_rfid.db import DatabaseManager
class Base(DeclarativeBase):
pass
class TagModel(Base):
__tablename__ = "rfid_tags"
id = Column(Integer, primary_key=True)
epc = Column(String(64), unique=True, nullable=False)
rssi = Column(Float)
created_at = Column(DateTime, default=datetime.utcnow)
db = DatabaseManager("sqlite:///rfid_tags.db")
db.register_models(TagModel)
db.create_tables()
with db.get_session() as session:
session.add(TagModel(epc="E200001175000001", rssi=-45.2))
Event Callback Contract
All devices share a consistent callback contract:
def on_event(device_name: str, event_type: str, event_data: dict):
# Common event_type values:
# connected, disconnected, tag, error
...
Examples Directory
The examples folder includes:
- examples/devices/RFID
- examples/devices/generic
- examples/devices/printer
- examples/db
- examples/api
- examples/email
- examples/license
- examples/smtx_db
- examples/utils
Run examples with:
python examples/devices/RFID/X714_SERIAL.py
python examples/db/showcase.py
Running Tests
pytest
To run only webhook-related tests:
pytest tests/utils/test_serialization.py -k webhook
Requirements
- Python 3.11+
- Works on Linux, macOS, and Windows (depending on device transport and drivers)
License
MIT
Support
Release files for smartx-rfid 12.4.2
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| smartx_rfid-12.4.2.tar.gz | 108.4 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| smartx_rfid-12.4.2-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 245.4 kB
Release files / smartx_rfid-12.4.2.tar.gz
| Download URL | smartx_rfid-12.4.2.tar.gz |
|---|---|
| Size | 108.4 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
2a4ca222b165e207cf8a1b06306f535aec2525f9375ad89921edb8f560b5c84b
|
|
BLAKE2b-256 checksum How to use checksums |
8ef59fe96ae7a1b2613e7c0ad4f7f48432bd080f0cbf0457cc28316bfef48a8d
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.11.16
|
Release files / smartx_rfid-12.4.2-py3-none-any.whl
| Download URL | smartx_rfid-12.4.2-py3-none-any.whl |
|---|---|
| Size | 137.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
e56fbcc5a1a1ddaef6f7f7c9bf588275a085a04f34cece851fe0428b910ee6c3
|
|
BLAKE2b-256 checksum How to use checksums |
5616e6f0c912db8dc00f5d74cf184b480b699b7dfe3e950b2fa3af0564b791db
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.11.16
|