Commit cb2098be authored by payno's avatar payno
Browse files

[scheme] add some documentation + python typing

parent f7d376e0
......@@ -36,6 +36,7 @@ import pprint
import base64
import pickle
import logging
import typing
from .node import Node
from .link import Link
from ast import literal_eval
......@@ -47,15 +48,16 @@ class Scheme(object):
class to define a workflow scheme from nodes and links
:param list nodes:
:param list links:
:param typing.Iterable nodes: set of Node
:param typing.Iterable list links: set of Link
_JSON_DESCRIPTION = 'description'
_JSON_TITLE = 'title'
_JSON_NODES = 'nodes'
_JSON_LINKS = 'links'
def __init__(self, nodes=None, links=None, description=None, title=None):
def __init__(self, nodes:typing.Iterable=None, links:typing.Iterable=None,
description: str=None, title: str=None):
self._reset(nodes=nodes, links=links, description=description,
......@@ -116,7 +118,7 @@ class Scheme(object):
return res
def save_to(self, output_file):
def save_to(self, output_file: str):
Save the scheme as an xml formated file to `stream`
......@@ -128,7 +130,7 @@ class Scheme(object):
def save_as_xml(self, output_file):
def save_as_xml(self, output_file: str):
save current scheme to a default xml format
......@@ -139,7 +141,7 @@ class Scheme(object):
def save_as_json(self, output_file):
def save_as_json(self, output_file: str):
save current scheme to a default json format
......@@ -203,6 +205,12 @@ class Scheme(object):
return links
def to_json(self):
Convert scheme to json
:return: json dict
:rtype: dict
return {
self._JSON_DESCRIPTION: self.description,
self._JSON_TITLE: self.title,
......@@ -211,7 +219,15 @@ class Scheme(object):
def from_json_file(json_file_path):
def from_json_file(json_file_path: str):
Create and load Scheme from a json file
:param str json_file_path: json file
:return: Scheme fitting the json description
:rtype: Scheme
:raises: ValueError if file not found or invalid
scheme = Scheme()
......@@ -221,7 +237,7 @@ class Scheme(object):
return scheme
def load_from_json_file(self, json_file_path):
def load_from_json_file(self, json_file_path: str):
:param str json_file_path: path to the json file containing the scheme
......@@ -438,20 +454,46 @@ class Scheme(object):
class SubScheme(Scheme, Node):
def __init__(self, nodes, links, description=None, error_handler=None):
Define a sub-scheme of the workflow (or subworkflow). SubScheme are
as Scheme expect that they are not 'root'.
:param nodes: set of Node
:type: typing.Iterable
:param links: set of Node
:type: typing.Iterable
:param description: description of the subscheme
:type: str
:param error_handler: ErrorHandler
def __init__(self, nodes: typing.Iterable, links: typing.Iterable,
description: str=None, error_handler=None):
Node.__init__(self, processing_pt=None, error_handler=error_handler)
Scheme.__init__(self, nodes=nodes, links=links, title='',
def contains_control_nodes(nodes_list):
def contains_control_nodes(nodes_list: typing.Iterable):
Return the list of the 'control' nodes.
:param typing.Iterable nodes_list:
for _node in nodes_list:
if _node.endless or contains_control_nodes(_node.upstream_nodes):
return True
return False
def loads(string, format):
def loads(string: str, format: str) -> object:
"""load stream from the given format.
:param str string: stream to load
:param str format: format
:return: object contained in the stream
:rtype: object
if format == "literal":
return literal_eval(string)
elif format == "json":
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment