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
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
smartx_rfid-12.0.1.tar.gz
(98.6 kB
view details)
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
smartx_rfid-12.0.1-py3-none-any.whl
(126.1 kB
view details)
File details
Details for the file smartx_rfid-12.0.1.tar.gz.
File metadata
- Download URL: smartx_rfid-12.0.1.tar.gz
- Upload date:
- Size: 98.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.11.16
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
5dc0638ee175199151d10db015fa0c71d16322c7a00a6e414058ff90892529df
|
|
| MD5 |
54737c79d0db762f2544152fdb39e9a3
|
|
| BLAKE2b-256 |
a8e3e38fc37cf033b9ad18ab79a8bfa0aaa5b60fd4b0aa242b368bb2b2afe20b
|
File details
Details for the file smartx_rfid-12.0.1-py3-none-any.whl.
File metadata
- Download URL: smartx_rfid-12.0.1-py3-none-any.whl
- Upload date:
- Size: 126.1 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.11.16
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
7d16f4265bc6de1ab75c93c6f15db3c7e1b1ba1187e2c924106f5fe0fbdf4b3d
|
|
| MD5 |
a8c699f9c9af17e8f1bb76133abf2a89
|
|
| BLAKE2b-256 |
b6700a67ef6b20f69c77ed34dc49fe38d014d78cbe07c13f894ae43370c57971
|
