mariano/soleprint
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

updates 38.1-164

Browse Source
This commit is contained in:
buenosairesam
2026-09-14 10:57:58 -03:00
parent 5ff513fcc5
commit 12ddd01940
15 changed files with 1002 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
soleprint/station/tools/common/cli.py
View File

@@ -63,7 +63,10 @@ Conventions it encodes, so they stop being re-decided:
--name/-n what to call it --force/-f write anyway
--dry-run print, do not do
Progress goes to stdout as plain print(). Errors go to stderr prefixed
Progress goes through say(), NOT print(). say() writes to stdout normally and
to stderr when --output is "-", because that is when stdout carries the
artifact and a stray progress line is a JSON syntax error one pipe stage
later. See common/entity.py. Errors go to stderr prefixed
'Error: ' and exit 1 — never a traceback, which tells a user nothing they
can act on. `fail()` is the only exit path for an *error*; a command that
ends badly but not wrongly (tests failed, drift found) returns an int and
@@ -96,6 +99,27 @@ COMMON = {
}
# Set by Tool.run when the command's --output is "-", i.e. stdout is carrying
# the data. Process-wide because a process runs exactly one command; `say` has
# to be callable from a cmd_* that was handed `args` and never saw the Tool.
_DATA_ON_STDOUT = False
def say(*args, **kwargs):
"""Progress, printed where it cannot corrupt the data channel.
`print()` to stdout is right until the day stdout carries the artifact, and
then it silently produces invalid JSON halfway down a pipe. Tools use this
instead and stop having to think about it.
"""
kwargs.setdefault("file", sys.stderr if _DATA_ON_STDOUT else sys.stdout)
print(*args, **kwargs)
def data_on_stdout():
return _DATA_ON_STDOUT
def fail(message, hint=None):
"""The only way out on error: a line a user can act on, never a traceback."""
print(f"Error: {message}", file=sys.stderr)
@@ -296,6 +320,8 @@ class Tool:
command does, exits 0.
"""
args = self.parser.parse_args(argv)
global _DATA_ON_STDOUT
_DATA_ON_STDOUT = str(getattr(args, "output", "") or "") == "-"
try:
result = args.func(args)
if isinstance(result, int) and result != 0:

157
soleprint/station/tools/common/entity.py Normal file
View File

@@ -0,0 +1,157 @@
"""
One way to hand a thing from one station tool to the next.
Four tools already have this seam and each invented it privately:
histgen plan.json "the expensive half and the mechanical half, split"
modelgen {models, relationships, source} a published contract
shuntgen reads that contract
keymatrix observed/<endpoint>.json
Same idea, four spellings. This is the shared one.
## The opinion baked in: each standalone part makes sense by itself
This is deliberately NOT a pipeline runner, a DAG or a scheduler. The pipe is
`|`. All this provides is what makes `|` safe -- a declared kind, a checked
seam, and progress kept off the data channel.
And every tool keeps working with plain file paths and no pipes at all. That is
the point rather than a concession: histgen runs in a repo that has never heard
of soleprint, modelgen ships as a pip package, rig's whole contract is
`rm -rf rig/`. A tool that only works as part of a pipeline has lost the
property this convention exists to protect.
## The envelope
{
"spr": 1,
"kind": "keymatrix.observed",
"schema": 1,
"produced_by": "keymatrix 0.1.0",
"produced_at": "2026-09-14T08:21:02+00:00",
"data": { ... }
}
`kind` is **declared, never inferred**. A tool that guesses what it was handed
fails three steps down the pipe with something confusing, instead of at the seam
with something obvious.
## Using it
from ..common.entity import read_entity, write_entity # or `from .entity`
doc = read_entity(args.source, expect_kind="keymatrix.observed")
write_entity(args.output, "keymatrix.findings", 1, findings)
`-` means stdin or stdout in the usual way. A bare payload -- a file that is not
an envelope -- is accepted, because `observed/*.json` and `plan.json` already
exist in that form and `plan.json` is a seam a *human* edits between `brief` and
`export`.
"""
import datetime
import json
import sys
from pathlib import Path
from .cli import fail
ENVELOPE_VERSION = 1
# The entities that exist as artefacts today, named after what writes them.
# Declaring a kind here is free; changing the shape behind it is not -- see
# modelgen.ir below.
KINDS = {
"keymatrix.observed": "One endpoint's live keyboard state, as probed.",
"keymatrix.findings": "What `check` concluded from observed state.",
"histgen.plan": "Commits, ordered and cut. Edited by hand between brief and export.",
"modelgen.ir": "{models, relationships, source} -- a PUBLISHED contract. "
"modelgen, datagen, shuntgen and cfg/amar all emit or consume it, "
"and modelgen/tests/test_extractors.py:240,394 assert it. "
"Declared here; NOT to be reshaped from here.",
}
def _now():
return (datetime.datetime.now(datetime.timezone.utc)
.replace(microsecond=0).isoformat())
def is_envelope(obj):
return isinstance(obj, dict) and "spr" in obj and "kind" in obj
def wrap(kind, schema, data, produced_by=None):
"""Build an envelope without writing it, for callers that hold their own IO."""
if kind not in KINDS:
fail(f"unknown entity kind: {kind!r}",
f"Known: {', '.join(sorted(KINDS))}. Add it to entity.KINDS first.")
return {
"spr": ENVELOPE_VERSION,
"kind": kind,
"schema": schema,
"produced_by": produced_by,
"produced_at": _now(),
"data": data,
}
def read_entity(path, expect_kind=None, what="input"):
"""The payload at `path`. `-` reads stdin.
Returns the `data` when handed an envelope, and the whole object when handed
a bare payload -- so a caller does not have to care which it got.
"""
try:
text = sys.stdin.read() if str(path) == "-" else Path(path).read_text(encoding="utf-8")
except FileNotFoundError:
fail(f"{what} not found: {path}")
except OSError as exc:
fail(f"could not read {what} {path}: {exc}")
if not text.strip():
fail(f"{what} is empty: {'stdin' if str(path) == '-' else path}",
"Nothing was piped in." if str(path) == "-" else None)
try:
obj = json.loads(text)
except ValueError as exc:
fail(f"{what} is not JSON: {exc}",
f"Read from {'stdin' if str(path) == '-' else path}.")
if not is_envelope(obj):
# A bare payload. Accepted on purpose: the files that predate this
# convention are still the ones on disk, and plan.json is edited by a
# human between two verbs.
if expect_kind:
return obj
return obj
if obj.get("spr") != ENVELOPE_VERSION:
fail(f"{what} uses envelope version {obj.get('spr')!r}, "
f"and this tool speaks {ENVELOPE_VERSION}")
if expect_kind and obj["kind"] != expect_kind:
# The whole reason `kind` is declared rather than sniffed.
fail(f"expected {expect_kind}, got {obj['kind']}",
f"That came from {obj.get('produced_by') or 'an unknown producer'}.")
return obj["data"]
def write_entity(path, kind, schema, data, produced_by=None, indent=2):
"""Write an envelope. `-` writes stdout.
Returns the path written, or None for stdout, so a caller can report it
without re-deciding where it went.
"""
envelope = wrap(kind, schema, data, produced_by)
text = json.dumps(envelope, indent=indent, sort_keys=True) + "\n"
if str(path) == "-":
sys.stdout.write(text)
return None
target = Path(path)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(text, encoding="utf-8")
return target

23
soleprint/station/tools/histgen/cli.py
View File

@@ -49,6 +49,27 @@ COMMON = {
}
# Set by Tool.run when the command's --output is "-", i.e. stdout is carrying
# the data. Process-wide because a process runs exactly one command; `say` has
# to be callable from a cmd_* that was handed `args` and never saw the Tool.
_DATA_ON_STDOUT = False
def say(*args, **kwargs):
"""Progress, printed where it cannot corrupt the data channel.
`print()` to stdout is right until the day stdout carries the artifact, and
then it silently produces invalid JSON halfway down a pipe. Tools use this
instead and stop having to think about it.
"""
kwargs.setdefault("file", sys.stderr if _DATA_ON_STDOUT else sys.stdout)
print(*args, **kwargs)
def data_on_stdout():
return _DATA_ON_STDOUT
def fail(message, hint=None):
"""The only way out on error: a line a user can act on, never a traceback."""
print(f"Error: {message}", file=sys.stderr)
@@ -249,6 +270,8 @@ class Tool:
command does, exits 0.
"""
args = self.parser.parse_args(argv)
global _DATA_ON_STDOUT
_DATA_ON_STDOUT = str(getattr(args, "output", "") or "") == "-"
try:
result = args.func(args)
if isinstance(result, int) and result != 0:

