Skip to main content

​
Overview

The multi-device server provides comprehensive device lifecycle management through REST APIs and persistent storage. Devices are registered dynamically, stored in devices.json, and managed in-memory through the DEVICES dictionary.

​
Device Data Structure

Each device is represented by a configuration object with connection and metadata information:
Device Configuration Schema
{
    "device_id": {  # Unique identifier (key in DEVICES dict)
        "ip": "192.168.1.205",              # Device IP address
        "port": 4370,                        # Communication port
        "password": 0,                       # Device access password
        "timeout": 5,                        # Connection timeout (seconds)
        "name": "Entrada Principal",         # Human-readable name
        "created_at": "2026-03-06 14:30:00", # Registration timestamp
        "updated_at": "2026-03-06 15:45:00"  # Last update (optional)
    }
}

​
Example Device Registry

devices.json
{
  "principal": {
    "ip": "192.168.1.205",
    "port": 4370,
    "password": 0,
    "timeout": 5,
    "name": "Entrada Principal",
    "created_at": "2026-03-06 14:30:00"
  },
  "bodega": {
    "ip": "192.168.1.207",
    "port": 4370,
    "password": 0,
    "timeout": 5,
    "name": "Bodega Principal",
    "created_at": "2026-03-06 15:20:00",
    "updated_at": "2026-03-06 15:45:00"
  }
}

​
The DEVICES Dictionary

The DEVICES dictionary is the in-memory registry of all configured devices:
source/server.py
# Global device registry
DEVICES = load_devices()

# Example structure:
# DEVICES = {
#     "principal": {"ip": "192.168.1.205", "port": 4370, ...},
#     "bodega": {"ip": "192.168.1.207", "port": 4370, ...}
# }
Why Dictionary Keys?Device IDs are used as dictionary keys for O(1) lookup performance. This is critical when validating device existence in request handlers.

​
Device ID Format

Device IDs are automatically normalized during registration:
source/server.py
device_id = str(body.get("id", "")).strip().lower().replace(" ", "_")
Examples:
  • "Principal" → "principal"
  • "Main Entrance" → "main_entrance"
  • "Bodega 2" → "bodega_2"

​
Persistent Storage

​
Loading Devices

Devices are loaded from devices.json on server startup:
source/server.py
DEVICES_FILE = os.getenv("DEVICES_FILE", "devices.json")

