How to Integrate a Barcode Scanner with a Local WMS and Access It from the Cloud

Architecture Overview

The stack has three layers. The edge layer runs on a workstation or server inside the warehouse a Python reader that captures barcode scanner input and a Flask WMS API that stores scan events and exposes an HTTP interface. The tunnel layer is a Localtonet HTTP tunnel that maps the local WMS API to a public HTTPS URL. The cloud layer is any external system an ERP, a SaaS logistics platform, a supplier portal, or a management dashboard that calls the WMS API over HTTPS or receives webhook notifications when scan events occur.

Read Barcode Scanner Input with Python

Most USB barcode scanners operate in HID keyboard emulation mode by default. When you scan a barcode, the scanner sends the barcode string to the operating system as if it were typed on a keyboard, followed by an Enter keystroke. On Linux, the most reliable way to capture this without it also typing into whatever application has keyboard focus is to read the scanner's raw input device using the evdev library. This gives you exclusive scanner input without any keyboard interference.

pip install evdev requests --break-system-packages

Find your scanner's input device

# List all input devices
python3 -c "
import evdev
devices = [evdev.InputDevice(path) for path in evdev.list_devices()]
for d in devices:
    print(d.path, d.name, d.phys)
"

Look for your scanner in the output. It typically appears as a device with a name like USB Barcode Scanner or the scanner's brand name. Note the path usually /dev/input/event2 or similar.

scanner_reader.py

#!/usr/bin/env python3
# scanner_reader.py
# Reads barcode scanner input via evdev and posts each scan to the local WMS API

import evdev
import requests
import logging
import time
import sys

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s'
)
logger = logging.getLogger(__name__)

# Path to the scanner's input device
# Run the device finder above to determine the correct path
SCANNER_DEVICE = '/dev/input/event2'

# Local WMS API endpoint
WMS_API_URL = 'http://127.0.0.1:5000/api/scans'

# Station identifier for this workstation
STATION_ID = 'station_01'

# Map evdev key codes to characters
KEY_MAP = {
    'KEY_1': '1', 'KEY_2': '2', 'KEY_3': '3', 'KEY_4': '4', 'KEY_5': '5',
    'KEY_6': '6', 'KEY_7': '7', 'KEY_8': '8', 'KEY_9': '9', 'KEY_0': '0',
    'KEY_A': 'A', 'KEY_B': 'B', 'KEY_C': 'C', 'KEY_D': 'D', 'KEY_E': 'E',
    'KEY_F': 'F', 'KEY_G': 'G', 'KEY_H': 'H', 'KEY_I': 'I', 'KEY_J': 'J',
    'KEY_K': 'K', 'KEY_L': 'L', 'KEY_M': 'M', 'KEY_N': 'N', 'KEY_O': 'O',
    'KEY_P': 'P', 'KEY_Q': 'Q', 'KEY_R': 'R', 'KEY_S': 'S', 'KEY_T': 'T',
    'KEY_U': 'U', 'KEY_V': 'V', 'KEY_W': 'W', 'KEY_X': 'X', 'KEY_Y': 'Y',
    'KEY_Z': 'Z', 'KEY_MINUS': '-', 'KEY_DOT': '.', 'KEY_SLASH': '/',
}

def post_scan(barcode):
    """Send scan event to local WMS API."""
    try:
        response = requests.post(WMS_API_URL, json={
            'barcode':    barcode,
            'station_id': STATION_ID,
            'timestamp':  time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
        }, timeout=2)
        if response.status_code == 200:
            data = response.json()
            logger.info(f'Scan accepted: {barcode} -> {data.get("item_name", "unknown")}')
        else:
            logger.warning(f'WMS API returned {response.status_code} for barcode {barcode}')
    except requests.RequestException as e:
        logger.error(f'Failed to post scan {barcode}: {e}')

def main():
    try:
        scanner = evdev.InputDevice(SCANNER_DEVICE)
        # Grab exclusive access so the scanner does not type into other applications
        scanner.grab()
        logger.info(f'Listening on scanner: {scanner.name} ({SCANNER_DEVICE})')
    except (FileNotFoundError, PermissionError) as e:
        logger.error(f'Cannot open scanner device {SCANNER_DEVICE}: {e}')
        logger.error('Run as root or add user to the input group: sudo usermod -a -G input $USER')
        sys.exit(1)

    barcode_buffer = []

    for event in scanner.read_loop():
        if event.type == evdev.ecodes.EV_KEY:
            key_event = evdev.categorize(event)
            if key_event.keystate == evdev.KeyEvent.key_down:
                key_name = key_event.keycode

                if key_name == 'KEY_ENTER':
                    if barcode_buffer:
                        barcode = ''.join(barcode_buffer)
                        logger.info(f'Scanned: {barcode}')
                        post_scan(barcode)
                        barcode_buffer = []
                elif isinstance(key_name, str) and key_name in KEY_MAP:
                    barcode_buffer.append(KEY_MAP[key_name])

if __name__ == '__main__':
    main()

Build the Local WMS API

