streamdeck-obs-replay/obsws/codegen/protocol.py

403 lines
12 KiB
Python

# TODO: No warning is given for struct fields whose names match function names.
# Maybe bring the reserved list back.
import json
import os
import sys
from typing import Dict, List, Tuple
package = "obsws"
doc = "https://github.com/Palakis/obs-websocket/blob/4.3-maintenance/docs/generated/protocol.md"
disclaimer = """
// This file is automatically generated.
// https://github.com/christopher-dG/go-obs-websocket/blob/master/codegen/protocol.py
"""
# TODO: Test the less clear ones.
type_map = {
"bool": "bool",
"boolean": "bool",
"int": "int",
"integer": "int",
"float": "float64",
"double": "float64",
"string": "string",
"array": "[]interface{}",
"object": "map[string]interface{}",
"array of objects": "[]map[string]interface{}",
"object|array": "interface{}",
"scene|array": "[]map[string]interface{}",
"source|array": "[]map[string]interface{}",
}
unknown_types = []
def process_json(d: Dict):
"""Generate Go code for the entire protocol file."""
if "--events" in sys.argv or "--all" in sys.argv:
gen_events(d["events"])
if "--requests" in sys.argv or "--all" in sys.argv:
gen_requests(d["requests"])
def gen_category(prefix: str, category: str, data: Dict):
"""Generate all events or requests in one category."""
func = gen_event if prefix == "events" else gen_request
content = "\n".join(
filter(
lambda s: not s.isspace(),
"\n".join(func(thing) for thing in data).split("\n"),
)
)
with open(f"{prefix}_{category}.go".replace(" ", "_"), "w") as f:
f.write(
f"""
package {package}
{disclaimer}
{content}
"""
)
def gen_events(events: Dict):
"""Generate all events."""
for category, data in events.items():
gen_category("events", category, data)
gen_event_utils(events)
def gen_event(data: Dict) -> str:
"""Write Go code with a type definition and interface functions."""
struct = f"""
type {data["name"]}Event struct {{
{go_struct_variables(go_variables(data.get("returns", [])))}
_event `json:",squash"`
}}
"""
description = newlinify(f"{data['name']}Event : {data['description']}")
if not description.endswith("."):
description += "."
if data.get("since"):
description += (
f"\n//\n// Since obs-websocket version: {data['since'].capitalize()}."
)
return f"""
{description}
//
// {doc}#{data["heading"]["text"].lower()}
{struct}
"""
def gen_requests(requests: Dict):
"""Generate all requests and responses."""
for category, data in requests.items():
gen_category("requests", category, data)
def gen_request(data: Dict) -> str:
"""Write Go code with type definitions and interface functions."""
struct = f"""
type {data["name"]}Request struct {{
{go_struct_variables(go_variables(data.get("params", [])))}
_request `json:",squash"`
response chan {data["name"]}Response
}}
"""
description = newlinify(f"{data['name']}Request : {data['description']}")
if description and not description.endswith("."):
description += "."
if data.get("since"):
description += (
f"\n//\n// Since obs-websocket version: {data['since'].capitalize()}."
)
request = f"""
{description}
//
// {doc}#{data["heading"]["text"].lower()}
{struct}
{gen_request_new(data)}
// Send sends the request.
func (r *{data["name"]}Request) Send(c *Client) error {{
if r.sent {{
return ErrAlreadySent
}}
future, err := c.sendRequest(r)
if err != nil {{
return err
}}
r.sent = true
go func() {{
m := <-future
var resp {data["name"]}Response
if err = mapToStruct(m, &resp); err != nil {{
r.err <- err
}} else if resp.Status() != StatusOK {{
r.err <- errors.New(resp.Error())
}} else {{
r.response <- resp
}}
}}()
return nil
}}
// Receive waits for the response.
func (r {data["name"]}Request) Receive() ({data["name"]}Response, error) {{
if !r.sent {{
return {data["name"]}Response{{}}, ErrNotSent
}}
if receiveTimeout == 0 {{
select {{
case resp := <-r.response:
return resp, nil
case err := <-r.err:
return {data["name"]}Response{{}}, err
}}
}} else {{
select {{
case resp := <-r.response:
return resp, nil
case err := <-r.err:
return {data["name"]}Response{{}}, err
case <-time.After(receiveTimeout):
return {data["name"]}Response{{}}, ErrReceiveTimeout
}}
}}
}}
// SendReceive sends the request then immediately waits for the response.
func (r {data["name"]}Request) SendReceive(c *Client) ({data["name"]}Response, error) {{
if err := r.Send(c); err != nil {{
return {data["name"]}Response{{}}, err
}}
return r.Receive()
}}
"""
if data.get("returns"):
struct = f"""
type {data["name"]}Response struct {{
{go_struct_variables(go_variables(data["returns"]))}
_response `json:",squash"`
}}
"""
else:
struct = (
f"""type {data["name"]}Response struct {{ _response `json:",squash"`}}"""
)
description = f"// {data['name']}Response : Response for {data['name']}Request."
if data.get("since"):
description += (
f"\n//\n// Since obs-websocket version: {data['since'].capitalize()}."
)
response = f"""
{description}
//
// {doc}#{data["heading"]["text"].lower()}
{struct}
"""
return f"{request}\n\n{response}"
def gen_request_new(request: Dict):
"""Generate Go code with a New___Request function for a request type."""
base = f"""
// New{request["name"]}Request returns a new {request["name"]}Request.
func New{request["name"]}Request(\
"""
variables = go_variables(request.get("params", []), export=False)
default_args = f"""
_request{{
ID_: getMessageID(),
Type_: "{request["name"]}",
err: make(chan error, 1),
}},
make(chan {request["name"]}Response, 1),
"""
if not variables:
sig = f"{base}) {request['name']}Request"
constructor_args = f"""{{
{default_args}
}}
"""
else:
args = "\n".join(
f"{'_type' if var['name'] == 'type' else var['name']} {var['type']},"
for var in variables
)
constructor_args = (
"{\n"
+ "\n".join(
"_type," if var["name"] == "type" else f"{var['name']},"
for var in variables
)
+ default_args
+ "}"
)
if len(variables) == 1:
sig = f"{base}{args}) {request['name']}Request"
else:
sig = f"""
{base}
{args}
) {request["name"]}Request\
"""
return f"{sig} {{ return {request['name']}Request{constructor_args} }}"
def gen_event_utils(events: Dict):
"""
Generate a Go file with a mappings from type names to structs,
and a function for dereferencing interface pointers.
"""
event_map = {}
event_list = []
for category in events.values():
for e in category:
event_map[e["name"]] = f"func() Event {{ return &{e['name']}Event{{}} }}"
event_list.append(f"*{e['name']}Event")
event_entries = "\n".join(f'"{k}": {v},' for k, v in event_map.items())
event_cases = "\n".join(f"case {e}:\nreturn *e" for e in event_list)
deref = f"""
func derefEvent(e Event) Event {{
switch e := e.(type) {{
{event_cases}
default:
return nil
}}
}}
"""
with open("event_utils.go", "w") as f:
f.write(
f"""
package {package}
{disclaimer}
var eventMap = map[string]func() Event {{
{event_entries}
}}
// derefEvent returns an Event struct from a pointer to an Event struct.
{deref}
"""
)
def go_variables(variables: List[Dict], export: bool = True) -> str:
"""
Convert a list of variable names into Go code to be put
inside a struct definition.
"""
vardicts, varnames = [], []
for v in variables:
typename, optional = optional_type(v["type"])
varname = go_var(v["name"], export=export)
vardicts.append(
{
"name": varname,
"type": type_map[typename.lower()],
"tag": f'`json:"{v["name"]}"`',
"description": v["description"].replace("\n", " "),
"optional": optional,
"unknown": typename.lower() in unknown_types,
"actual_type": v["type"],
"duplicate": varname in varnames,
}
)
varnames.append(varname)
return vardicts
def go_var(s: str, export: bool = True) -> str:
"""Convert a variable name in the input file to a Go variable name."""
s = f"{(str.upper if export else str.lower)(s[0])}{s[1:]}"
for sep in ["-", "_", ".*.", "[].", "."]:
while sep in s:
_len = len(sep)
if s.endswith(sep):
s = s[:-_len]
continue
i = s.find(sep)
s = f"{s[:i]}{s[i+_len].upper()}{s[i+_len+1:]}"
return s.replace("Id", "ID").replace("Obs", "OBS").replace("Fps", "FPS")
def go_struct_variables(variables: List[Dict]) -> str:
"""Generate Go code containing struct field definitions."""
lines = []
for var in variables:
if var["description"]:
description = (
var["description"]
.replace("e.g. ", "e.g.")
.replace(". ", "\n")
.replace("e.g.", "e.g. ")
)
for desc_line in description.split("\n"):
desc_line = desc_line.strip()
if desc_line and not desc_line.endswith("."):
desc_line += "."
lines.append(f"// {desc_line}")
lines.append(f"// Required: {'Yes' if not var['optional'] else 'No'}.")
todos = []
if var["unknown"]:
todos.append(f"Unknown type ({var['actual_type']})")
if var["duplicate"]:
todos.append("Duplicate name")
todos = " ".join(f"TODO: {todo}." for todo in todos)
if todos:
lines.append(f"// {todos}")
lines.append(f"{var['name']} {var['type']} {var['tag']}")
return "\n".join(lines)
def newlinify(s: str, comment: bool = True) -> str:
"""Put each sentence of a string onto its own line."""
s = s.replace("e.g. ", "e.g.").replace(". ", "\n").replace("e.g.", "e.g. ")
if comment:
s = "\n".join(
[f"// {_s}" if not _s.startswith("//") else _s for _s in s.split("\n")]
)
return s
def optional_type(s: str) -> Tuple[str, bool]:
"""Determine if a type is optional and parse the actual type name."""
if s.endswith("(optional)"):
return s[: s.find("(optional)")].strip(), True
return s, False
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Missing filename argument")
exit(1)
if not os.path.isfile(sys.argv[1]):
print(f"file '{sys.argv[1]}' does not exist")
exit(1)
with open(sys.argv[1]) as f:
d = json.load(f)
process_json(d)
os.system("goimports -w *.go")