def load_devices():
    """Load devices from devices.json. If not exists, create defaults."""
    if os.path.exists(DEVICES_FILE):
        with open(DEVICES_FILE, "r", encoding="utf-8") as f:
            return json.load(f)
    
    # Default device for first-time setup
    default = {
        "principal": {
            "ip": "192.168.1.205",
            "port": 4370,
            "password": 0,
            "timeout": 5,
            "name": "Entrada Principal",
            "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        }
    }
    save_devices(default)
    return default

# Load on startup
DEVICES = load_devices()
log.info(f"Dispositivos cargados: {list(DEVICES.keys())}")
First-Time SetupIf devices.json doesn’t exist, the server creates it with a default device configuration. This ensures the server is immediately usable after installation.

​
Saving Devices

Every device modification triggers an immediate save to disk:
source/server.py
def save_devices(data):
    """Persist device registry to disk."""
    with open(DEVICES_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
Thread SafetyAll save_devices() calls occur within a devices_lock context to prevent concurrent writes that could corrupt the file.

​
Device Lifecycle

​
Registration Flow

1

Receive Registration Request

Client sends POST request to /devices with device configuration:
{
  "id": "bodega",
  "name": "Bodega Principal",
  "ip": "192.168.1.207",
  "port": 4370,
  "password": 0,
  "timeout": 5
}
2

Validate and Normalize

Server validates required fields and normalizes the device ID:
source/server.py
device_id = str(body.get("id", "")).strip().lower().replace(" ", "_")
name = str(body.get("name", "")).strip()
ip = str(body.get("ip", "")).strip()

if not device_id or not name or not ip:
    return _json({"success": False, "error": "id, name e ip son requeridos."}, 400)
3

Check for Conflicts

Verify the device ID doesn’t already exist:
source/server.py
with devices_lock:
    if device_id in DEVICES:
        return _json({"success": False, "error": f"El id '{device_id}' ya existe."}, 409)
4

Create Device and Lock

Add device to registry and create its dedicated lock:
source/server.py
DEVICES[device_id] = {
    "ip": ip,
    "port": int(body.get("port", 4370)),
    "password": int(body.get("password", 0)),
    "timeout": int(body.get("timeout", 5)),
    "name": name,
    "created_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
}
_locks[device_id] = threading.Lock()
save_devices(DEVICES)
5

Log and Respond

Log the registration and return success response:
source/server.py
log.info(f"Dispositivo registrado: [{device_id}] {name} → {ip}")
return _json({
    "success": True,
    "message": f"Dispositivo '{device_id}' registrado.",
    "data": DEVICES[device_id]
})

​
Update Flow

Devices can be updated via PUT request to /devices/{device_id}:
source/server.py
@app.route("/devices/<device_id>", methods=["PUT"])
def update_device(device_id):
    if device_id not in DEVICES:
        return device_not_found(device_id)
    
    body = request.get_json(silent=True)
    if not body:
        return _json({"success": False, "error": "Body JSON requerido."}, 400)
    
    with devices_lock:
        cfg = DEVICES[device_id]
        
        # Update only provided fields
        if "name" in body: cfg["name"] = str(body["name"]).strip()
        if "ip" in body: cfg["ip"] = str(body["ip"]).strip()
        if "port" in body: cfg["port"] = int(body["port"])
        if "password" in body: cfg["password"] = int(body["password"])
        if "timeout" in body: cfg["timeout"] = int(body["timeout"])
        
        # Add update timestamp
        cfg["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        save_devices(DEVICES)
    
    return _json({"success": True, "message": f"Dispositivo '{device_id}' actualizado."})
Partial Updates SupportedYou can update just the fields you need. Omitted fields retain their current values.

​
Deletion Flow

Removing a device cleans up both the registry and its lock:
source/server.py
@app.route("/devices/<device_id>", methods=["DELETE"])
def remove_device(device_id):
    if device_id not in DEVICES:
        return device_not_found(device_id)
    
    with devices_lock:
        del DEVICES[device_id]
        _locks.pop(device_id, None)  # Remove device lock
        save_devices(DEVICES)
    
    log.info(f"Dispositivo eliminado: [{device_id}]")
    return _json({"success": True, "message": f"Dispositivo '{device_id}' eliminado."})

​
Per-Device Lock Management

Each device gets its own threading lock to allow parallel operations:
source/server.py
# Initialize locks for all loaded devices
_locks = {k: threading.Lock() for k in DEVICES}

def get_lock(device_id):
    """Returns (creating if needed) the lock for a device."""
    with devices_lock:
        if device_id not in _locks:
            _locks[device_id] = threading.Lock()
        return _locks[device_id]

​
Lock Lifecycle

  • Server startup: Locks created for all devices in devices.json
  • Device registration: New lock created when device is added
  • First access: get_lock() creates lock if missing (race condition safety)
  • Device deletion: Lock removed from _locks dictionary
  • Server shutdown: All locks released automatically

​
Device Validation

Every device-specific endpoint validates device existence:
source/server.py
def device_not_found(device_id):
    return _json({
        "success": False,
        "error": f"Dispositivo '{device_id}' no encontrado.",
        "disponibles": list(DEVICES.keys()),
    }), 404

@app.route("/devices/<device_id>/users", methods=["GET"])
def get_users(device_id):
    if device_id not in DEVICES:
        return device_not_found(device_id)
    
    # Proceed with operation...
Error Response Example: 
{
  "success": false,
  "error": "Dispositivo 'oficina' no encontrado.",
  "disponibles": ["principal", "bodega"]
}
Including available device IDs in the error helps developers quickly identify the correct device ID to use.

​
Device Information Retrieval

​
List All Devices

Get a summary of all registered devices:
source/server.py
@app.route("/devices", methods=["GET"])
def list_devices():
    with devices_lock:
        data = [
            {
                "id": k,
                "name": v["name"],
                "ip": v["ip"],
                "port": v["port"],
                "created_at": v.get("created_at", ""),
            }
            for k, v in DEVICES.items()
        ]
    return _json({"success": True, "total": len(data), "data": data})
Response Example: 
{
  "success": true,
  "total": 2,
  "data": [
    {
      "id": "principal",
      "name": "Entrada Principal",
      "ip": "192.168.1.205",
      "port": 4370,
      "created_at": "2026-03-06 14:30:00"
    },
    {
      "id": "bodega",
      "name": "Bodega Principal",
      "ip": "192.168.1.207",
      "port": 4370,
      "created_at": "2026-03-06 15:20:00"
    }
  ]
}

​
Get Single Device

Retrieve detailed configuration for a specific device:
source/server.py
@app.route("/devices/<device_id>", methods=["GET"])
def get_device(device_id):
    if device_id not in DEVICES:
        return device_not_found(device_id)
    return _json({"success": True, "data": {**DEVICES[device_id], "id": device_id}})

​
Ping Device

Test connectivity and retrieve device time:
source/server.py
@app.route("/devices/<device_id>/ping", methods=["GET"])
def ping_device(device_id):
    if device_id not in DEVICES:
        return device_not_found(device_id)
    
    conn = None
    try:
        log.info(f"[{device_id}] Probando conexion...")
        conn = conectar(device_id)
        t = conn.get_time()
        return _json({
            "success": True,
            "message": f"Conexion exitosa con '{device_id}'.",
            "device_time": str(t),
            "ip": DEVICES[device_id]["ip"],
        })
    except Exception as e:
        log.error(f"[{device_id}] Ping fallido: {e}")
        return _json({
            "success": False, 
            "error": str(e), 
            "ip": DEVICES[device_id]["ip"]
        }, 503)
    finally:
        if conn:
            conn.disconnect()

​
Configuration via Environment

The device storage location can be customized: 
export DEVICES_FILE="/var/lib/zkteco/devices.json"
export API_HOST="0.0.0.0"
export API_PORT="5000"

​
Startup Sequence

Here’s what happens when the multi-device server starts:
1

Load Configuration

Environment variables are read for server and storage paths
2

Initialize Logging

Logging system configured with timestamp format
3

Load Devices

load_devices() reads devices.json or creates default configuration
4

Create Locks

Per-device locks initialized: _locks = {k: threading.Lock() for k in DEVICES}
5

Generate SSL Certificates

If certificates don’t exist, auto-generate self-signed certificates
6

Start Flask Server

Server starts listening on configured host:port with or without SSL
Startup Log Example: 
2026-03-06 14:30:15 [INFO] Dispositivos cargados: ['principal', 'bodega']
2026-03-06 14:30:15 [INFO] Certificados SSL generados.
2026-03-06 14:30:15 [INFO] ==========================================================
2026-03-06 14:30:15 [INFO]   Servidor ZKTeco — Multi-dispositivo dinamico
2026-03-06 14:30:15 [INFO]   https://0.0.0.0:5000
2026-03-06 14:30:15 [INFO]   Dispositivos activos: 2
2026-03-06 14:30:15 [INFO]     [principal]  Entrada Principal  →  192.168.1.205:4370
2026-03-06 14:30:15 [INFO]     [bodega]  Bodega Principal  →  192.168.1.207:4370
2026-03-06 14:30:15 [INFO] ==========================================================
Next Steps:

Build docs developers (and LLMs) love

Get started for freeTalk to us