The local WMS API receives scan events from the reader, stores them in SQLite, updates inventory state, and exposes REST endpoints for cloud systems to query. SQLite is sufficient for a single-facility WMS handling hundreds of scans per hour. For higher volumes or multiple concurrent workstations, replace SQLite with PostgreSQL.

pip install flask --break-system-packages

wms_api.py

#!/usr/bin/env python3
# wms_api.py
# Local Warehouse Management System API

import sqlite3
import time
import json
import requests
import logging
from flask import Flask, request, jsonify, g

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)

app = Flask(__name__)
DB_PATH = 'wms.db'

# Optional: webhook URL for cloud ERP notifications
CLOUD_WEBHOOK_URL = ''   # Set this to your cloud ERP webhook endpoint

# --- Database ---

def get_db():
    if 'db' not in g:
        g.db = sqlite3.connect(DB_PATH)
        g.db.row_factory = sqlite3.Row
    return g.db

@app.teardown_appcontext
def close_db(error):
    db = g.pop('db', None)
    if db:
        db.close()

def init_db():
    with sqlite3.connect(DB_PATH) as conn:
        conn.executescript('''
            CREATE TABLE IF NOT EXISTS items (
                barcode    TEXT PRIMARY KEY,
                name       TEXT NOT NULL,
                sku        TEXT,
                quantity   INTEGER DEFAULT 0,
                location   TEXT
            );

            CREATE TABLE IF NOT EXISTS scan_events (
                id         INTEGER PRIMARY KEY AUTOINCREMENT,
                barcode    TEXT NOT NULL,
                station_id TEXT,
                action     TEXT DEFAULT 'scan',
                quantity   INTEGER DEFAULT 1,
                scanned_at TEXT NOT NULL
            );

            INSERT OR IGNORE INTO items (barcode, name, sku, quantity, location)
            VALUES
                ('1234567890128', 'Widget A',      'SKU-001', 100, 'A-01-01'),
                ('9780201379624', 'Component B',   'SKU-002',  50, 'B-03-02'),
                ('5901234123457', 'Assembly Part', 'SKU-003',  75, 'C-02-04');
        ''')

# --- API endpoints ---

@app.route('/api/scans', methods=['POST'])
def receive_scan():
    """Receive a scan event from the scanner reader."""
    data = request.get_json()
    if not data or 'barcode' not in data:
        return jsonify(error='barcode field required'), 400

    barcode    = data['barcode'].strip()
    station_id = data.get('station_id', 'unknown')
    action     = data.get('action', 'scan')
    quantity   = int(data.get('quantity', 1))
    scanned_at = data.get('timestamp', time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()))

    db   = get_db()
    item = db.execute('SELECT * FROM items WHERE barcode = ?', (barcode,)).fetchone()

    if not item:
        db.execute(
            'INSERT INTO scan_events (barcode, station_id, action, quantity, scanned_at) VALUES (?,?,?,?,?)',
            (barcode, station_id, 'unknown_barcode', quantity, scanned_at)
        )
        db.commit()
        logger.warning(f'Unknown barcode scanned: {barcode} at {station_id}')
        return jsonify(status='unknown_barcode', barcode=barcode), 200

    # Record the scan event
    db.execute(
        'INSERT INTO scan_events (barcode, station_id, action, quantity, scanned_at) VALUES (?,?,?,?,?)',
        (barcode, station_id, action, quantity, scanned_at)
    )

    # Update inventory quantity
    if action == 'inbound':
        db.execute('UPDATE items SET quantity = quantity + ? WHERE barcode = ?', (quantity, barcode))
    elif action == 'outbound':
        db.execute('UPDATE items SET quantity = MAX(0, quantity - ?) WHERE barcode = ?', (quantity, barcode))

    db.commit()

    response_data = {
        'status':      'ok',
        'barcode':     barcode,
        'item_name':   item['name'],
        'sku':         item['sku'],
        'location':    item['location'],
        'action':      action,
        'quantity':    quantity,
        'scanned_at':  scanned_at
    }

    logger.info(f'Scan recorded: {barcode} ({item["name"]}) at {station_id}')

    # Push webhook to cloud ERP if configured
    if CLOUD_WEBHOOK_URL:
        try:
            requests.post(CLOUD_WEBHOOK_URL, json=response_data, timeout=3)
        except Exception as e:
            logger.warning(f'Webhook delivery failed: {e}')

    return jsonify(response_data), 200


@app.route('/api/inventory', methods=['GET'])
def get_inventory():
    """Return current inventory state."""
    db    = get_db()
    items = db.execute('SELECT * FROM items ORDER BY location').fetchall()
    return jsonify([dict(row) for row in items]), 200


@app.route('/api/inventory/<barcode>', methods=['GET'])
def get_item(barcode):
    """Return a single item by barcode."""
    db   = get_db()
    item = db.execute('SELECT * FROM items WHERE barcode = ?', (barcode,)).fetchone()
    if not item:
        return jsonify(error='not found'), 404
    return jsonify(dict(item)), 200


