640 lines
27 KiB
Python
640 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
# Offline tests for the MTConnect agent prototype.
|
|
#
|
|
# Exercises the probe generator, the probe<->registry consistency, the
|
|
# observation buffer + streams, and the asset builder without a running
|
|
# LinuxCNC. Run: python3 test_mtc.py
|
|
import os
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import xml.etree.ElementTree as ET
|
|
|
|
# The mtc package installs to $EMC2_HOME/lib/python; make it importable when this
|
|
# is run from the config directory (rip-environment already puts it on the path).
|
|
_EMC2_HOME = os.environ.get("EMC2_HOME")
|
|
if _EMC2_HOME:
|
|
sys.path.insert(0, os.path.join(_EMC2_HOME, "lib", "python"))
|
|
|
|
from mtc.ini_reader import IniReader
|
|
from mtc import kinematics as kin
|
|
from mtc.device_model import DeviceConfig, probe_xml, MTC_NS, EXT_NS
|
|
from mtc.observations import build_dataitems
|
|
from mtc.streams import (ObservationBuffer, current_xml, sample_xml, assets_xml,
|
|
STREAMS_NS)
|
|
from mtc.lcnc_source import ToolAsset, LcncSource
|
|
from mtc.agent import AgentState
|
|
from mtc.http_agent import HttpAgent
|
|
from mtc.shdr_agent import ShdrAgent
|
|
|
|
# Resolve config paths relative to this file, not the current directory, so the
|
|
# suite runs from any CWD (e.g. when wired into tests/ and driven by runtests).
|
|
_HERE = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def _cfg(relpath):
|
|
return os.path.normpath(os.path.join(_HERE, relpath))
|
|
|
|
|
|
CONFIGS = {
|
|
"3axis": _cfg("../axis.ini"),
|
|
"5axis": _cfg("../vismach/5axis/table-rotary-tilting/xyzac-trt.ini"),
|
|
}
|
|
TS = "2026-07-01T12:00:00.000Z"
|
|
|
|
|
|
def load(path):
|
|
ini = IniReader(path)
|
|
model = kin.build_model(ini)
|
|
config = DeviceConfig.from_ini(ini)
|
|
return ini, model, config
|
|
|
|
|
|
def test_kinematics():
|
|
_, model, _ = load(CONFIGS["5axis"])
|
|
assert model.kins_module == "xyzac-trt-kins", model.kins_module
|
|
assert model.coordinates == "XYZAC", model.coordinates
|
|
assert model.joints_count == 5
|
|
jmap = model.joint_axis_map()
|
|
assert jmap == {0: "X", 1: "Y", 2: "Z", 3: "A", 4: "C"}, jmap
|
|
kinds = {a.letter: a.kind for a in model.axes}
|
|
assert kinds["A"] == "ANGULAR" and kinds["X"] == "LINEAR", kinds
|
|
# A rotates about X, C about Z
|
|
vecs = {a.letter: a.vector for a in model.axes}
|
|
assert vecs["A"] == (1.0, 0.0, 0.0) and vecs["C"] == (0.0, 0.0, 1.0), vecs
|
|
print("ok kinematics (5-axis XYZAC mapping + rotary vectors)")
|
|
|
|
|
|
def test_probe_registry_consistency():
|
|
for label, path in CONFIGS.items():
|
|
_, model, config = load(path)
|
|
xml = probe_xml(model, config, creation_time=TS)
|
|
root = ET.fromstring(xml)
|
|
probe_ids = {}
|
|
for di in root.iter("{%s}DataItem" % MTC_NS):
|
|
probe_ids[di.get("id")] = (di.get("category"), di.get("type"),
|
|
di.get("subType"))
|
|
for d in build_dataitems(model, config):
|
|
assert d.id in probe_ids, "%s: %s missing from probe" % (label, d.id)
|
|
cat, typ, sub = probe_ids[d.id]
|
|
expected_type = ("x:" + d.type) if d.ext else d.type
|
|
assert (cat, typ, sub) == (d.category, expected_type, d.subType), \
|
|
"%s: %s attrs %s != %s" % (label, d.id, (cat, typ, sub),
|
|
(d.category, d.type, d.subType))
|
|
# kinematics extension present
|
|
kext = root.find(".//{%s}Kinematics" % EXT_NS)
|
|
assert kext is not None, "%s: no kinematics extension" % label
|
|
assert kext.get("module") == model.kins_module
|
|
print("ok probe/registry consistency (%s, %d DataItems)"
|
|
% (label, len(probe_ids)))
|
|
|
|
|
|
def test_buffer_and_streams():
|
|
_, model, config = load(CONFIGS["3axis"])
|
|
dataitems = build_dataitems(model, config)
|
|
buf = ObservationBuffer(dataitems)
|
|
|
|
n = buf.ingest({"execution": "READY", "pos_x": 1.0, "pos_y": 2.0}, TS)
|
|
assert n == 3, n
|
|
assert buf.ingest({"pos_x": 1.0}, TS) == 0, "unchanged value must not record"
|
|
assert buf.ingest({"pos_x": 5.0}, TS) == 1
|
|
assert buf.last_sequence == 4, buf.last_sequence
|
|
|
|
# /current: every DataItem present, unset ones UNAVAILABLE
|
|
cur = ET.fromstring(current_xml(buf, config, TS))
|
|
observed = {e.get("dataItemId"): e.text
|
|
for e in cur.iter() if e.get("dataItemId")}
|
|
assert observed["pos_x"] == "5.0", observed.get("pos_x")
|
|
assert observed["pos_z"] == "UNAVAILABLE", observed.get("pos_z")
|
|
assert observed["execution"] == "READY"
|
|
|
|
# /sample: sequence range 1..2 returns exactly the first two observations
|
|
smp = ET.fromstring(sample_xml(buf, config, TS, from_seq=1, count=2))
|
|
seqs = sorted(int(e.get("sequence")) for e in smp.iter()
|
|
if e.get("sequence"))
|
|
assert seqs == [1, 2], seqs
|
|
# ComponentStream grouping + Samples/Events wrappers exist
|
|
assert cur.find(".//{%s}ComponentStream" % STREAMS_NS) is not None
|
|
print("ok buffer + current/sample streams")
|
|
|
|
|
|
def test_work_offset_table():
|
|
_, model, config = load(CONFIGS["3axis"])
|
|
buf = ObservationBuffer(build_dataitems(model, config))
|
|
buf.ingest({"workoffset": {"G55": {"X": 1.5, "Y": -2.0, "Z": 0.0}}}, TS)
|
|
cur = ET.fromstring(current_xml(buf, config, TS))
|
|
# WORK_OFFSET is a standard EVENT type; TABLE representation streams as
|
|
# <WorkOffsetTable> (not <WorkOffset>).
|
|
wo = cur.find(".//{%s}WorkOffsetTable" % STREAMS_NS)
|
|
assert wo is not None and wo.get("count") == "1", wo
|
|
entry = wo.find("{%s}Entry" % STREAMS_NS)
|
|
assert entry.get("key") == "G55", entry.attrib
|
|
cells = {c.get("key"): c.text for c in entry.findall("{%s}Cell" % STREAMS_NS)}
|
|
assert cells == {"X": "1.5", "Y": "-2", "Z": "0"}, cells
|
|
# probe advertises it as a standard WORK_OFFSET EVENT with TABLE representation
|
|
probe = ET.fromstring(probe_xml(model, config))
|
|
di = [d for d in probe.iter("{%s}DataItem" % MTC_NS)
|
|
if d.get("id") == "workoffset"][0]
|
|
assert di.get("representation") == "TABLE" and di.get("type") == "WORK_OFFSET"
|
|
assert di.get("category") == "EVENT", di.get("category")
|
|
print("ok work offset TABLE (WorkOffsetTable Entry/Cell + probe representation)")
|
|
|
|
|
|
def test_tool_offset_and_rotation():
|
|
_, model, config = load(CONFIGS["3axis"])
|
|
buf = ObservationBuffer(build_dataitems(model, config))
|
|
buf.ingest({"tooloffset": {"T3": {"X": 0.0, "Y": 0.0, "Z": 2.5}},
|
|
"xyrotation": 30.0}, TS)
|
|
cur = ET.fromstring(current_xml(buf, config, TS))
|
|
|
|
# TOOL_OFFSET is a standard EVENT type -> <ToolOffsetTable> in STREAMS_NS.
|
|
to = cur.find(".//{%s}ToolOffsetTable" % STREAMS_NS)
|
|
assert to is not None, "no ToolOffsetTable in stream"
|
|
entry = to.find("{%s}Entry" % STREAMS_NS)
|
|
assert entry.get("key") == "T3"
|
|
zc = [c for c in entry.findall("{%s}Cell" % STREAMS_NS) if c.get("key") == "Z"][0]
|
|
assert zc.text == "2.5", zc.text
|
|
# xyrotation has no standard MTConnect type, so it stays a LinuxCNC extension.
|
|
rot = cur.find(".//{%s}CoordinateRotation" % EXT_NS)
|
|
assert rot is not None and rot.text == "30.0", rot
|
|
|
|
probe = ET.fromstring(probe_xml(model, config))
|
|
types = {d.get("id"): d.get("type") for d in probe.iter("{%s}DataItem" % MTC_NS)}
|
|
assert types["tooloffset"] == "TOOL_OFFSET", types.get("tooloffset") # standard now
|
|
assert types["xyrotation"] == "x:COORDINATE_ROTATION", types.get("xyrotation")
|
|
print("ok tool offset (standard ToolOffsetTable) + XY rotation (x: extension)")
|
|
|
|
|
|
def test_assets():
|
|
_, _, config = load(CONFIGS["3axis"])
|
|
tools = [
|
|
ToolAsset(tool_no=1, pocket=1, in_spindle=True, diameter=6.35, length_z=50.0,
|
|
comment="1/4 flat endmill "), # trailing space should be trimmed
|
|
ToolAsset(tool_no=2, pocket=5, in_spindle=False, diameter=3.0),
|
|
]
|
|
root = ET.fromstring(assets_xml(tools, config, TS))
|
|
ns = "{urn:mtconnect.org:MTConnectAssets:1.7}"
|
|
cts = root.findall(".//%sCuttingTool" % ns)
|
|
assert len(cts) == 2, len(cts)
|
|
assert cts[0].get("assetId") == "tool-1"
|
|
assert cts[0].get("serialNumber") == "tool-1", cts[0].attrib # required attr
|
|
loc = cts[0].find(".//%sLocation" % ns)
|
|
assert loc.get("type") == "SPINDLE" and loc.text == "1", loc.attrib
|
|
desc = cts[0].find("%sDescription" % ns)
|
|
assert desc is not None and desc.text == "1/4 flat endmill", repr(desc.text)
|
|
assert cts[1].find("%sDescription" % ns) is None # empty comment -> no element
|
|
# length offset is FunctionalLength (LF), not BodyLengthMax
|
|
lf = cts[0].find(".//%sFunctionalLength" % ns)
|
|
assert lf is not None and lf.get("code") == "LF" and lf.text == "50", ET.tostring(cts[0])
|
|
assert cts[0].find(".//%sBodyLengthMax" % ns) is None
|
|
# diameter is a CuttingItem measurement, not a tool-level one
|
|
dia = cts[0].find(".//%sCuttingItems/%sCuttingItem/%sMeasurements/%sCuttingDiameter"
|
|
% (ns, ns, ns, ns))
|
|
assert dia is not None and dia.text == "6.35" and dia.get("code") == "DC", ET.tostring(cts[0])
|
|
print("ok assets (serialNumber + FunctionalLength + CuttingItem diameter + comment)")
|
|
|
|
|
|
def test_probe_standard_blocks():
|
|
# spindle Specifications, Coolant system, and the kinematics extension living
|
|
# inside the (schema-valid) Description host.
|
|
_, model, config = load(_cfg("example.ini"))
|
|
root = ET.fromstring(probe_xml(model, config, creation_time=TS))
|
|
ns = "{%s}" % MTC_NS
|
|
spec = root.find(".//%sRotary[@id='spindle']//%sSpecification" % (ns, ns))
|
|
assert spec is not None and spec.get("type") == "ROTARY_VELOCITY", "no spindle spec"
|
|
assert spec.find("%sMaximum" % ns).text == "6000"
|
|
cool = root.find(".//%sCoolant" % ns)
|
|
assert cool is not None, "no Coolant component"
|
|
assert cool.find(".//%sDataItem[@id='coolant_flood']" % ns).get("type") == "x:FLOOD"
|
|
# x:Kinematics is nested in a Description (the lax xs:any extension point)
|
|
kin_ext = root.find(".//%sDescription/{%s}Kinematics" % (ns, EXT_NS))
|
|
assert kin_ext is not None, "x:Kinematics not under Description"
|
|
assert kin_ext.get("nativeLinearUnits") == "INCH"
|
|
# Agent device is present (schema-required alongside the machine Device)
|
|
assert root.find(".//%sAgent" % ns) is not None, "no Agent device"
|
|
print("ok probe standard blocks (spindle Spec, Coolant, kinematics host, Agent)")
|
|
|
|
|
|
def test_spindle_constraints():
|
|
ns = "{%s}" % MTC_NS
|
|
|
|
def spdl_cmd(path):
|
|
_, model, config = load(path)
|
|
root = ET.fromstring(probe_xml(model, config, creation_time=TS))
|
|
for di in root.iter("%sDataItem" % ns):
|
|
if di.get("id") == "spdl_speed_cmd":
|
|
return di
|
|
raise AssertionError("spdl_speed_cmd missing")
|
|
|
|
# example.ini declares [SPINDLE_0] MIN/MAX_FORWARD_VELOCITY -> Constraints
|
|
di = spdl_cmd(_cfg("example.ini"))
|
|
con = di.find("%sConstraints" % ns)
|
|
assert con is not None, "no Constraints on spdl_speed_cmd"
|
|
assert con.find("%sMinimum" % ns).text == "100", ET.tostring(di)
|
|
assert con.find("%sMaximum" % ns).text == "6000", ET.tostring(di)
|
|
|
|
# a config without spindle velocity limits emits no Constraints (no bogus
|
|
# ~2.1e9 default range fabricated).
|
|
assert spdl_cmd(CONFIGS["3axis"]).find("%sConstraints" % ns) is None
|
|
print("ok spindle speed range Constraints (from [SPINDLE_0], omitted when absent)")
|
|
|
|
|
|
def test_schema_validation():
|
|
# Gated: needs the xmlschema package (XSD 1.1) and the official 1.7 XSDs
|
|
# (local dir via $MTC_XSD_DIR, else fetched). Skips cleanly otherwise so
|
|
# the core suite stays dependency- and network-free.
|
|
import os
|
|
try:
|
|
import io
|
|
import xmlschema
|
|
except ImportError:
|
|
print("skip schema validation (xmlschema not installed)")
|
|
return
|
|
from mtc.agent import asset_dir
|
|
xsd_dir = os.environ.get("MTC_XSD_DIR")
|
|
base = (lambda n: os.path.join(xsd_dir, n)) if xsd_dir else \
|
|
(lambda n: "http://schemas.mtconnect.org/schemas/" + n)
|
|
ext = os.path.join(asset_dir(), "mtconnect-linuxcnc-1.xsd")
|
|
state = AgentState(_cfg("example.ini"))
|
|
state.poll_once()
|
|
checks = [("MTConnectDevices_1.7.xsd", state.probe_document()),
|
|
("MTConnectAssets_1.7.xsd", state.assets_document()),
|
|
(ext, state.current_document())]
|
|
try:
|
|
for xsd, doc in checks:
|
|
loc = xsd if xsd == ext else base(xsd)
|
|
schema = xmlschema.XMLSchema11(loc, validation="skip")
|
|
errs = list(schema.iter_errors(io.StringIO(doc)))
|
|
assert not errs, "%s: %s" % (xsd, (errs[0].reason or errs[0].message))
|
|
except (OSError, xmlschema.XMLSchemaException) as exc:
|
|
if xsd_dir:
|
|
raise
|
|
print("skip schema validation (schemas unreachable: %s)" % exc)
|
|
return
|
|
print("ok schema validation (probe/assets base, current base+extension)")
|
|
|
|
|
|
def test_source_offline():
|
|
_, model, config = load(CONFIGS["3axis"])
|
|
src = LcncSource(model, config)
|
|
# No linuxcnc extension / no running instance -> graceful empty results.
|
|
if not src.available():
|
|
assert src.sample_values() == {}
|
|
assert src.tool_assets() == []
|
|
print("ok lcnc_source offline (graceful, no stat)")
|
|
else:
|
|
print("ok lcnc_source live (stat available)")
|
|
|
|
|
|
def test_http_endpoints():
|
|
state = AgentState(CONFIGS["3axis"])
|
|
state.buffer.ingest({"execution": "ACTIVE", "pos_x": 3.5}, TS)
|
|
state._assets = [ToolAsset(tool_no=7, pocket=7, in_spindle=True, diameter=10.0)]
|
|
|
|
http = HttpAgent(state, host="127.0.0.1", port=0)
|
|
http.start()
|
|
try:
|
|
base = "http://127.0.0.1:%d" % http.port
|
|
|
|
def get(path):
|
|
with urllib.request.urlopen(base + path, timeout=5) as r:
|
|
return r.status, r.read().decode()
|
|
|
|
st, body = get("/probe")
|
|
assert st == 200 and "MTConnectDevices" in body, st
|
|
st, body = get("/current")
|
|
assert st == 200 and "ACTIVE" in body, body[:200]
|
|
st, body = get("/sample?from=1&count=10")
|
|
assert st == 200 and 'sequence="1"' in body, body[:200]
|
|
st, body = get("/assets")
|
|
assert st == 200 and "tool-7" in body, body[:200]
|
|
|
|
# Bad route and out-of-range sequence return MTConnectError docs.
|
|
try:
|
|
get("/nope")
|
|
assert False, "expected 404"
|
|
except urllib.error.HTTPError as e:
|
|
assert e.code == 404 and "MTConnectError" in e.read().decode()
|
|
try:
|
|
get("/sample?from=999999")
|
|
assert False, "expected 406"
|
|
except urllib.error.HTTPError as e:
|
|
assert e.code == 406 and "OUT_OF_RANGE" in e.read().decode()
|
|
print("ok embedded HTTP agent (probe/current/sample/assets + errors)")
|
|
finally:
|
|
http.stop()
|
|
|
|
|
|
def _is_iso_ts(field):
|
|
# e.g. 2026-07-28T12:00:00.000Z -- cheap structural check (no regex import).
|
|
return (len(field) >= 20 and field[4] == "-" and field[7] == "-"
|
|
and field[10] == "T" and field.endswith("Z"))
|
|
|
|
|
|
def test_shdr():
|
|
# Offline SHDR adapter test: no LinuxCNC, values injected via _last_values
|
|
# (what AgentState.latest_values() returns), which is the poll snapshot the
|
|
# adapter diffs and streams.
|
|
state = AgentState(CONFIGS["3axis"])
|
|
|
|
shdr = ShdrAgent(state, port=0, host="127.0.0.1")
|
|
shdr.start()
|
|
client = None
|
|
try:
|
|
client = socket.create_connection(("127.0.0.1", shdr.port), timeout=2)
|
|
client.settimeout(2)
|
|
|
|
# Wait until the server has registered the client (its handler thread ran
|
|
# _add_client) so the broadcast below is guaranteed to reach us.
|
|
deadline = time.time() + 2
|
|
while time.time() < deadline:
|
|
with shdr._lock:
|
|
registered = len(shdr._clients)
|
|
if registered:
|
|
break
|
|
time.sleep(0.01)
|
|
assert registered == 1, registered
|
|
|
|
buf = [b""]
|
|
|
|
def recv_line():
|
|
while b"\n" not in buf[0]:
|
|
chunk = client.recv(4096)
|
|
if not chunk:
|
|
raise AssertionError("connection closed before a full line")
|
|
buf[0] += chunk
|
|
line, _, rest = buf[0].partition(b"\n")
|
|
buf[0] = rest
|
|
return line.decode()
|
|
|
|
# (a) ingest/poll some values, then stream the changes. A structured
|
|
# (TABLE) value must be skipped, never emitted as a malformed line.
|
|
state._last_values = {"execution": "ACTIVE", "pos_x": 3.5,
|
|
"workoffset": {"G54": {"X": 1.0}}}
|
|
shdr.publish_changes()
|
|
|
|
line = recv_line()
|
|
fields = line.split("|")
|
|
assert len(fields) >= 3, "not a well-formed SHDR line: %r" % line
|
|
assert _is_iso_ts(fields[0]), "first field not ISO ts: %r" % fields[0]
|
|
ids = fields[1::2]
|
|
assert "execution" in ids and "pos_x" in ids, ids
|
|
assert "workoffset" not in line, "TABLE value must be skipped: %r" % line
|
|
|
|
# unchanged values produce nothing; a real change is streamed
|
|
shdr.publish_changes()
|
|
state._last_values = {"execution": "READY", "pos_x": 3.5,
|
|
"workoffset": {"G54": {"X": 1.0}}}
|
|
shdr.publish_changes()
|
|
line2 = recv_line()
|
|
f2 = line2.split("|")
|
|
assert _is_iso_ts(f2[0]) and "execution" in f2[1::2], line2
|
|
assert "READY" in line2, line2
|
|
|
|
# (b) heartbeat: '* PING' -> '* PONG 10000'
|
|
client.sendall(b"* PING\n")
|
|
pong = recv_line()
|
|
assert pong == "* PONG 10000", repr(pong)
|
|
|
|
print("ok SHDR adapter (initial/changed ts|id|value lines, TABLE skipped, PING/PONG)")
|
|
finally:
|
|
if client is not None:
|
|
client.close()
|
|
shdr.stop()
|
|
|
|
|
|
def test_hal_items():
|
|
import tempfile
|
|
from mtc.hal_items import HalSource
|
|
ini_text = (
|
|
"[EMC]\nMACHINE = hal-item-demo\n"
|
|
"[TRAJ]\nCOORDINATES = X Y Z\nLINEAR_UNITS = inch\n"
|
|
"[KINS]\nKINEMATICS = trivkins\nJOINTS = 3\n"
|
|
"[AXIS_X]\nMIN_LIMIT = -10\nMAX_LIMIT = 10\n"
|
|
"[AXIS_Y]\nMIN_LIMIT = -6\nMAX_LIMIT = 6\n"
|
|
"[AXIS_Z]\nMIN_LIMIT = -8\nMAX_LIMIT = 0\n"
|
|
"[MTCONNECT]\n"
|
|
"HAL_ITEM = pin=spindle.0.load, id=spindle_load, type=LOAD, units=PERCENT, component=spindle\n"
|
|
"HAL_ITEM = pin=hm2.temp, id=board_temp, type=TEMPERATURE, units=CELSIUS\n"
|
|
"HAL_ITEM = pin=bad.one, id=bad_custom, type=WIDGET_COUNT\n" # rejected
|
|
"HAL_ITEM = pin=x, id=1nope, type=LOAD\n" # bad id, rejected
|
|
)
|
|
d = tempfile.mkdtemp()
|
|
p = os.path.join(d, "hal.ini")
|
|
with open(p, "w") as fh:
|
|
fh.write(ini_text)
|
|
|
|
ini = IniReader(p)
|
|
model = kin.build_model(ini)
|
|
config = DeviceConfig.from_ini(ini)
|
|
# only the two valid items survive parsing
|
|
assert [i.id for i in config.hal_items] == ["spindle_load", "board_temp"], \
|
|
[i.id for i in config.hal_items]
|
|
|
|
# they appear in the registry as SAMPLE items of the right type
|
|
reg = {d.id: d for d in build_dataitems(model, config)}
|
|
assert reg["spindle_load"].type == "LOAD" and reg["spindle_load"].category == "SAMPLE"
|
|
assert reg["spindle_load"].comp_id == "spindle" # targeted the spindle
|
|
assert reg["board_temp"].comp_id == "sensors" # default Sensor host
|
|
|
|
# probe: generic item under Auxiliaries/Sensor, targeted item under the spindle
|
|
probe = ET.fromstring(probe_xml(model, config))
|
|
ns = "{%s}" % MTC_NS
|
|
sensor = probe.find(".//%sAuxiliaries/%sComponents/%sSensor" % (ns, ns, ns))
|
|
assert sensor is not None, "no Auxiliaries/Sensor"
|
|
sensor_ids = {di.get("id") for di in sensor.iter("%sDataItem" % ns)}
|
|
assert "board_temp" in sensor_ids and "spindle_load" not in sensor_ids, sensor_ids
|
|
sp = probe.find(".//%sRotary[@id='spindle']" % ns)
|
|
assert "spindle_load" in {di.get("id") for di in sp.iter("%sDataItem" % ns)}
|
|
|
|
# offline HalSource (no hal extension) yields nothing, gracefully
|
|
assert HalSource(config.hal_items).sample_values() == {}
|
|
print("ok HAL items (parse/validate, placement, offline HalSource)")
|
|
|
|
|
|
def test_solid_models():
|
|
import os
|
|
import tempfile
|
|
from mtc.models import MachineModels, MeshRef
|
|
|
|
_, model, config = load(CONFIGS["3axis"])
|
|
|
|
# No models configured -> no SolidModel / Configuration emitted.
|
|
plain = ET.fromstring(probe_xml(model, config))
|
|
assert plain.find(".//{%s}SolidModel" % MTC_NS) is None
|
|
|
|
tmp = tempfile.mkdtemp()
|
|
path = os.path.join(tmp, "z_head.stl")
|
|
with open(path, "wb") as fh:
|
|
fh.write(b"solid dummy\nendsolid dummy\n")
|
|
ref = MeshRef("z_head.stl", path, "STL", "model/stl", True)
|
|
mm = MachineModels(units="MILLIMETER", base=ref, axis={"Z": ref},
|
|
chain=["dev_base", "axis_x", "axis_y", "axis_z"],
|
|
parents={"axis_x": "dev_base", "axis_y": "axis_x",
|
|
"axis_z": "axis_y"},
|
|
served={"z_head.stl": ref})
|
|
|
|
probe = ET.fromstring(probe_xml(model, config, models=mm))
|
|
sms = probe.findall(".//{%s}SolidModel" % MTC_NS)
|
|
hrefs = {sm.get("href") for sm in sms}
|
|
assert "/models/z_head.stl" in hrefs, hrefs
|
|
assert probe.find(".//{%s}CoordinateSystem[@type='MACHINE']" % MTC_NS) is not None
|
|
# serial chain: Z link hangs off Y's motion
|
|
mz = [m for m in probe.iter("{%s}Motion" % MTC_NS) if m.get("id") == "motion_z"][0]
|
|
assert mz.get("parentIdRef") == "motion_y", mz.attrib
|
|
|
|
# branched: re-root Z at the base (knee-mill quill) -> no parentIdRef
|
|
mm.parents["axis_z"] = "dev_base"
|
|
probe2 = ET.fromstring(probe_xml(model, config, models=mm))
|
|
mz2 = [m for m in probe2.iter("{%s}Motion" % MTC_NS) if m.get("id") == "motion_z"][0]
|
|
assert mz2.get("parentIdRef") is None, mz2.attrib
|
|
|
|
# inverted axis: Motion <Axis> vector is negated (work-carrying link)
|
|
mm.invert = {"Z"}
|
|
probe3 = ET.fromstring(probe_xml(model, config, models=mm))
|
|
mz3 = [m for m in probe3.iter("{%s}Motion" % MTC_NS) if m.get("id") == "motion_z"][0]
|
|
assert mz3.find("{%s}Axis" % MTC_NS).text == "0 0 -1", mz3.find("{%s}Axis" % MTC_NS).text
|
|
|
|
# /models route serves the file; unknown model -> 404
|
|
state = AgentState(CONFIGS["3axis"])
|
|
state.models = mm
|
|
http = HttpAgent(state, host="127.0.0.1", port=0)
|
|
http.start()
|
|
try:
|
|
with urllib.request.urlopen(
|
|
"http://127.0.0.1:%d/models/z_head.stl" % http.port, timeout=5) as r:
|
|
assert r.status == 200 and b"endsolid" in r.read()
|
|
try:
|
|
urllib.request.urlopen("http://127.0.0.1:%d/models/nope.stl" % http.port)
|
|
assert False, "expected 404"
|
|
except urllib.error.HTTPError as e:
|
|
assert e.code == 404
|
|
print("ok solid models (SolidModel/CoordinateSystems/chain + /models route)")
|
|
finally:
|
|
http.stop()
|
|
|
|
|
|
def test_auto_geometry():
|
|
import os
|
|
import tempfile
|
|
from mtc.models import build_models
|
|
|
|
ini_text = (
|
|
"[EMC]\nMACHINE = auto-demo\n"
|
|
"[TRAJ]\nCOORDINATES = X Y Z\nLINEAR_UNITS = inch\n"
|
|
"[KINS]\nKINEMATICS = trivkins\nJOINTS = 3\n"
|
|
"[AXIS_X]\nMIN_LIMIT = -10\nMAX_LIMIT = 10\n"
|
|
"[AXIS_Y]\nMIN_LIMIT = -6\nMAX_LIMIT = 6\n"
|
|
"[AXIS_Z]\nMIN_LIMIT = -8\nMAX_LIMIT = 0\n"
|
|
"[MTCONNECT]\nMODEL_AUTO = 1\nMODEL_PARENT_Z = BASE\nMODEL_INVERT = X Y\n"
|
|
)
|
|
d = tempfile.mkdtemp()
|
|
p = os.path.join(d, "auto.ini")
|
|
with open(p, "w") as fh:
|
|
fh.write(ini_text)
|
|
|
|
ini = IniReader(p)
|
|
model = kin.build_model(ini)
|
|
config = DeviceConfig.from_ini(ini)
|
|
mm = build_models(ini, model, config)
|
|
assert mm.enabled() and mm.base is not None
|
|
assert set(mm.axis) == {"X", "Y", "Z"}, set(mm.axis)
|
|
# Geometry is served in the MTConnect canonical unit (millimetre); the inch
|
|
# travel limits are scaled up by 25.4 in the generated boxes.
|
|
assert mm.units == "MILLIMETER", mm.units
|
|
assert all(v.startswith("solid") for v in mm.generated.values())
|
|
# X box half-width tracks the 20" span * 25.4 -> ~500 mm scale, not ~20.
|
|
xs = [float(tok) for line in mm.generated["axis_x.stl"].splitlines()
|
|
if line.strip().startswith("vertex") for tok in [line.split()[1]]]
|
|
assert max(xs) > 100, max(xs) # would be < 20 if still in inches
|
|
|
|
probe = ET.fromstring(probe_xml(model, config, models=mm))
|
|
hrefs = {sm.get("href") for sm in probe.findall(".//{%s}SolidModel" % MTC_NS)}
|
|
assert {"/models/dev_base.stl", "/models/axis_x.stl", "/models/axis_z.stl"} <= hrefs, hrefs
|
|
|
|
# the agent serves the generated STL bytes (no file on disk)
|
|
state = AgentState(p)
|
|
http = HttpAgent(state, host="127.0.0.1", port=0)
|
|
http.start()
|
|
try:
|
|
with urllib.request.urlopen(
|
|
"http://127.0.0.1:%d/models/axis_x.stl" % http.port, timeout=5) as r:
|
|
assert r.status == 200 and b"endsolid" in r.read()
|
|
print("ok auto geometry (generated boxes served + probe SolidModels)")
|
|
finally:
|
|
http.stop()
|
|
|
|
|
|
def test_ha_discovery():
|
|
# ha.py is an optional contrib (Home Assistant is not part of the core agent),
|
|
# so skip cleanly if the contrib is not present.
|
|
import json
|
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
|
"contrib"))
|
|
try:
|
|
import ha
|
|
except ImportError:
|
|
print("skip HA discovery (contrib not installed)")
|
|
return
|
|
|
|
_, model, config = load(CONFIGS["3axis"])
|
|
sensors = ha.build_sensors(model, config)
|
|
keys = {s["key"] for s in sensors}
|
|
assert {"execution", "spdl_speed", "pos_x", "pos_y", "pos_z"} <= keys, keys
|
|
|
|
px = [s for s in sensors if s["key"] == "pos_x"][0]
|
|
d = ha.discovery_payload(px, config, "st/topic", "av/topic")
|
|
assert d["state_topic"] == "st/topic"
|
|
assert d["value_template"] == "{{ value_json.pos_x }}"
|
|
assert d["unit_of_measurement"] == "mm" # canonical (values are mm)
|
|
assert d["device"]["identifiers"] == [config.uuid]
|
|
assert d["unique_id"] == "%s_pos_x" % config.uuid
|
|
|
|
state = json.loads(ha.state_json(
|
|
{"execution": "ACTIVE", "pos_x": 1.5, "pos_z": "UNAVAILABLE",
|
|
"workoffset": {"G54": {}}}, sensors))
|
|
assert state["execution"] == "ACTIVE" and state["pos_x"] == 1.5
|
|
assert "pos_z" not in state # UNAVAILABLE skipped
|
|
assert "workoffset" not in state # structured value skipped
|
|
print("ok HA discovery (sensors + discovery payload + state JSON)")
|
|
|
|
|
|
def test_entry_dump_probe():
|
|
# The installed entry point (bin/mtconnect-agent, on PATH via rip-environment).
|
|
exe = os.path.join(_EMC2_HOME, "bin", "mtconnect-agent") if _EMC2_HOME \
|
|
else "mtconnect-agent"
|
|
if not (os.path.isfile(exe) or _EMC2_HOME is None):
|
|
print("skip entry --dump-probe (mtconnect-agent not built/installed)")
|
|
return
|
|
out = subprocess.check_output([exe, "--dump-probe", CONFIGS["5axis"]], text=True)
|
|
assert "xyzac-trt-kins" in out and "MTConnectDevices" in out
|
|
print("ok entry --dump-probe")
|
|
|
|
|
|
def main():
|
|
test_kinematics()
|
|
test_probe_registry_consistency()
|
|
test_buffer_and_streams()
|
|
test_work_offset_table()
|
|
test_tool_offset_and_rotation()
|
|
test_assets()
|
|
test_probe_standard_blocks()
|
|
test_spindle_constraints()
|
|
test_schema_validation()
|
|
test_source_offline()
|
|
test_http_endpoints()
|
|
test_shdr()
|
|
test_hal_items()
|
|
test_solid_models()
|
|
test_auto_geometry()
|
|
test_ha_discovery()
|
|
test_entry_dump_probe()
|
|
print("\nALL TESTS PASSED")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|