Files
OpenUniFi/scripts/docs/generate-wiki.py
T
KodaandKoda 3fcd99af1f
Build and publish release / metadata (push) Successful in 44s
Build and publish release / create-release (push) Successful in 49s
Build and publish release / publish-release (push) Canceled after 0s
Build and publish release / build (push) Canceled after 7m44s
refactor #1 + adding documentation (#36)
Reviewed-on: #36
Co-authored-by: Koda YeenBean <n122330@gmail.com>
2026-07-19 21:26:28 +01:00

582 lines
21 KiB
Python
Executable File

#!/usr/bin/env python3
"""Generate and lint the openUF Gitea Wiki architecture documentation."""
from __future__ import annotations
import argparse
import dataclasses
import re
import sys
from pathlib import Path
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
SOURCE_ROOT = REPOSITORY_ROOT / "src"
DEFAULT_WIKI_ROOT = REPOSITORY_ROOT / "wiki"
MAX_IMPLEMENTATION_LINES = 900
CONTROL_WORDS = {
"for", "if", "return", "sizeof", "switch", "while", "do",
"case", "defined", "typeof", "alignof",
}
NON_ENGLISH_COMMENT_WORDS = {
"algunos", "configuración", "configuramos", "controlador", "cuando",
"contraseña", "desde", "descripción", "envía", "lectura", "modelo",
"obligatorio", "opcional", "para", "puerto", "requiere", "siempre",
"sin", "tienen", "traduce",
}
FUNCTION_PATTERN = re.compile(
r"(?m)^[ \t]*"
r"(?P<prefix>(?:static\s+)?(?:inline\s+)?(?:const\s+)?"
r"(?:(?:struct|enum)\s+[A-Za-z_]\w*\s*\*?\s*|"
r"[A-Za-z_]\w*(?:\s+|\s*\*+\s*))+?)"
r"(?P<name>[A-Za-z_]\w*)\s*"
r"\((?P<parameters>[^;{}]*)\)\s*\{"
)
@dataclasses.dataclass(frozen=True)
class Function:
name: str
relative_file: str
line: int
body: str
is_static: bool
@property
def node_id(self) -> str:
raw = f"{self.relative_file}_{self.name}"
return "function_" + re.sub(r"[^A-Za-z0-9_]", "_", raw)
MODULE_DESCRIPTIONS = {
"main.c": "Daemon lifecycle and one-second scheduler.",
"config.c": "Static daemon configuration parser and defaults.",
"state.c": "Persistent adoption and controller state.",
"models.c": "Emulated hardware model registry.",
"announce.c": "UniFi layer-2 UDP discovery.",
"lldp.c": "LLDP frame transmission and neighbor collection.",
"http.c": "Minimal HTTP/1.0 transport.",
"crypto.c": "AES-CBC, AES-GCM, and encoding helpers.",
"sysinfo.c": "Kernel and nl80211 device telemetry.",
"clients.c": "Wireless and bridge client telemetry.",
"inform/inform.c": "One inform request/response exchange.",
"inform/payload.c": "Inform JSON telemetry assembly.",
"inform/packet.c": "TNBU binary envelope codec.",
"inform/response.c": "Controller command and provisioning dispatch.",
"wlan/common.c": "Shared Wi-Fi translations and validators.",
"wlan/uci.c": "Reusable UCI, VLAN, steering, and cleanup operations.",
"wlan/radio.c": "Runtime radio mapping and radio settings.",
"wlan/provision.c": "Modern setstate Wi-Fi provisioning.",
"wlan/legacy.c": "Legacy system_cfg translation.",
"wlan/telemetry.c": "Managed VAP telemetry from UCI and nl80211.",
}
def mask_comments_and_literals(source: str) -> str:
"""Replace comments and literals with spaces while preserving newlines."""
output = list(source)
index = 0
state = "code"
while index < len(source):
current = source[index]
following = source[index + 1] if index + 1 < len(source) else ""
if state == "code" and current == "/" and following == "*":
output[index] = output[index + 1] = " "
state = "block_comment"
index += 2
continue
if state == "code" and current == "/" and following == "/":
output[index] = output[index + 1] = " "
state = "line_comment"
index += 2
continue
if state == "code" and current in {'"', "'"}:
output[index] = " "
state = "string" if current == '"' else "character"
index += 1
continue
if state == "block_comment":
if current == "*" and following == "/":
output[index] = output[index + 1] = " "
state = "code"
index += 2
continue
if current != "\n":
output[index] = " "
elif state == "line_comment":
if current == "\n":
state = "code"
else:
output[index] = " "
elif state in {"string", "character"}:
if current == "\\" and following:
output[index] = " "
if following != "\n":
output[index + 1] = " "
index += 2
continue
terminator = '"' if state == "string" else "'"
if current == terminator:
state = "code"
if current != "\n":
output[index] = " "
index += 1
return "".join(output)
def matching_brace(source: str, opening_brace: int) -> int:
depth = 0
for index in range(opening_brace, len(source)):
if source[index] == "{":
depth += 1
elif source[index] == "}":
depth -= 1
if depth == 0:
return index
raise ValueError(f"unmatched opening brace at byte {opening_brace}")
def discover_functions() -> list[Function]:
functions: list[Function] = []
for path in sorted(SOURCE_ROOT.rglob("*.c")):
source = path.read_text(encoding="utf-8")
masked = mask_comments_and_literals(source)
relative_file = path.relative_to(SOURCE_ROOT).as_posix()
for match in FUNCTION_PATTERN.finditer(masked):
name = match.group("name")
if name in CONTROL_WORDS:
continue
opening_brace = match.end() - 1
try:
closing_brace = matching_brace(masked, opening_brace)
except ValueError as error:
raise ValueError(
f"cannot parse {relative_file}:{name}: {error}") from error
functions.append(Function(
name=name,
relative_file=relative_file,
line=source.count("\n", 0, match.start()) + 1,
body=masked[opening_brace + 1:closing_brace],
is_static="static" in match.group("prefix").split(),
))
return functions
def resolve_calls(functions: list[Function]) -> dict[Function, list[Function]]:
by_name: dict[str, list[Function]] = {}
for function in functions:
by_name.setdefault(function.name, []).append(function)
result: dict[Function, list[Function]] = {}
for caller in functions:
callees: set[Function] = set()
for name in re.findall(r"\b([A-Za-z_]\w*)\s*\(", caller.body):
candidates = by_name.get(name, [])
local_candidates = [
candidate for candidate in candidates
if candidate.relative_file == caller.relative_file
]
if local_candidates:
callees.update(local_candidates)
elif len(candidates) == 1:
callees.add(candidates[0])
else:
callees.update(
candidate for candidate in candidates
if not candidate.is_static
)
result[caller] = sorted(
callees, key=lambda item: (item.relative_file, item.line, item.name)
)
return result
def module_rows() -> str:
rows = []
for path in sorted(SOURCE_ROOT.rglob("*.c")):
relative = path.relative_to(SOURCE_ROOT).as_posix()
description = MODULE_DESCRIPTIONS.get(relative, "Implementation module.")
rows.append(f"| `src/{relative}` | {description} |")
return "\n".join(rows)
def developer_guide() -> str:
return f"""# Developer Guide
This guide describes the current openUF implementation. The daemon is a
single-process, single-threaded OpenWrt service. Its main loop wakes once per
second and schedules discovery, LLDP, and controller inform work.
## Safety first
openUF runs as root. It can replace managed Wi-Fi configuration, create VLAN
devices, send raw Ethernet frames, reboot the device, and persist adoption
credentials. Do not run the daemon itself on a development workstation.
Compile it with the OpenWrt toolchain and perform runtime tests on a disposable
OpenWrt access point.
Protocol debug level 2 can log decrypted credentials. Never enable it by
default, attach those logs to issues, or commit controller payload captures.
## Source layout
| Module | Responsibility |
| --- | --- |
{module_rows()}
Public contracts remain in `src/*.h`. The `src/inform/` and `src/wlan/`
directories contain private implementation units; their `*_internal.h` files
are not stable interfaces for other subsystems.
## Runtime architecture
`main()` loads static configuration, persistent controller state, and the
emulated model. It discovers the LAN MAC/IP and an initial controller from the
default route when no explicit controller is configured. The main loop then
schedules:
1. `announce_send()` for UniFi UDP discovery.
2. `lldp_send_frame()` for each model Ethernet port.
3. `inform_send()` for adoption, telemetry, commands, and provisioning.
See [[Runtime Flow|Runtime-Flow]] for the decision flow and
[[Function Call Graph|Call-Graph]] for generated caller/callee relationships.
## Inform and adoption invariants
- The TNBU layout, big-endian fields, flag meanings, authenticated header, and
cipher selection are controller compatibility constraints.
- An unadopted device always encrypts with `DEFAULT_AUTH_KEY`, even if stale
state contains another key.
- The controller response is decoded before command dispatch. Adoption can be
completed through either legacy `set-adopt` or modern `setparam` data.
- `cfgversion` records configuration that was successfully applied locally.
Failed provisioning resets it to `"0"` so the controller retries.
- Increase `OPENUF_CONFIG_SCHEMA` only when an existing persisted
configuration must be reapplied after an upgrade.
## Wi-Fi ownership and provisioning
Only UCI `wifi-iface` sections prefixed with `openuf_` belong to this daemon.
Cleanup must preserve every unrelated section. Controller VAP ObjectIds are
stored in `openuf_vap_id` so client topology remains stable across reprovision.
Modern `setstate` data is applied by `wlan_apply_config()`. Legacy
newline-separated `system_cfg` data is first translated to the same JSON shape
by `wlan_apply_system_cfg()`, keeping one provisioning path responsible for UCI
commits and radio startup.
Model band and port assumptions belong in `models.c`. The runtime radio mapper
may resolve a model band to a different local PHY, but protocol and telemetry
code must not invent model-specific mappings.
## Common development tasks
### Add telemetry
1. Add a bounded reader to `sysinfo.c`, `clients.c`, or another focused module.
2. Add the controller field in `inform/payload.c`.
3. Document units, fallback behavior, ownership, and any sensitive content.
4. Regenerate this Wiki and cross-build the package.
### Add a controller command
1. Extend dispatch in `inform/response.c`.
2. Validate controller values before changing state or invoking a command.
3. Save state only after a coherent transition.
4. Preserve adoption-key and `cfgversion` behavior.
### Add a Wi-Fi setting
1. Parse controller aliases in `wlan/provision.c` or `wlan/legacy.c`.
2. Put reusable UCI work in `wlan/uci.c` and translations in `wlan/common.c`.
3. Report the effective value from `wlan/telemetry.c` when the controller
expects it in `vap_table`.
4. Test on device; a successful cross-build cannot validate netifd/hostapd
behavior.
### Add or change a model
Update `ufmodel.h`, the complete model entry in `models.c`, and every affected
telemetry or WLAN consumer together. Do not scatter model checks across the
protocol implementation.
## Build and validation
From the OpenWrt root:
```sh
make package/OpenUniFi/compile
```
Use `V=s` for detailed compiler diagnostics. If a clean package rebuild is
needed, clean only this package:
```sh
make package/OpenUniFi/clean
make package/OpenUniFi/compile
```
`Makefile.standalone` is only for compiling directly on an OpenWrt device with
development packages installed. It is not a host-side test substitute.
## Documentation workflow
Generate pages after changing C code or architecture:
```sh
./scripts/docs/generate-wiki.py
```
Run the linter-style drift and structure check in CI or before committing:
```sh
./scripts/docs/generate-wiki.py --check
```
The checker parses C functions recursively, rebuilds internal call edges,
validates required runtime entry points, rejects known non-English comment
fragments, and limits each implementation unit to
{MAX_IMPLEMENTATION_LINES} lines.
## Publishing to the Gitea Wiki
Gitea stores a repository Wiki in a separate Git repository whose URL normally
ends in `.wiki.git`. The generated `wiki/` directory is ready for that remote.
Run `scripts/docs/publish-wiki.sh` from a trusted machine with suitable
credentials. It derives the Wiki remote from `origin`; an explicit URL may be
passed when needed. The publisher regenerates and checks the pages,
updates only the generated Markdown files, commits changed pages, and pushes
them to the Wiki repository.
"""
def home_page() -> str:
return """# openUF Developer Wiki
This Wiki is generated from the current source tree and maintained in the main
repository so architecture documentation changes can be reviewed with code.
- [[Developer Guide|Developer-Guide]] — architecture, invariants, extension
points, build validation, and Wiki publishing.
- [[Runtime Flow|Runtime-Flow]] — lifecycle, inform, response, and Wi-Fi
provisioning flow charts.
- [[Function Call Graph|Call-Graph]] — generated internal caller/callee graph
and searchable function table.
Run `./scripts/docs/generate-wiki.py --check` to verify these pages are current.
"""
def call_graph_page(functions: list[Function], calls: dict[Function, list[Function]]) -> str:
lines = [
"# Function Call Graph", "",
"Generated from `src/**/*.c` by `scripts/docs/generate-wiki.py`.", "",
"```mermaid", "flowchart LR",
]
by_file: dict[str, list[Function]] = {}
for function in functions:
by_file.setdefault(function.relative_file, []).append(function)
for index, (relative_file, file_functions) in enumerate(sorted(by_file.items())):
lines.append(f' subgraph module_{index}["src/{relative_file}"]')
for function in file_functions:
lines.append(f' {function.node_id}["{function.name}()"]')
lines.append(" end")
for caller in functions:
for callee in calls[caller]:
lines.append(f" {caller.node_id} --> {callee.node_id}")
lines.extend(["```", "", "## Caller/callee index", "",
"| Source | Caller | Internal callees |",
"| --- | --- | --- |"])
for caller in functions:
callees = ", ".join(
f"`{callee.name}()`" for callee in calls[caller]
) or "—"
lines.append(
f"| `src/{caller.relative_file}:{caller.line}` | "
f"`{caller.name}()` | {callees} |"
)
return "\n".join(lines) + "\n"
def runtime_flow_page() -> str:
return """# Runtime Flow
The flow is generated alongside the call graph. Function names are validated
against the current C sources so renamed or removed stages fail documentation
checks.
## Daemon scheduler
```mermaid
flowchart TD
start([main]) --> config[config_load]
config --> state[state_load]
state --> model[ufmodel_find]
model --> identity[Read LAN MAC, IP, and default gateway]
identity --> persist[state_save]
persist --> announceInit[announce_init when enabled]
announceInit --> loop{One-second main loop}
loop -->|announce due| announce[announce_send]
announce --> loop
loop -->|LLDP due| ports{For every model port}
ports --> lldp[lldp_send_frame]
lldp --> loop
loop -->|inform due| refresh[Refresh controller URL and device IP]
refresh --> inform[inform_send]
inform --> loop
loop -->|nothing due| sleep[sleep 1 second]
sleep --> loop
```
## Inform exchange
```mermaid
flowchart TD
send([inform_send]) --> url{inform URL available?}
url -->|no| error[Return an error]
url -->|yes| key{Device adopted?}
key -->|no| defaultKey[Use DEFAULT_AUTH_KEY]
key -->|yes| stateKey[Use persisted device key]
defaultKey --> payload[inform_build_payload]
stateKey --> payload
payload --> packet[inform_packet_build]
packet --> post[http_post]
post --> status{HTTP 200?}
status -->|no, first attempt| retry[Retry with alternate CBC or GCM cipher]
retry --> packet
status -->|no, final attempt| error
status -->|yes| decode[inform_packet_parse]
decode --> json[Parse controller JSON]
json --> handle[inform_handle_response]
```
## Controller response and provisioning
```mermaid
flowchart TD
response([inform_handle_response]) --> type{_type}
type -->|noop| done[No state change]
type -->|upgrade| version[Persist requested firmware version]
type -->|reboot or reset| reboot[reboot_openwrt]
type -->|cmd adopt| adopt[Persist key, URL, and adopted state]
type -->|setparam| params[Parse management parameters]
params --> legacy{system_cfg present?}
legacy -->|yes| legacyApply[wlan_apply_system_cfg]
legacyApply --> normalize[Build radio_table and vap_table JSON]
normalize --> apply[wlan_apply_config]
type -->|setstate| apply
apply --> ids[wlan_ensure_vap_ids]
ids --> clear[wlan_clear managed openuf_ VAPs]
clear --> radios[wlan_apply_radio for each band]
radios --> vaps[Create controller VAP sections]
vaps --> commit[Commit UCI and configure usteer]
commit --> start[Start and verify radios sequentially]
start --> result{Apply succeeded?}
result -->|yes| saved[Save cfgversion and config schema]
result -->|no| retryConfig[Reset cfgversion to 0]
```
"""
def sidebar() -> str:
return """- [[Home]]
- [[Developer Guide|Developer-Guide]]
- [[Runtime Flow|Runtime-Flow]]
- [[Function Call Graph|Call-Graph]]
"""
def validate_source(functions: list[Function]) -> list[str]:
errors: list[str] = []
names = {function.name for function in functions}
required = {
"main", "config_load", "state_load", "state_save", "ufmodel_find",
"announce_init", "announce_send", "lldp_send_frame", "inform_send",
"inform_build_payload", "inform_packet_build", "http_post",
"inform_packet_parse", "inform_handle_response", "reboot_openwrt",
"wlan_apply_system_cfg", "wlan_apply_config", "wlan_ensure_vap_ids",
"wlan_clear", "wlan_apply_radio",
}
missing = sorted(required - names)
if missing:
errors.append("runtime documentation references missing functions: " +
", ".join(missing))
for path in sorted(SOURCE_ROOT.rglob("*.c")):
source = path.read_text(encoding="utf-8")
line_count = source.count("\n") + 1
if line_count > MAX_IMPLEMENTATION_LINES:
errors.append(
f"{path.relative_to(REPOSITORY_ROOT)} has {line_count} lines; "
f"split modules above {MAX_IMPLEMENTATION_LINES} lines"
)
comments = "\n".join(re.findall(r"/\*.*?\*/|//[^\n]*", source,
flags=re.DOTALL)).lower()
found = sorted(
word for word in NON_ENGLISH_COMMENT_WORDS
if re.search(rf"\b{re.escape(word)}\b", comments)
)
if found:
errors.append(
f"{path.relative_to(REPOSITORY_ROOT)} contains non-English "
f"comment words: {', '.join(found)}"
)
return errors
def render_pages(functions: list[Function]) -> dict[str, str]:
calls = resolve_calls(functions)
return {
"Home.md": home_page(),
"Developer-Guide.md": developer_guide(),
"Call-Graph.md": call_graph_page(functions, calls),
"Runtime-Flow.md": runtime_flow_page(),
"_Sidebar.md": sidebar(),
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--check", action="store_true",
help="fail if generated pages or source checks are stale")
parser.add_argument("--output", type=Path, default=DEFAULT_WIKI_ROOT,
help="Wiki output directory (default: repository wiki/)")
arguments = parser.parse_args()
functions = discover_functions()
errors = validate_source(functions)
pages = render_pages(functions)
output = arguments.output.resolve()
if arguments.check:
for filename, expected in pages.items():
path = output / filename
if not path.exists():
errors.append(f"missing generated page: {path}")
elif path.read_text(encoding="utf-8") != expected:
errors.append(f"generated page is stale: {path}")
else:
output.mkdir(parents=True, exist_ok=True)
for filename, content in pages.items():
(output / filename).write_text(content, encoding="utf-8")
if errors:
for error in errors:
print(f"documentation error: {error}", file=sys.stderr)
if arguments.check:
print("run ./scripts/docs/generate-wiki.py to refresh pages",
file=sys.stderr)
return 1
action = "verified" if arguments.check else "generated"
print(f"{action} {len(pages)} Wiki pages from {len(functions)} C functions")
return 0
if __name__ == "__main__":
raise SystemExit(main())