@app.route('/api/scans/recent', methods=['GET'])
def get_recent_scans():
    """Return the last 50 scan events."""
    db     = get_db()
    limit  = min(int(request.args.get('limit', 50)), 500)
    events = db.execute(
        'SELECT * FROM scan_events ORDER BY id DESC LIMIT ?', (limit,)
    ).fetchall()
    return jsonify([dict(e) for e in events]), 200


@app.route('/api/health', methods=['GET'])
def health():
    return jsonify(status='ok', timestamp=time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())), 200


if __name__ == '__main__':
    init_db()
    logger.info('WMS API starting on http://0.0.0.0:5000')
    app.run(host='0.0.0.0', port=5000)

python3 wms_api.py

Verify the API is responding:

# Check health
curl http://localhost:5000/api/health

# View current inventory
curl http://localhost:5000/api/inventory

# Simulate a scan
curl -X POST http://localhost:5000/api/scans \
  -H "Content-Type: application/json" \
  -d '{"barcode":"1234567890128","station_id":"station_01","action":"scan"}'

Register both services as systemd units so they start automatically at boot:

sudo tee /etc/systemd/system/wms-api.service <<'EOF'
[Unit]
Description=Local WMS API
After=network.target

[Service]
ExecStart=/usr/bin/python3 /home/user/wms/wms_api.py
WorkingDirectory=/home/user/wms
Restart=always
User=user

[Install]
WantedBy=multi-user.target
EOF

sudo tee /etc/systemd/system/scanner-reader.service <<'EOF'
[Unit]
Description=Barcode Scanner Reader
After=wms-api.service

[Service]
ExecStart=/usr/bin/python3 /home/user/wms/scanner_reader.py
WorkingDirectory=/home/user/wms
Restart=always
User=root

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl enable wms-api scanner-reader
sudo systemctl start wms-api scanner-reader

Expose the WMS API with Localtonet

With the WMS API running on port 5000, create a Localtonet HTTP tunnel to give it a public HTTPS URL. Cloud ERP systems, SaaS logistics platforms, and remote dashboards will use this URL to query inventory and receive scan event notifications.

localtonet --authtoken <YOUR_TOKEN>

Go to the HTTP tunnel page, set local IP to 127.0.0.1 and port to 5000. Click Create and start the tunnel. The dashboard shows a public HTTPS URL such as https://abc123.localto.net.

Verify the public endpoint is reachable:

curl https://abc123.localto.net/api/health
curl https://abc123.localto.net/api/inventory

Register Localtonet as a service so the tunnel persists after reboots:

sudo localtonet --install-service --authtoken <YOUR_TOKEN>
sudo localtonet --start-service --authtoken <YOUR_TOKEN>

Connect Cloud ERP or SaaS to the WMS API

Once the tunnel is running, any cloud system can query the WMS API over HTTPS. The integration pattern depends on the cloud platform.

# Cloud-side Python script queries WMS inventory every 15 minutes
import requests
import schedule
import time

WMS_URL    = 'https://abc123.localto.net'
API_KEY    = 'your-api-key'   # See security section

def sync_inventory():
    response = requests.get(
        f'{WMS_URL}/api/inventory',
        headers={'X-API-Key': API_KEY},
        timeout=10
    )
    if response.status_code == 200:
        items = response.json()
        for item in items:
            print(f'{item["sku"]} ({item["name"]}): {item["quantity"]} units at {item["location"]}')
            # Update cloud ERP inventory records here
    else:
        print(f'WMS query failed: {response.status_code}')

schedule.every(15).minutes.do(sync_inventory)

while True:
    schedule.run_pending()
    time.sleep(1)

# Mobile app or web frontend queries item details
curl https://abc123.localto.net/api/inventory/1234567890128

Push Scan Events as Webhooks to Cloud Systems

Instead of polling, some cloud platforms prefer to receive webhook notifications when scan events occur. The WMS API already includes webhook delivery support via the CLOUD_WEBHOOK_URL variable in wms_api.py. Set it to your cloud ERP's inbound webhook URL:

CLOUD_WEBHOOK_URL = 'https://your-erp.com/webhooks/warehouse-scans'

Every scan event that the WMS API processes will be forwarded to that URL as an HTTP POST with a JSON body. The cloud platform receives real-time notifications without polling. If the webhook delivery fails, the scan event is still recorded locally in SQLite no data is lost on a temporary cloud connectivity interruption.

For n8n, Zapier, or Make.com workflows, create a webhook trigger in your workflow, copy the generated webhook URL, and set it as CLOUD_WEBHOOK_URL. Every scan in the warehouse triggers the workflow in real time.

Security Considerations

import os
from functools import wraps
from flask import request, jsonify

API_KEY = os.environ.get('WMS_API_KEY', 'change-this-secret-key')

def require_api_key(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        key = request.headers.get('X-API-Key') or request.args.get('api_key')
        if key != API_KEY:
            return jsonify(error='Unauthorized'), 401
        return f(*args, **kwargs)
    return decorated

# Apply to all routes that should be protected
@app.route('/api/inventory', methods=['GET'])
@require_api_key
def get_inventory():
    # ...
    pass

Frequently Asked Questions