140
soleprint/station/tools/keymatrix/__main__.py
View File

@@ -12,32 +12,41 @@ folder out and use its Makefile.
import json
from pathlib import Path
from .cli import Tool, fail
from . import __version__
from .cli import Tool, fail, say
from .entity import read_entity, write_entity
def _load_observed(args):
"""Every observed/<endpoint>.json under --source, keyed by endpoint."""
root = Path(args.source).expanduser() if args.source else Path("observed")
if root.name != "observed":
root = root / "observed"
if not root.is_dir():
fail(f"no observed state at {root}",
"Run `keymatrix probe --output <dir>` first.")
"""The same documents, indexed chord -> binding, which is what shadow wants."""
out = {}
for path in sorted(root.glob("*.json")):
doc = json.loads(path.read_text())
for endpoint, doc in _load_docs(args).items():
index = {}
for b in doc.get("bindings", []):
index.setdefault(b["chord"], b) # first declaration wins
out[doc["endpoint"]] = index
if not out:
fail(f"{root} holds no observed state")
out[endpoint] = index
return out
def _load_docs(args):
"""The full observed documents, which the checks need (not the chord index)."""
root = Path(args.source).expanduser() if args.source else Path("observed")
"""Observed documents by endpoint. Accepts a pipe, a file, or a directory.
All three exist for a reason: `-` is the pipe, a file is one endpoint, and a
directory is how the peripheral-matrix repo stores the fleet. A tool that
only took the pipe would stop being usable on its own.
"""
source = args.source
if str(source) == "-":
doc = read_entity("-", expect_kind="keymatrix.observed",
what="observed state")
return {doc["endpoint"]: doc}
root = Path(source).expanduser() if source else Path("observed")
if root.is_file():
doc = read_entity(root, expect_kind="keymatrix.observed",
what="observed state")
return {doc["endpoint"]: doc}
if root.name != "observed":
root = root / "observed"
if not root.is_dir():
@@ -45,7 +54,8 @@ def _load_docs(args):
"Run `keymatrix probe --output <dir>` first.")
docs = {}
for path in sorted(root.glob("*.json")):
doc = json.loads(path.read_text())
doc = read_entity(path, expect_kind="keymatrix.observed",
what=f"observed state ({path.name})")
docs[doc["endpoint"]] = doc
if not docs:
fail(f"{root} holds no observed state")
@@ -57,22 +67,31 @@ def cmd_probe(args):
from .endpoint import resolve_endpoint
name = args.name or resolve_endpoint()
doc, warnings = probe.run(name)
# --ahk-root makes the Windows endpoint probeable from anywhere the tree is
# readable, including an sshfs mount -- and equally probeable from the
# fixture when no mount exists. Neither is privileged.
kind = "ahk" if args.ahk_root else None
doc, warnings = probe.run(name, kind=kind, ahk_root=args.ahk_root)
for severity, message in warnings:
print(f" {severity}: {message}")
say(f" {severity}: {message}")
if args.dry_run:
print(json.dumps(doc, indent=2))
say(json.dumps(doc, indent=2))
return
root = Path(args.output).expanduser() if args.output else Path(".")
out = root / "observed"
out.mkdir(parents=True, exist_ok=True)
path = out / f"{name}.json"
path.write_text(json.dumps(doc, indent=2, sort_keys=True) + "\n")
print(f"Wrote {path} ({len(doc['bindings'])} bindings, "
f"{len(doc['unreadable'])} unreadable)")
# "-" pipes one endpoint's state onward; a directory keeps the observed/
# layout the peripheral-matrix repo holds.
if str(args.output) == "-":
target = "-"
else:
root = Path(args.output).expanduser() if args.output else Path(".")
target = root / "observed" / f"{name}.json"
written = write_entity(target, "keymatrix.observed", 1, doc,
produced_by=f"keymatrix {__version__}")
say(f"Wrote {written or 'stdout'} ({len(doc['bindings'])} bindings, "
f"{len(doc['unreadable'])} unreadable)")
def cmd_shadow(args):
@@ -84,26 +103,51 @@ def cmd_shadow(args):
for name in postures:
p = sh.POSTURES[name]
print(f"\nPosture {name} — {p['summary']}")
print(f" peripherals on {p['peripherals']}; "
say(f"\nPosture {name} — {p['summary']}")
say(f" peripherals on {p['peripherals']}; "
f"chain: {' -> '.join(h['near'] + '/' + h['transport'] for h in p['hops'])}")
print()
print(f" {'chord':<22}{'verdict':<20}{'eaten by'}")
print(" " + "-" * 72)
say()
states = sh.grab_states(sh.TRANSPORTS[p["hops"][0]["transport"]])
head = "".join(f"{s or 'verdict':<20}" for s in states)
say(f" {'chord':<22}{head}{'eaten by'}")
say(" " + "-" * (24 + 20 * len(states) + 28))
for raw in args.chord or sorted({c for e in observed.values() for c in e}):
rows = sh.walk(Chord.parse(raw), name, observed)
hop, v = rows[0]
print(f" {raw:<22}{v.outcome:<20}{v.eater or '—'}")
cells, eater = [], None
for st in states:
_, v = sh.walk(Chord.parse(raw), name, observed, state=st)[0]
cells.append(f"{v.outcome:<20}")
eater = eater or v.eater
say(f" {raw:<22}{''.join(cells)}{eater or '—'}")
def cmd_freespace(args):
from .freespace import survey, format_rows
observed = _load_observed(args)
say(f"Chord families across: {', '.join(sorted(observed))}")
say("")
for line in format_rows(survey(observed), verbose=args.verbose):
say(line)
def cmd_check(args):
from .check import run_checks
findings = run_checks(_load_docs(args))
for f in findings:
print(f" {f['severity']:<6} {f['kind']:<22} {f['message']}")
errors = sum(f["severity"] == "error" for f in findings)
print(f"\n{len(findings)} finding(s), {errors} at error severity")
if args.output:
write_entity(args.output, "keymatrix.findings", 1,
{"findings": findings,
"counts": {"total": len(findings), "errors": errors}},
produced_by=f"keymatrix {__version__}")
for f in findings:
say(f" {f['severity']:<6} {f['kind']:<22} {f['message']}")
say(f"\n{len(findings)} finding(s), {errors} at error severity")
# A finding is a true report, not a malfunction -- so this is an int return
# rather than fail(), and it does not get an "Error:" in front of it.
return 1 if errors else 0
@@ -116,18 +160,34 @@ def main(argv=None):
help="Directory holding observed/ (default: here)"),
name=dict(default=None,
help="Endpoint name (default: resolved from the host)"))
tool.argument("probe", "--ahk-root", default=None, metavar="DIR",
help="Probe an AutoHotkey tree at DIR instead of this host. "
"Use the fixture at keymatrix/examples/ahk when no "
"Windows tree is reachable.")
tool.command("shadow", cmd_shadow, "Which endpoint receives a chord, per posture")
tool.common("shadow", "source",
source=dict(default=None, help="Directory holding observed/"))
source=dict(default=None,
help='Directory, file, or "-" for a pipe'))
tool.argument("shadow", "--posture", choices=("A", "B"), default=None,
help="Only this posture (default: both)")
tool.argument("shadow", "--chord", action="append", default=None,
help="Only this chord. Repeatable.")
tool.command("freespace", cmd_freespace,
"Chord families nobody has claimed")
tool.common("freespace", "source",
source=dict(default=None,
help='Directory, file, or "-" for a pipe'))
tool.argument("freespace", "--verbose", "-v", action="store_true",
help="Show what took each partly-claimed family")
tool.command("check", cmd_check, "Drift, dead commands, collisions")
tool.common("check", "source",
source=dict(default=None, help="Directory holding observed/"))
tool.common("check", "source", "output",
source=dict(default=None,
help='Directory, file, or "-" for a pipe'),
output=dict(default=None,
help='Also write findings here; "-" for a pipe'))
tool.run(argv)

23
soleprint/station/tools/keymatrix/cli.py
View File

@@ -23,6 +23,27 @@ COMMON = {
}
# Set by Tool.run when the command's --output is "-", i.e. stdout is carrying
# the data. Process-wide because a process runs exactly one command; `say` has
# to be callable from a cmd_* that was handed `args` and never saw the Tool.
_DATA_ON_STDOUT = False
def say(*args, **kwargs):
"""Progress, printed where it cannot corrupt the data channel.
`print()` to stdout is right until the day stdout carries the artifact, and
then it silently produces invalid JSON halfway down a pipe. Tools use this
instead and stop having to think about it.
"""
kwargs.setdefault("file", sys.stderr if _DATA_ON_STDOUT else sys.stdout)
print(*args, **kwargs)
def data_on_stdout():
return _DATA_ON_STDOUT
def fail(message, hint=None):
"""The only way out on error: a line a user can act on, never a traceback."""
print(f"Error: {message}", file=sys.stderr)
@@ -223,6 +244,8 @@ class Tool:
command does, exits 0.
"""
args = self.parser.parse_args(argv)
global _DATA_ON_STDOUT
_DATA_ON_STDOUT = str(getattr(args, "output", "") or "") == "-"
try:
result = args.func(args)
if isinstance(result, int) and result != 0:

157
soleprint/station/tools/keymatrix/entity.py Normal file
View File

@@ -0,0 +1,157 @@
"""
One way to hand a thing from one station tool to the next.
Four tools already have this seam and each invented it privately:
histgen plan.json "the expensive half and the mechanical half, split"
modelgen {models, relationships, source} a published contract
shuntgen reads that contract
keymatrix observed/<endpoint>.json
Same idea, four spellings. This is the shared one.
## The opinion baked in: each standalone part makes sense by itself
This is deliberately NOT a pipeline runner, a DAG or a scheduler. The pipe is
`|`. All this provides is what makes `|` safe -- a declared kind, a checked
seam, and progress kept off the data channel.
And every tool keeps working with plain file paths and no pipes at all. That is
the point rather than a concession: histgen runs in a repo that has never heard
of soleprint, modelgen ships as a pip package, rig's whole contract is
`rm -rf rig/`. A tool that only works as part of a pipeline has lost the
property this convention exists to protect.
## The envelope
{
"spr": 1,
"kind": "keymatrix.observed",
"schema": 1,
"produced_by": "keymatrix 0.1.0",
"produced_at": "2026-09-14T08:21:02+00:00",
"data": { ... }
}
`kind` is **declared, never inferred**. A tool that guesses what it was handed
fails three steps down the pipe with something confusing, instead of at the seam
with something obvious.
## Using it
from ..common.entity import read_entity, write_entity # or `from .entity`
doc = read_entity(args.source, expect_kind="keymatrix.observed")
write_entity(args.output, "keymatrix.findings", 1, findings)
`-` means stdin or stdout in the usual way. A bare payload -- a file that is not
an envelope -- is accepted, because `observed/*.json` and `plan.json` already
exist in that form and `plan.json` is a seam a *human* edits between `brief` and
`export`.
"""
import datetime
import json
import sys
from pathlib import Path
from .cli import fail
ENVELOPE_VERSION = 1
# The entities that exist as artefacts today, named after what writes them.
# Declaring a kind here is free; changing the shape behind it is not -- see
# modelgen.ir below.
KINDS = {
"keymatrix.observed": "One endpoint's live keyboard state, as probed.",
"keymatrix.findings": "What `check` concluded from observed state.",
"histgen.plan": "Commits, ordered and cut. Edited by hand between brief and export.",
"modelgen.ir": "{models, relationships, source} -- a PUBLISHED contract. "
"modelgen, datagen, shuntgen and cfg/amar all emit or consume it, "
"and modelgen/tests/test_extractors.py:240,394 assert it. "
"Declared here; NOT to be reshaped from here.",
}
def _now():
return (datetime.datetime.now(datetime.timezone.utc)
.replace(microsecond=0).isoformat())
def is_envelope(obj):
return isinstance(obj, dict) and "spr" in obj and "kind" in obj
def wrap(kind, schema, data, produced_by=None):
"""Build an envelope without writing it, for callers that hold their own IO."""
if kind not in KINDS:
fail(f"unknown entity kind: {kind!r}",
f"Known: {', '.join(sorted(KINDS))}. Add it to entity.KINDS first.")
return {
"spr": ENVELOPE_VERSION,
"kind": kind,
"schema": schema,
"produced_by": produced_by,
"produced_at": _now(),
"data": data,
}
def read_entity(path, expect_kind=None, what="input"):
"""The payload at `path`. `-` reads stdin.
Returns the `data` when handed an envelope, and the whole object when handed
a bare payload -- so a caller does not have to care which it got.
"""
try:
text = sys.stdin.read() if str(path) == "-" else Path(path).read_text(encoding="utf-8")
except FileNotFoundError:
fail(f"{what} not found: {path}")
except OSError as exc:
fail(f"could not read {what} {path}: {exc}")
if not text.strip():
fail(f"{what} is empty: {'stdin' if str(path) == '-' else path}",
"Nothing was piped in." if str(path) == "-" else None)
try:
obj = json.loads(text)
except ValueError as exc:
fail(f"{what} is not JSON: {exc}",
f"Read from {'stdin' if str(path) == '-' else path}.")
if not is_envelope(obj):
# A bare payload. Accepted on purpose: the files that predate this
# convention are still the ones on disk, and plan.json is edited by a
# human between two verbs.
if expect_kind:
return obj
return obj
if obj.get("spr") != ENVELOPE_VERSION:
fail(f"{what} uses envelope version {obj.get('spr')!r}, "
f"and this tool speaks {ENVELOPE_VERSION}")
if expect_kind and obj["kind"] != expect_kind:
# The whole reason `kind` is declared rather than sniffed.
fail(f"expected {expect_kind}, got {obj['kind']}",
f"That came from {obj.get('produced_by') or 'an unknown producer'}.")
return obj["data"]
def write_entity(path, kind, schema, data, produced_by=None, indent=2):
"""Write an envelope. `-` writes stdout.
Returns the path written, or None for stdout, so a caller can report it
without re-deciding where it went.
"""
envelope = wrap(kind, schema, data, produced_by)
text = json.dumps(envelope, indent=indent, sort_keys=True) + "\n"
if str(path) == "-":
sys.stdout.write(text)
return None
target = Path(path)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(text, encoding="utf-8")
return target

30
soleprint/station/tools/keymatrix/examples/ahk/README.md Normal file
View File

@@ -0,0 +1,30 @@
# examples/ahk — a fixture, not a copy
A miniature AutoHotkey v2 tree carrying every hotkey *shape* `probe/ahk.py` has to
read. It exists so the Windows probe can be built and tested with **no Windows, no
mount and nothing installed** — the same discipline `histgen/selftest.py` uses when
it builds its own repo rather than pointing at one.
## Why a fixture and not the real thing
The real tree lives on a DLP-monitored work machine reached over an sshfs mount that
**may not be there**. A probe developed against it would be untestable the moment it
was unmounted, and copying it in would carry a username, a deploy path and an NTFS
alternate data stream (`shortcuts.txt:sec.endpointdlp`) into this repo.
So these files are written from the *shapes*, never copied. No hostnames, no
usernames, no paths, no client identifiers. `distill.json`'s removal from history is
the standing precedent.
## The shapes, and why each is here
| shape | why it matters |
| --- | --- |
| `^!1::` `^+1::` | static hotkeys, the easy case |
| `^#c::` `^+#c::` | four-modifier chords, and `#` = Super |
| `#3::` | Super+digit, which a GNOME client claims upstream |
| `^!i::` | Ctrl+Alt+letter — the family that is free on both sides |
| `Hotkey chord, action` in a loop | **the load-bearing case.** Every real alias is bound this way, so a static `::`-scanner reports all of them as unbound |
| `.env`-supplied chord lists | the loop's chords are config, not code, so the probe has to resolve them |
| `; ^!q::` commented out | must NOT be reported as bound |
| `^!` (truncated) | must be REPORTED as unparseable, never silently dropped |

29
soleprint/station/tools/keymatrix/examples/ahk/lib/config.ahk Normal file
View File

@@ -0,0 +1,29 @@
; lib/config.ahk — .env parsing, the grammar the fixture's .env follows.
; Present so the fixture tree is shaped like the real one; the probe
; reimplements these rules in Python rather than executing this.
EnvParse(path) {
global _env := Map()
Loop Read, path {
line := Trim(A_LoopReadLine)
if (line = "" || SubStr(line, 1, 1) = "#")
continue
pos := InStr(line, "=")
if (!pos)
continue
_env[Trim(SubStr(line, 1, pos - 1))] := Trim(SubStr(line, pos + 1))
}
}
EnvVal(key, fallback := "") {
global _env
return _env.Has(key) ? _env[key] : fallback
}
; Each key also becomes a global of the same name, which is how the modules
; refer to them — VNC_CLIP_PUSH, not EnvVal("VNC_CLIP_PUSH").
EnvExport() {
global
for k, v in _env
%k% := v
}

9
soleprint/station/tools/keymatrix/examples/ahk/main.ahk Normal file
View File

@@ -0,0 +1,9 @@
; main.ahk — include order is load-bearing: config first, then modules.
#SingleInstance Force
SendMode "Input"
#Include %A_ScriptDir%\lib\config.ahk
EnvParse(A_ScriptDir . "\.env")
#Include %A_ScriptDir%\modules\static.ahk
#Include %A_ScriptDir%\modules\dynamic.ahk

52
soleprint/station/tools/keymatrix/examples/ahk/modules/dynamic.ahk Normal file
View File

@@ -0,0 +1,52 @@
; modules/dynamic.ahk — chords that are CONFIG, not code.
;
; This is the shape that breaks a naive scanner. Nothing here contains a `::`,
; so a static scan reports every one of these as unbound — while they are in
; fact the aliases that carry the fleet when the primary chord cannot cross the
; wire.
;
; lib/config.ahk turns each .env key into a global of the same name, so the
; chord is never a literal anywhere in the source. Note there is a WRAPPER in
; between: nothing here calls Hotkey directly except Bind() itself. A matcher
; keyed on the wrapper's name would stop working the day it is renamed; the
; durable signal is "an identifier that is also an .env key, in a file that
; calls Hotkey at all".
Bind(chord, action, label) {
if (Trim(chord) = "")
return
try {
Hotkey chord, action
Log("bound " chord " -> " label)
} catch as e
Log("could NOT bind '" chord "' (" label "): " e.Message)
}
; Bare globals, one per alias.
Bind(VNC_CLIP_PUSH, (*) => ClipPush(), "clip push")
Bind(VNC_CLIP_TYPE, (*) => ClipType(), "clip type-in")
Bind(VNC_EDITOR, (*) => Apps.Activate("editor"), "editor")
Bind(VNC_BROWSER, (*) => Apps.Activate("browser"), "browser")
; A family: a modifier prefix concatenated with a digit at bind time, so the
; chord never appears whole in the source.
if (Trim(VNC_SLOT_MODS) != "") {
Loop 9 {
n := A_Index
Bind(VNC_SLOT_MODS . n, AppSlot.Bind(n), "app slot " n)
}
}
; A list global, iterated. Bind several and use whichever gets through the
; client, without editing code.
for _tvKey in TASKVIEW_HOTKEYS {
try {
Hotkey _tvKey, (*) => TaskView()
} catch as e
Log("taskview: could NOT bind '" _tvKey "'")
}
; Deliberately malformed: modifiers with no key. The probe must REPORT this,
; not skip it. An unparsed hotkey is an invisible binding, which is the exact
; failure this tool exists to end.
Hotkey "^!", (*) => Broken()

27
soleprint/station/tools/keymatrix/examples/ahk/modules/static.ahk Normal file
View File

@@ -0,0 +1,27 @@
; modules/static.ahk — every static hotkey form the probe must read.
; ^ = Ctrl, ! = Alt, + = Shift, # = Super/Win.
^!1::SwitchDesktop(1)
^!2::SwitchDesktop(2)
^+1::MoveWindowToDesktop(1)
; Super+digit. A GNOME client binds <Super>1..9 for switch-to-application, so
; this never arrives when driven from one — which is the whole point of shadow.
#3::Apps.Activate("editor")
#4::Apps.Activate("browser")
; Four modifiers at once.
^#c::ClipPush()
^+#c::ClipType()
; Ctrl+Alt+letter: unclaimed upstream, which is why the aliases live here.
^!i::ToggleIndicator()
; Commented out. Must NOT be reported as bound.
; ^!q::NeverBound()
; A block-bodied hotkey, which AHK v2 allows.
^!m:: {
DoSomething()
}

80
soleprint/station/tools/keymatrix/freespace.py Normal file
View File

@@ -0,0 +1,80 @@
"""
Which chord families nobody has claimed — recomputed, so a hand result stays true.
`modules/vnckeys.ahk`'s header contains a survey done by hand:
Surveying every binding nrft claims (all five schemas plus its custom
keybindings) leaves one roomy gap:
Ctrl+Alt+<letter> -- entirely unclaimed. Ctrl+Alt+<digit> is NOT.
That survey is **correct** — checked family by family against a live probe. This
module exists because it will not stay correct on its own. It is a regression
net for a right answer, not a correction of a wrong one: the day someone binds
`Ctrl+Alt+e` in GNOME, four aliases silently stop arriving and nothing says so.
A family is free on an endpoint when the endpoint binds no member of it. Free
*across* endpoints is the interesting question, since an alias has to survive
the near side and still mean something on the far side.
"""
import string
from .chord import Chord
DIGITS = tuple(str(n) for n in range(1, 10))
LETTERS = tuple(string.ascii_lowercase)
FKEYS = tuple(f"F{n}" for n in range(1, 13))
# The modifier sets worth asking about. Bare and single-modifier families are
# left out deliberately: they collide with ordinary typing and application
# shortcuts, so "free" there would be misleading.
FAMILIES = [
("Ctrl+Alt", ("Ctrl", "Alt")),
("Ctrl+Shift", ("Ctrl", "Shift")),
("Ctrl+Super", ("Ctrl", "Super")),
("Alt+Super", ("Alt", "Super")),
("Super", ("Super",)),
("Ctrl+Alt+Shift", ("Ctrl", "Alt", "Shift")),
("Ctrl+Shift+Super", ("Ctrl", "Shift", "Super")),
]
RANGES = [("digit", DIGITS), ("letter", LETTERS), ("F-key", FKEYS)]
def survey(observed_by_endpoint):
"""One row per (family, range, endpoint set): what is free and what took it."""
rows = []
for fam_name, mods in FAMILIES:
for range_name, keys in RANGES:
per_endpoint = {}
for endpoint, index in sorted(observed_by_endpoint.items()):
taken = {}
for key in keys:
hit = index.get(str(Chord(mods, key)))
if hit:
taken[key] = hit.get("label") or hit["source"]
per_endpoint[endpoint] = taken
free_everywhere = [k for k in keys
if not any(k in t for t in per_endpoint.values())]
rows.append({
"family": f"{fam_name}+<{range_name}>",
"total": len(keys),
"free_everywhere": free_everywhere,
"taken": per_endpoint,
})
return rows
def format_rows(rows, verbose=False):
out = []
for r in rows:
n_free, total = len(r["free_everywhere"]), r["total"]
mark = "FREE" if n_free == total else ("none" if n_free == 0 else "part")
out.append(f" {r['family']:<26} {n_free:>2}/{total:<3} {mark}")
if verbose or 0 < n_free < total:
for endpoint, taken in r["taken"].items():
if taken:
shown = ", ".join(f"{k} ({v})" for k, v in sorted(taken.items())[:4])
more = "" if len(taken) <= 4 else f", +{len(taken) - 4} more"
out.append(f" {endpoint}: {shown}{more}")
return out

7
soleprint/station/tools/keymatrix/probe/__init__.py
View File

@@ -13,7 +13,7 @@ import datetime
from ..endpoint import platform_kind
def run(endpoint, kind=None):
def run(endpoint, kind=None, ahk_root=None):
"""(observed document, warnings) for this machine."""
kind = kind or platform_kind()
warnings = []
@@ -46,6 +46,11 @@ def run(endpoint, kind=None):
doc["layers"]["keyd"] = keyd.collect(report, doc["unreadable"])
doc["input"] = xkb.collect(report)
doc["transports"]["deskflow"] = deskflow.collect(report)
elif kind == "ahk":
from . import ahk
doc["platform"] = {"kind": "windows-ahk"}
doc["bindings"], meta = ahk.collect(ahk_root, report, doc["unreadable"])
doc["platform"]["ahk"] = meta
else:
report("warn", f"no probe backend for {kind!r}; nothing was collected")
doc["unreadable"].append(

249
soleprint/station/tools/keymatrix/probe/ahk.py Normal file
View File

@@ -0,0 +1,249 @@
"""
Read an AutoHotkey tree's declared chords, from wherever it happens to be.
Takes a path. There is no default location and no mount assumed: the real tree
lives on a DLP-monitored machine behind an sshfs mount that **may not be there**,
and a probe that required it would be untestable the moment it was unmounted.
An absent root is recorded as unreadable and the run continues -- the same
discipline `probe/keyd.py` uses for the root-only config it cannot read.
`examples/ahk/` in this package is a fixture carrying every shape below, so this
module is developed and tested with no Windows, no mount and nothing installed.
## Static is the easy half
^!1::SwitchDesktop(1) Ctrl+Alt+1
^+#c::ClipType() Ctrl+Shift+Super+c
## Dynamic is the half that matters
`lib/config.ahk` turns every `.env` key into a global, and the real code binds
through those globals -- never through a literal:
VncBind(VNC_CLIP_PUSH, (*) => ClipPush(), "clip push") ; a bare global
Loop 9 {
VncBind(VNC_SLOT_MODS . n, ...) ; a family
}
for _tvKey in TASKVIEW_HOTKEYS { Hotkey _tvKey, ... } ; a list global
None of that contains `::`, so a static scan reports every one as unbound -- and
those are precisely the aliases that carry the fleet when the primary chord
cannot cross the wire.
So the signal is not a wrapper's name, which varies (`VncBind`, `Hotkey`,
whatever comes next). It is: **an identifier that matches a key in `.env`,
appearing in a file that calls `Hotkey` at all.** That holds regardless of how
many helper layers sit in between, which is the property worth having -- the
alternative is a matcher that silently stops working when someone renames a
function.
## What is not read
Anything whose path contains `:`. NTFS alternate data streams surface over sshfs
and reading one touches DLP classification metadata.
"""
import re
from pathlib import Path
from ..chord import Chord
# `KEY=value`, `#` comments on their own line, no quotes, trim both sides, last
# wins. Reimplemented from lib/config.ahk:EnvParse() rather than executed.
_ENV_LINE = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*=(.*)$")
# A static hotkey: chord, then `::`, at the start of a line.
_STATIC = re.compile(r"^\s*([\^!+#<>*~$]*[^\s:;][^\s:]*?)::")
# `Hotkey <something>, <action>` -- the dynamic form.
_DYNAMIC = re.compile(r"""^\s*Hotkey\s+(?P<what>"[^"]*"|EnvVal\(\s*"[^"]*"\s*\)|[A-Za-z_]\w*)""")
# A bare modifier run: a family, not a chord.
_BARE_MODS = re.compile(r"^[\^!+#]+$")
def parse_env(text):
out = {}
for line in text.split("\n"):
if not line.strip() or line.lstrip().startswith("#"):
continue
m = _ENV_LINE.match(line)
if m:
out[m.group(1).strip()] = m.group(2).strip()
return out
def _iter_files(root):
"""Every .ahk under root, skipping alternate data streams."""
for path in sorted(root.rglob("*.ahk")):
if ":" in path.name:
continue
yield path
def _record(bindings, chord_text, source, kind, report, detail=None):
try:
chord = Chord.parse_ahk(chord_text)
except ValueError as exc:
report("warn", f"unparsed AHK hotkey {chord_text!r} in {source}: {exc}")
return False
if not chord.key:
# `^!` with no key. An unparsed hotkey is an invisible binding, which is
# the exact failure this tool exists to end, so it is reported rather
# than dropped.
report("warn", f"AHK hotkey {chord_text!r} in {source} has modifiers but "
f"no key; it binds nothing")
return False
bindings.append({
"chord": str(chord),
"raw": chord_text,
"source": f"ahk:{source}",
"label": detail,
"command": None,
"binding_kind": kind,
})
return True
def collect(root, report, unreadable, env_overrides=None):
"""(bindings, meta) for the AHK tree at `root`, which may not exist."""
bindings, meta = [], {"root": str(root) if root else None,
"env": {}, "files": [], "dynamic_unresolved": []}
if not root:
unreadable.append({"what": "the AHK tree",
"reason": "no root given; pass --ahk-root"})
return bindings, meta
# Absolute, once, here. %A_ScriptDir% expands to this string, and a
# relative root made the expansion look like a relative include -- which
# then got root prepended a second time. It worked against the real tree
# only because that path happened to be absolute.
root = Path(root).expanduser().resolve()
if not root.is_dir():
# The mount is gone, or this endpoint simply has no AHK. Both are facts,
# not failures.
report("warn", f"no AHK tree at {root}; nothing was collected from it")
unreadable.append({"what": f"the AHK tree at {root}",
"reason": "path does not exist (the mount may be absent)"})
return bindings, meta
env_path = root / ".env"
env = {}
if env_path.exists():
env = parse_env(env_path.read_text(encoding="utf-8", errors="replace"))
else:
report("warn", f"no .env beside {root}; dynamic hotkeys cannot be resolved")
env.update(env_overrides or {})
meta["env"] = env
live = _reachable(root, report)
meta["entry_reachable"] = sorted(p.relative_to(root).as_posix() for p in live)
for path in _iter_files(root):
rel = path.relative_to(root).as_posix()
meta["files"].append(rel)
if live and path not in live:
# ref/, debug/ and 3rd/ hold prototypes and scratch. They declare
# chords that nothing binds, and counting them would report the
# near side claiming keys it does not.
meta.setdefault("skipped_unreachable", []).append(rel)
continue
try:
text = path.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
report("warn", f"could not read {rel}: {exc}")
unreadable.append({"what": rel, "reason": str(exc)})
continue
for raw_line in text.split("\n"):
line = raw_line.rstrip()
if not line or line.lstrip().startswith(";"):
continue # a commented-out hotkey is not bound
m = _STATIC.match(line)
if m:
_record(bindings, m.group(1), rel, "static", report)
continue
# `Hotkey "^!c", ...` -- a literal, which is rare but is also the
# only place a malformed chord shows up as itself. Keeping this is
# what exercises the report-what-you-cannot-read path.
lit = re.match(r'^\s*Hotkey\s+"([^"]*)"', line)
if lit:
_record(bindings, lit.group(1), rel, "dynamic-literal", report)
if "Hotkey" not in text:
continue
# Every .env key named as an identifier in a file that binds hotkeys.
for key, value in sorted(env.items()):
if not value:
continue
for use in re.finditer(rf"\b{re.escape(key)}\b(\s*\.\s*\w+)?", text):
concatenated = bool(use.group(1))
_from_env(bindings, env, key, rel, report,
expand_digits=concatenated)
break # one record per key per file, not per mention
return bindings, meta
def _reachable(root, report):
"""Files reachable from main.ahk by #Include, transitively.
Include order is load-bearing in these trees and the entry point is the only
authority on what is live. Without this, `ref/` prototypes get counted as
bindings -- on the real tree that was 103 static hotkeys, most of them from
three abandoned OSD experiments.
Returns an empty set if there is no entry point, in which case everything is
scanned rather than nothing.
"""
entry = root / "main.ahk"
if not entry.exists():
report("note", f"no main.ahk in {root}; scanning every .ahk instead of "
f"only what the entry point includes")
return set()
seen, queue = set(), [entry.resolve()]
while queue:
path = queue.pop()
if path in seen or not path.exists():
continue
seen.add(path.resolve())
try:
text = path.read_text(encoding="utf-8", errors="replace")
except OSError:
continue
for inc in re.findall(r"^\s*#Include\s+(.+)$", text, re.M):
inc = inc.strip().strip('"')
inc = inc.replace("%A_ScriptDir%", str(root)).replace("\\", "/")
candidate = Path(inc)
if not candidate.is_absolute():
candidate = root / inc
queue.append(candidate.resolve())
return seen
return bindings, meta
def _from_env(bindings, env, key, source, report, expand_digits=False):
value = env.get(key, "")
if not value:
report("note", f"{source}: {key} is unset in .env, so nothing is bound for it")
return
if "," in value and not _BARE_MODS.match(value):
# A list global, iterated with `for x in KEY`.
for part in (v.strip() for v in value.split(",")):
if part:
_record(bindings, part, source, "dynamic", report, detail=key)
return
if _BARE_MODS.match(value) or expand_digits:
if not _BARE_MODS.match(value):
report("warn", f"{source}: {key}={value!r} is used as a family prefix "
f"but is not a bare modifier run")
return
for n in range(1, 10):
_record(bindings, f"{value}{n}", source, "dynamic-family", report,
detail=f"{key} slot {n}")
return
_record(bindings, value, source, "dynamic", report, detail=key)

41
soleprint/station/tools/keymatrix/shadow.py
View File

@@ -59,13 +59,32 @@ class Verdict:
return f"<{self.outcome}{' by ' + self.eater if self.eater else ''}>"
def reaches(chord, transport, near_observed):
def grab_states(transport):
"""The client states this transport behaves differently in.
One entry for every transport but RDP, which has a documented knob. Keeping
this a lookup rather than an expression language is deliberate: one setting,
in one transport, does not justify a grammar.
"""
grab = transport.get("grab", "unknown")
return tuple(grab) if isinstance(grab, dict) else (None,)
def reaches(chord, transport, near_observed, state=None):
"""Does `chord`, pressed on the near side, arrive at the far side?
`near_observed` maps a normal-form chord string to the binding that claims
it -- the probe's output for whichever endpoint is near in this posture.
`state` picks a column when the transport's grab is conditional.
"""
grab = transport.get("grab", "unknown")
if isinstance(grab, dict):
state = state or transport.get("default_state")
if state not in grab:
raise ValueError(
f"{transport['name']} needs a client state, one of "
f"{sorted(grab)}; got {state!r}")
grab = grab[state]
if grab not in GRABS:
raise ValueError(f"unknown grab kind {grab!r}")
@@ -110,15 +129,21 @@ TRANSPORTS = {
},
"rdp-grd": {
"name": "rdp-grd",
"grab": "unknown",
# Conditional, and this is the only conditional grab in the fleet. RDP
# has a knob RealVNC simply does not: keyboardhook decides whether
# Windows key combinations are applied locally or sent to the remote.
# So posture B is *configurable* where posture A is not.
"grab": {"fullscreen": "effective", "windowed": "none"},
"default_state": "fullscreen",
"intercepts": [],
"evidence": (
"Posture B: ethbox's RDP client into nrft's gnome-remote-desktop "
"(system unit active on *:3389, so a HEADLESS session -- a separate "
"GNOME instance, same dconf). mstsc has a documented 'Apply Windows "
"key combinations' setting with no RealVNC equivalent, so this "
"direction may be configurable where posture A is not. UNTESTED: the "
"setting's value here is unknown, so the grab is unknown."),
"GNOME instance, same dconf). nrft.rdp carries `keyboardhook:i:2` = "
"'apply Windows key combinations only when using the full screen', "
"and `redirectclipboard:i:1`. READ FROM CONFIG, NOT OBSERVED: this "
"is what the profile says the client will do, not a keypress anyone "
"has watched arrive."),
},
"physical-switch": {
"name": "physical-switch",
@@ -171,7 +196,7 @@ POSTURES = {
}
def walk(chord, posture, observed_by_endpoint):
def walk(chord, posture, observed_by_endpoint, state=None):
"""Follow a chord along a posture's chain until something eats it.
Returns one row per hop, so a three-hop posture shows where the chord died
@@ -183,7 +208,7 @@ def walk(chord, posture, observed_by_endpoint):
if near is None:
rows.append((hop, Verdict("UNKNOWN", f"{hop['near']} has never been probed")))
break
v = reaches(chord, TRANSPORTS[hop["transport"]], near)
v = reaches(chord, TRANSPORTS[hop["transport"]], near, state=state)
rows.append((hop, v))
if not v.reaches_far_side:
break