Title: No title
Domain: github.com
address: — single tag (supports B3:0/5 bit notation)
POST /api/tag/ — write value (bit-level write supported)
POST /api/tag//description — set human-readable label
POST /api/tag//group — assign product:station group
GET /api/io — live I:0 / O:0 bit arrays from cached tags
GET /api/health — server + PLC connection status
Tag metadata (descriptions, groups) is persisted to tag_metadata.json.
"""
import json
import logging
import logging.handlers
import os
import sys
import threading
import time
from datetime import datetime, timezone
import yaml
from flask import Flask, jsonify, request
from pycomm3 import SLCDriver
# ── Config ────────────────────────────────────────────────────────────────────
with open('config.yaml') as f:
CFG = yaml.safe_load(f)
PLC_IP = CFG['plc']['ip']
PROBE_RANGES = CFG['probe']['ranges']
PROBE_INTERVAL = CFG['probe']['interval_seconds']
SERVER_HOST = CFG['server']['host']
SERVER_PORT = CFG['server']['port']
INPUT_COUNT = CFG['io']['input_count']
OUTPUT_COUNT = CFG['io']['output_count']
LOG_FILE = CFG.get('logging', {}).get('file', 'tag_server.log')
LOG_MAX_BYTES = CFG.get('logging', {}).get('max_bytes', 5_000_000)
LOG_BACKUP_COUNT= CFG.get('logging', {}).get('backup_count', 3)
SUBFILE_TYPES = {'T4', 'C5', 'R6'}
TAG_METADATA_FILE = 'tag_metadata.json'
# ── Logging ───────────────────────────────────────────────────────────────────
log = logging.getLogger('tag_server')
log.setLevel(logging.DEBUG)
_fmt = logging.Formatter('%(asctime)s [%(levelname)s] %(message)s')
_fh = logging.handlers.RotatingFileHandler(LOG_FILE, maxBytes=LOG_MAX_BYTES, backupCount=LOG_BACKUP_COUNT)
_fh.setFormatter(_fmt)
log.addHandler(_fh)
_sh = logging.StreamHandler(sys.stdout)
_sh.setFormatter(_fmt)
log.addHandler(_sh)
# ── State ─────────────────────────────────────────────────────────────────────
app = Flask(__name__)
tags = {}
tags_lock = threading.Lock()
last_probe_time = None # UTC datetime of the most recent completed probe
last_probe_ok = False # True if the last probe connected successfully
if os.path.exists(TAG_METADATA_FILE):
with open(TAG_METADATA_FILE) as f:
tag_metadata = json.load(f)
else:
tag_metadata = {}
def _atomic_save_metadata(data):
"""Write metadata to a temp file then rename — safe against mid-write crashes."""
tmp = TAG_METADATA_FILE + '.tmp'
with open(tmp, 'w') as f:
json.dump(data, f, indent=2)
os.replace(tmp, TAG_METADATA_FILE)
# ── Tag discovery ─────────────────────────────────────────────────────────────
def _do_probe():
"""
Connect to the PLC, read all configured data-file ranges, and return a
dict of {address: tag_dict}. Returns an empty dict on connection failure.
"""
global last_probe_time, last_probe_ok
discovered = {}
try:
with SLCDriver(PLC_IP) as plc:
for file_name, count in PROBE_RANGES.items():
for i in range(count):
addr = f"{file_name}:{i}"
try:
resp = plc.read(addr)
if resp and resp.value is not None:
meta = tag_metadata.get(addr, {})
discovered[addr] = {
'address': addr,
'value': resp.value,
'type': type(resp.value).__name__,
'description': meta.get('description', ''),
'group': meta.get('group', {}),
'last_updated': datetime.now(timezone.utc).isoformat(),
}
except Exception as e:
log.debug("Read failed %s: %s", addr, e)
if file_name in SUBFILE_TYPES:
for suffix in ('.ACC', '.PRE'):
ext = f"{file_name}:{i}{suffix}"
try:
resp = plc.read(ext)
if resp and resp.value is not None:
meta = tag_metadata.get(ext, {})
discovered[ext] = {
'address': ext,
'value': resp.value,
'type': type(resp.value).__name__,
'description': meta.get('description', ''),
'group': meta.get('group', {}),
'last_updated': datetime.now(timezone.utc).isoformat(),
}
except Exception as e:
log.debug("Read failed %s: %s", ext, e)
last_probe_ok = True
except Exception as e:
log.error("Probe connection error: %s", e)
last_probe_ok = False
last_probe_time = datetime.now(timezone.utc)
return discovered
def probe_tags_periodic():
"""Background thread: probe once immediately, then on PROBE_INTERVAL."""
while True:
discovered = _do_probe()
if discovered:
with tags_lock:
tags.clear()
tags.update(discovered)
log.info("Probe complete — %d tags.", len(discovered))
time.sleep(PROBE_INTERVAL)
# ── Helpers ───────────────────────────────────────────────────────────────────
def find_tag(address):
"""Case-insensitive lookup, ignoring any trailing /bit suffix."""
key = address.lower().split('/')[0]
with tags_lock:
for tag in tags:
if tag.lower() == key:
return tag
return None
def _update_tag_value(key, new_value):
"""Update a tag's value and last_updated timestamp in the cache."""
with tags_lock:
tags[key]['value'] = new_value
tags[key]['last_updated'] = datetime.now(timezone.utc).isoformat()
# ── Routes ────────────────────────────────────────────────────────────────────
@app.route('/api/health', methods=['GET'])
def health():
with tags_lock:
tag_count = len(tags)
return jsonify({
'status': 'ok' if last_probe_ok else 'plc_unreachable',
'plc_ip': PLC_IP,
'tag_count': tag_count,
'last_probe': last_probe_time.isoformat() if last_probe_time else None,
'probe_interval': PROBE_INTERVAL,
})
@app.route('/api/tags', methods=['GET'])
def get_tags():
with tags_lock:
return jsonify(dict(tags))
@app.route('/api/tag/Links: