459 lines
19 KiB
Python
Executable File
459 lines
19 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Tuple, Union
|
|
|
|
import yaml
|
|
|
|
if __name__ == "__main__":
|
|
sys.path.insert(0, str(Path(__file__).parent.parent)) # add src/wireviz to PATH
|
|
|
|
from wireviz.DataClasses import Metadata, Options, Tweak
|
|
from wireviz.Harness import Harness
|
|
from wireviz.wv_helper import (
|
|
expand,
|
|
get_single_key_and_value,
|
|
is_arrow,
|
|
open_file_read,
|
|
smart_file_resolve,
|
|
)
|
|
|
|
from . import APP_NAME
|
|
|
|
|
|
def parse(
|
|
inp: Union[Path, str, Dict],
|
|
return_types: Union[None, str, Tuple[str]] = None,
|
|
output_formats: Union[None, str, Tuple[str]] = None,
|
|
output_dir: Union[str, Path] = None,
|
|
output_name: Union[None, str] = None,
|
|
image_paths: Union[Path, str, List] = [],
|
|
) -> Any:
|
|
"""
|
|
This function takes an input, parses it as a WireViz Harness file,
|
|
and outputs the result as one or more files and/or as a function return value
|
|
|
|
Accepted inputs:
|
|
* A path to a YAML source file to parse
|
|
* A string containing the YAML data to parse
|
|
* A Python Dict containing the pre-parsed YAML data
|
|
|
|
Supported return types:
|
|
* "png": the diagram as raw PNG data
|
|
* "svg": the diagram as raw SVG data
|
|
* "harness": the diagram as a Harness Python object
|
|
|
|
Supported output formats:
|
|
* "csv": the BOM, as a comma-separated text file
|
|
* "gv": the diagram, as a GraphViz source file
|
|
* "html": the diagram and (depending on the template) the BOM, as a HTML file
|
|
* "png": the diagram, as a PNG raster image
|
|
* "pdf": the diagram and (depending on the template) the BOM, as a PDF file
|
|
* "svg": the diagram, as a SVG vector image
|
|
* "tsv": the BOM, as a tab-separated text file
|
|
|
|
Args:
|
|
inp (Path | str | Dict):
|
|
The input to be parsed (see above for accepted inputs).
|
|
return_types (optional):
|
|
One of the supported return types (see above), or a tuple of multiple return types.
|
|
If set to None, no output is returned by the function.
|
|
output_formats (optional):
|
|
One of the supported output types (see above), or a tuple of multiple output formats.
|
|
If set to None, no files are generated.
|
|
output_dir (Path | str, optional):
|
|
The directory to place the generated output files.
|
|
Defaults to inp's parent directory, or cwd if inp is not a path.
|
|
output_name (str, optional):
|
|
The name to use for the generated output files (without extension).
|
|
Defaults to inp's file name (without extension).
|
|
Required parameter if inp is not a path.
|
|
image_paths (Path | str | List, optional):
|
|
Paths to use when resolving any image paths included in the data.
|
|
Note: If inp is a path to a YAML file,
|
|
its parent directory will automatically be included in the list.
|
|
|
|
Returns:
|
|
Depending on the return_types parameter, may return:
|
|
* None
|
|
* one of the following, or a tuple containing two or more of the following:
|
|
* PNG data
|
|
* SVG data
|
|
* a Harness object
|
|
"""
|
|
|
|
if not output_formats and not return_types:
|
|
raise Exception("No output formats or return types specified")
|
|
|
|
yaml_data, yaml_file = _get_yaml_data_and_path(inp)
|
|
assert isinstance(
|
|
yaml_data, dict
|
|
), f"Expected a dict as top-level YAML input, but got: {type(yaml_data)}"
|
|
if output_formats:
|
|
# need to write data to file, determine output directory and filename
|
|
output_dir = _get_output_dir(yaml_file, output_dir)
|
|
output_name = _get_output_name(yaml_file, output_name)
|
|
output_file = output_dir / output_name
|
|
|
|
if yaml_file:
|
|
# if reading from file, ensure that input file's parent directory is included in image_paths
|
|
default_image_path = yaml_file.parent.resolve()
|
|
if not default_image_path in [Path(x).resolve() for x in image_paths]:
|
|
image_paths.append(default_image_path)
|
|
|
|
# define variables =========================================================
|
|
# containers for parsed component data and connection sets
|
|
template_connectors = {}
|
|
template_cables = {}
|
|
connection_sets = []
|
|
# actual harness
|
|
harness = Harness(
|
|
metadata=Metadata(**yaml_data.get("metadata", {})),
|
|
options=Options(**yaml_data.get("options", {})),
|
|
tweak=Tweak(**yaml_data.get("tweak", {})),
|
|
)
|
|
# others
|
|
# store mapping of components to their respective template
|
|
designators_and_templates = {}
|
|
# keep track of auto-generated designators to avoid duplicates
|
|
autogenerated_designators = {}
|
|
|
|
# When title is not given, either deduce it from filename, or use default text.
|
|
if "title" not in harness.metadata:
|
|
harness.metadata["title"] = output_name or f"{APP_NAME} diagram and BOM"
|
|
|
|
# add items
|
|
# parse YAML input file ====================================================
|
|
|
|
sections = ["connectors", "cables", "connections"]
|
|
types = [dict, dict, list]
|
|
for sec, ty in zip(sections, types):
|
|
if sec in yaml_data and type(yaml_data[sec]) == ty: # section exists
|
|
if len(yaml_data[sec]) > 0: # section has contents
|
|
if ty == dict:
|
|
for key, attribs in yaml_data[sec].items():
|
|
# The Image dataclass might need to open an image file with a relative path.
|
|
image = attribs.get("image")
|
|
if isinstance(image, dict):
|
|
image_path = image["src"]
|
|
if image_path and not Path(image_path).is_absolute():
|
|
# resolve relative image path
|
|
image["src"] = smart_file_resolve(
|
|
image_path, image_paths
|
|
)
|
|
if sec == "connectors":
|
|
template_connectors[key] = attribs
|
|
elif sec == "cables":
|
|
template_cables[key] = attribs
|
|
else: # section exists but is empty
|
|
pass
|
|
else: # section does not exist, create empty section
|
|
if ty == dict:
|
|
yaml_data[sec] = {}
|
|
elif ty == list:
|
|
yaml_data[sec] = []
|
|
|
|
connection_sets = yaml_data["connections"]
|
|
|
|
# go through connection sets, generate and connect components ==============
|
|
|
|
template_separator_char = harness.options.template_separator
|
|
|
|
def resolve_designator(inp, separator):
|
|
if separator in inp: # generate a new instance of an item
|
|
if inp.count(separator) > 1:
|
|
raise Exception(f"{inp} - Found more than one separator ({separator})")
|
|
template, designator = inp.split(separator)
|
|
if designator == "":
|
|
autogenerated_designators[template] = (
|
|
autogenerated_designators.get(template, 0) + 1
|
|
)
|
|
designator = f"__{template}_{autogenerated_designators[template]}"
|
|
# check if redefining existing component to different template
|
|
if designator in designators_and_templates:
|
|
if designators_and_templates[designator] != template:
|
|
raise Exception(
|
|
f"Trying to redefine {designator} from {designators_and_templates[designator]} to {template}"
|
|
)
|
|
else:
|
|
designators_and_templates[designator] = template
|
|
else:
|
|
template, designator = (inp, inp)
|
|
if designator in designators_and_templates:
|
|
pass # referencing an exiting connector, no need to add again
|
|
else:
|
|
designators_and_templates[designator] = template
|
|
return (template, designator)
|
|
|
|
# utilities to check for alternating connectors and cables/arrows ==========
|
|
|
|
alternating_types = ["connector", "cable/arrow"]
|
|
expected_type = None
|
|
|
|
def check_type(designator, template, actual_type):
|
|
nonlocal expected_type
|
|
if not expected_type: # each connection set may start with either section
|
|
expected_type = actual_type
|
|
|
|
if actual_type != expected_type: # did not alternate
|
|
raise Exception(
|
|
f'Expected {expected_type}, but "{designator}" ("{template}") is {actual_type}'
|
|
)
|
|
|
|
def alternate_type(): # flip between connector and cable/arrow
|
|
nonlocal expected_type
|
|
expected_type = alternating_types[1 - alternating_types.index(expected_type)]
|
|
|
|
for connection_set in connection_sets:
|
|
|
|
# figure out number of parallel connections within this set
|
|
connectioncount = []
|
|
for entry in connection_set:
|
|
if isinstance(entry, list):
|
|
connectioncount.append(len(entry))
|
|
elif isinstance(entry, dict):
|
|
connectioncount.append(len(expand(list(entry.values())[0])))
|
|
# e.g.: - X1: [1-4,6] yields 5
|
|
else:
|
|
pass # strings do not reveal connectioncount
|
|
if not any(connectioncount):
|
|
# no item in the list revealed connection count;
|
|
# assume connection count is 1
|
|
connectioncount = [1]
|
|
# Example: The following is a valid connection set,
|
|
# even though no item reveals the connection count;
|
|
# the count is not needed because only a component-level mate happens.
|
|
# -
|
|
# - CONNECTOR
|
|
# - ==>
|
|
# - CONNECTOR
|
|
|
|
# check that all entries are the same length
|
|
if len(set(connectioncount)) > 1:
|
|
raise Exception(
|
|
"All items in connection set must reference the same number of connections"
|
|
)
|
|
# all entries are the same length, connection count is set
|
|
connectioncount = connectioncount[0]
|
|
|
|
# expand string entries to list entries of correct length
|
|
for index, entry in enumerate(connection_set):
|
|
if isinstance(entry, str):
|
|
connection_set[index] = [entry] * connectioncount
|
|
|
|
# resolve all designators
|
|
for index, entry in enumerate(connection_set):
|
|
if isinstance(entry, list):
|
|
for subindex, item in enumerate(entry):
|
|
template, designator = resolve_designator(
|
|
item, template_separator_char
|
|
)
|
|
connection_set[index][subindex] = designator
|
|
elif isinstance(entry, dict):
|
|
key = list(entry.keys())[0]
|
|
template, designator = resolve_designator(key, template_separator_char)
|
|
value = entry[key]
|
|
connection_set[index] = {designator: value}
|
|
else:
|
|
pass # string entries have been expanded in previous step
|
|
|
|
# expand all pin lists
|
|
for index, entry in enumerate(connection_set):
|
|
if isinstance(entry, list):
|
|
connection_set[index] = [{designator: 1} for designator in entry]
|
|
elif isinstance(entry, dict):
|
|
designator = list(entry.keys())[0]
|
|
pinlist = expand(entry[designator])
|
|
connection_set[index] = [{designator: pin} for pin in pinlist]
|
|
else:
|
|
pass # string entries have been expanded in previous step
|
|
|
|
# Populate wiring harness ==============================================
|
|
|
|
expected_type = None # reset check for alternating types
|
|
# at the beginning of every connection set
|
|
# since each set may begin with either type
|
|
|
|
# generate components
|
|
for entry in connection_set:
|
|
for item in entry:
|
|
designator = list(item.keys())[0]
|
|
template = designators_and_templates[designator]
|
|
|
|
if designator in harness.connectors: # existing connector instance
|
|
check_type(designator, template, "connector")
|
|
elif template in template_connectors.keys():
|
|
# generate new connector instance from template
|
|
check_type(designator, template, "connector")
|
|
harness.add_connector(
|
|
name=designator, **template_connectors[template]
|
|
)
|
|
|
|
elif designator in harness.cables: # existing cable instance
|
|
check_type(designator, template, "cable/arrow")
|
|
elif template in template_cables.keys():
|
|
# generate new cable instance from template
|
|
check_type(designator, template, "cable/arrow")
|
|
harness.add_cable(name=designator, **template_cables[template])
|
|
|
|
elif is_arrow(designator):
|
|
check_type(designator, template, "cable/arrow")
|
|
# arrows do not need to be generated here
|
|
else:
|
|
raise Exception(
|
|
f"{template} is an unknown template/designator/arrow."
|
|
)
|
|
|
|
alternate_type() # entries in connection set must alternate between connectors and cables/arrows
|
|
|
|
# transpose connection set list
|
|
# before: one item per component, one subitem per connection in set
|
|
# after: one item per connection in set, one subitem per component
|
|
connection_set = list(map(list, zip(*connection_set)))
|
|
|
|
# connect components
|
|
for index_entry, entry in enumerate(connection_set):
|
|
for index_item, item in enumerate(entry):
|
|
designator = list(item.keys())[0]
|
|
|
|
if designator in harness.cables:
|
|
if index_item == 0:
|
|
# list started with a cable, no connector to join on left side
|
|
from_name, from_pin = (None, None)
|
|
else:
|
|
from_name, from_pin = get_single_key_and_value(
|
|
entry[index_item - 1]
|
|
)
|
|
via_name, via_pin = (designator, item[designator])
|
|
if index_item == len(entry) - 1:
|
|
# list ends with a cable, no connector to join on right side
|
|
to_name, to_pin = (None, None)
|
|
else:
|
|
to_name, to_pin = get_single_key_and_value(
|
|
entry[index_item + 1]
|
|
)
|
|
harness.connect(
|
|
from_name, from_pin, via_name, via_pin, to_name, to_pin
|
|
)
|
|
|
|
elif is_arrow(designator):
|
|
if index_item == 0: # list starts with an arrow
|
|
raise Exception(
|
|
"An arrow cannot be at the start of a connection set"
|
|
)
|
|
elif index_item == len(entry) - 1: # list ends with an arrow
|
|
raise Exception(
|
|
"An arrow cannot be at the end of a connection set"
|
|
)
|
|
|
|
from_name, from_pin = get_single_key_and_value(
|
|
entry[index_item - 1]
|
|
)
|
|
via_name, via_pin = (designator, None)
|
|
to_name, to_pin = get_single_key_and_value(entry[index_item + 1])
|
|
if "-" in designator: # mate pin by pin
|
|
harness.add_mate_pin(
|
|
from_name, from_pin, to_name, to_pin, designator
|
|
)
|
|
elif "=" in designator and index_entry == 0:
|
|
# mate two connectors as a whole
|
|
harness.add_mate_component(from_name, to_name, designator)
|
|
|
|
# warn about unused templates
|
|
|
|
proposed_components = list(template_connectors.keys()) + list(
|
|
template_cables.keys()
|
|
)
|
|
used_components = set(designators_and_templates.values())
|
|
forgotten_components = [c for c in proposed_components if not c in used_components]
|
|
if len(forgotten_components) > 0:
|
|
print(
|
|
"Warning: The following components are not referenced in any connection set:"
|
|
)
|
|
print(", ".join(forgotten_components))
|
|
|
|
# harness population completed =============================================
|
|
|
|
if "additional_bom_items" in yaml_data:
|
|
for line in yaml_data["additional_bom_items"]:
|
|
harness.add_bom_item(line)
|
|
|
|
if output_formats:
|
|
harness.output(filename=output_file, fmt=output_formats, view=False)
|
|
|
|
if return_types:
|
|
returns = []
|
|
if isinstance(return_types, str): # only one return type speficied
|
|
return_types = [return_types]
|
|
|
|
return_types = [t.lower() for t in return_types]
|
|
|
|
for rt in return_types:
|
|
if rt == "png":
|
|
returns.append(harness.png)
|
|
if rt == "svg":
|
|
returns.append(harness.svg)
|
|
if rt == "harness":
|
|
returns.append(harness)
|
|
|
|
return tuple(returns) if len(returns) != 1 else returns[0]
|
|
|
|
|
|
def _get_yaml_data_and_path(inp: Union[str, Path, Dict]) -> (Dict, Path):
|
|
# determine whether inp is a file path, a YAML string, or a Dict
|
|
if not isinstance(inp, Dict): # received a str or a Path
|
|
try:
|
|
yaml_path = Path(inp).expanduser().resolve(strict=True)
|
|
# if no FileNotFoundError exception happens, get file contents
|
|
yaml_str = open_file_read(yaml_path).read()
|
|
except (FileNotFoundError, OSError) as e:
|
|
# if inp is a long YAML string, Pathlib will raise OSError: [errno.ENAMETOOLONG]
|
|
# (in Windows, it seems OSError [errno.EINVAL] might be raised in some cases)
|
|
# when trying to expand and resolve it as a path.
|
|
# Catch this error, but raise any others
|
|
from errno import EINVAL, ENAMETOOLONG
|
|
|
|
if type(e) is OSError and e.errno not in (EINVAL, ENAMETOOLONG):
|
|
raise e
|
|
# file does not exist; assume inp is a YAML string
|
|
yaml_str = inp
|
|
yaml_path = None
|
|
yaml_data = yaml.safe_load(yaml_str)
|
|
else:
|
|
# received a Dict, use as-is
|
|
yaml_data = inp
|
|
yaml_path = None
|
|
return yaml_data, yaml_path
|
|
|
|
|
|
def _get_output_dir(input_file: Path, default_output_dir: Path) -> Path:
|
|
if default_output_dir: # user-specified output directory
|
|
output_dir = Path(default_output_dir)
|
|
else: # auto-determine appropriate output directory
|
|
if input_file: # input comes from a file; place output in same directory
|
|
output_dir = input_file.parent
|
|
else: # input comes from str or Dict; fall back to cwd
|
|
output_dir = Path.cwd()
|
|
return output_dir.resolve()
|
|
|
|
|
|
def _get_output_name(input_file: Path, default_output_name: Path) -> str:
|
|
if default_output_name: # user-specified output name
|
|
output_name = default_output_name
|
|
else: # auto-determine appropriate output name
|
|
if input_file: # input comes from a file; use same file stem
|
|
output_name = input_file.stem
|
|
else: # input comes from str or Dict; no fallback available
|
|
raise Exception("No output file name provided")
|
|
return output_name
|
|
|
|
|
|
def main():
|
|
print("When running from the command line, please use wv_cli.py instead.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|