Skip to content
GitLab
Commit e99104e1 authored by payno's avatar payno
Browse files

add python typing. /close #5

parent 51427967
Hide whitespace changes
Inline Side-by-side
scheme/link.py
View file @ e99104e1
......@@ -27,9 +27,12 @@ __authors__ = ["H.Payno"]
__license__ = "MIT"
__date__ = "29/05/2017"
from typing import Union
import logging
_logger = logging.getLogger(__name__)
# TODO: next lines should be removed
global next_link_free_id
next_link_free_id = 0
......@@ -61,8 +64,8 @@ class Link(object):
_JSON_SINK_NODE_ID = 'sink_node_id'
_JSON_LINK_ID = 'link_id'
def __init__(self, source_node, sink_node, source_channel='default',
sink_channel='default', id=None):
def __init__(self, source_node, sink_node, source_channel: str = 'default',
sink_channel: str = 'default', id=None):
self.id = get_next_link_free_id() if id is None else id
if isinstance(source_node, int):
self.source_node_id = source_node
......@@ -77,7 +80,7 @@ class Link(object):
self.source_channel = source_channel
self.sink_channel = sink_channel
def to_json(self):
def to_json(self) -> dict:
"""
:return: Link description to the json format
......@@ -92,7 +95,7 @@ class Link(object):
}
@staticmethod
def from_json(json_data):
def from_json(json_data: dict):
"""
:param json_data: link description
......
scheme/node.py
View file @ e99104e1
......@@ -30,9 +30,9 @@ __date__ = "29/05/2017"
import functools
import logging
import traceback
import json
import inspect
import importlib
from typing import Union
from importlib.machinery import SourceFileLoader
_logger = logging.getLogger(__file__)
......@@ -41,7 +41,7 @@ global next_node_free_idF
next_node_free_id = 0
def get_next_node_free_id():
def get_next_node_free_id() -> int:
global next_node_free_id
_id = next_node_free_id
next_node_free_id += 1
......@@ -88,7 +88,8 @@ class Node(object):
_JSON_PROPERTIES = 'properties'
_JSON_ERROR_HANDLER = 'error_handler'
def __init__(self, processing_pt, id=None, properties=None,
def __init__(self, processing_pt, id: Union[None, int] = None,
properties:Union[None, dict] = None,
error_handler=None):
self.id = get_next_node_free_id() if id is None else id
"""int of the node id"""
......@@ -109,14 +110,14 @@ class Node(object):
self.outData = None
@property
def handlers(self):
def handlers(self) -> dict:
return self._handlers
@property
def process_pt(self):
return self._process_pt
def isfinal(self):
def isfinal(self) -> bool:
"""
:return: True if the node is at the end of a branch.
......@@ -124,7 +125,7 @@ class Node(object):
"""
return len(self.downstream_nodes) is 0
def isstart(self):
def isstart(self) -> bool:
"""
:return: True if the node does not requires any input
......@@ -132,12 +133,15 @@ class Node(object):
"""
return len(self.upstream_nodes) is 0
def load_handlers(self):
def load_handlers(self) -> None:
"""
load handlers from the `processing_pt` defined.
For callable it will always be `processing_pt`.
But for not-callable class it will be class function defined in the
`inputs` variable.
:raises: ValueError if unable to find sme handlers in the classes
definition
"""
self._handlers.clear()
assert self._process_pt is not None
......@@ -178,7 +182,8 @@ class Node(object):
raise ValueError('Fail to init handlers, none defined for ' + str(self._process_pt))
@staticmethod
def execute(process_pt, properties, input_name, input_data):
def execute(process_pt, properties: dict, input_name: str,
input_data: object):
"""
Create an instance of a node with 'process_pt' and execute it with the
given input_name, properties and input_data.
......@@ -204,7 +209,7 @@ class Node(object):
else:
return out
def to_json(self):
def to_json(self) -> dict:
"""
:return: json description of the node
......@@ -225,7 +230,7 @@ class Node(object):
}
@staticmethod
def load_node_info_from_json(json_data):
def load_node_info_from_json(json_data: dict) -> tuple:
"""
load fom json stream the Node Information
......@@ -256,7 +261,7 @@ class Node(object):
return _id, _properties, _process_pt
@staticmethod
def from_json(json_data):
def from_json(json_data: dict):
"""
:param json_data: node description
......
scheme/scheme.py
View file @ e99104e1
......@@ -31,6 +31,7 @@ __date__ = "17/12/2018"
from xml.etree.ElementTree import TreeBuilder, Element, ElementTree
from collections import defaultdict
from itertools import count, chain
from typing import Union
import json
import pprint
import base64
......@@ -61,8 +62,10 @@ class Scheme(object):
_JSON_NODES = 'nodes'
_JSON_LINKS = 'links'
def __init__(self, nodes: typing.Iterable=None, links: typing.Iterable=None,
description: str=None, title: str=None):
def __init__(self, nodes: typing.Iterable = None,
links: typing.Iterable = None,
description: str = None,
title: str = None):
self.__rnodes = None
self.__rlinks = None
self.__rsub_schemes = None
......@@ -71,7 +74,7 @@ class Scheme(object):
title=title)
@property
def rnodes(self):
def rnodes(self) -> Union[None, list]:
"""All the nodes recursively. So this mean contained in this scheme or
in a subscheme"""
if self.__rnodes is None:
......@@ -81,7 +84,7 @@ class Scheme(object):
return self.__rnodes
@property
def rlinks(self):
def rlinks(self) -> Union[None, dict]:
"""all the links recursively. So this mean contained in this scheme or
in a subscheme"""
if self.__rlinks is None:
......@@ -91,12 +94,12 @@ class Scheme(object):
return self.__rlinks
@property
def sub_schemes(self):
def sub_schemes(self) -> Union[None, list]:
"""list of sub schemes contained by this scheme"""
return self.__sub_schemes
@property
def rsub_schemes(self):
def rsub_schemes(self) -> Union[None, list]:
"""list of all sub schemes contained in this scheme recursively
"""
if self.__rsub_schemes is None:
......@@ -105,7 +108,7 @@ class Scheme(object):
self.__rsub_schemes.extend(sub_scheme.rsub_schemes)
return self.__rsub_schemes
def _reset(self, nodes, links, description, title):
def _reset(self, nodes, links, description, title) -> None:
self.title = title or ''
self.description = description or ''
# clear structure
......@@ -134,7 +137,7 @@ class Scheme(object):
if links is not None:
self._update_nodes_from_links()
def final_nodes(self):
def final_nodes(self) -> list:
"""
:return: list of final nodes (with no output) and which hasn't any
......@@ -147,7 +150,7 @@ class Scheme(object):
res.append(node)
return res
def start_nodes(self):
def start_nodes(self) -> list:
"""
:return: list of nodes starting the workflow. Those does not require
......@@ -161,7 +164,7 @@ class Scheme(object):
res.append(node)
return res
def endlessNodes(self):
def endlessNodes(self) -> list:
"""
:return: list of final nodes.
......@@ -174,7 +177,7 @@ class Scheme(object):
res.append(node)
return res
def save_to(self, output_file: str):
def save_to(self, output_file: str) -> None:
"""
Save the scheme as an xml formated file to `stream`
......@@ -186,7 +189,7 @@ class Scheme(object):
else:
self.save_as_xml(output_file)
def save_as_xml(self, output_file: str):
def save_as_xml(self, output_file: str) -> None:
"""
save current scheme to a default xml format
......@@ -197,7 +200,7 @@ class Scheme(object):
tree.write(output_file)
def save_as_json(self, output_file: str):
def save_as_json(self, output_file: str) -> None:
"""
save current scheme to a default json format
......@@ -206,7 +209,7 @@ class Scheme(object):
with open(output_file, 'w') as json_file:
json.dump(self.to_json(), json_file)
def nodes_to_json(self):
def nodes_to_json(self) -> list:
"""
:return: nodes to json compatible format
......@@ -218,7 +221,7 @@ class Scheme(object):
return res
@staticmethod
def nodes_from_json(json_data):
def nodes_from_json(json_data: dict) -> tuple:
"""
:param json_data: data containing the json definition
......@@ -237,7 +240,7 @@ class Scheme(object):
nodes.append(Node.from_json(node_json_data))
return nodes, sub_schemes
def links_to_json(self):
def links_to_json(self) -> list:
"""
:return: links to json compatible format
......@@ -249,7 +252,7 @@ class Scheme(object):
return res
@staticmethod
def links_from_json(json_data):
def links_from_json(json_data: dict) -> list:
"""
:param json_data: data containing the json definition
......@@ -261,7 +264,7 @@ class Scheme(object):
links.append(Link.from_json(link_json_data))
return links
def to_json(self):
def to_json(self) -> dict:
"""
Convert scheme to json
......@@ -312,7 +315,7 @@ class Scheme(object):
self.load_from_json(json_data=json_data)
@staticmethod
def load_scheme_info_from_json(json_data):
def load_scheme_info_from_json(json_data: dict) -> tuple:
"""
load fom json stream the Scheme Information
......@@ -354,7 +357,7 @@ class Scheme(object):
nodes = None
return nodes, links, sub_schemes, title, description
def load_from_json(self, json_data):
def load_from_json(self, json_data: dict):
"""
:param json_data: scheme description
......@@ -372,7 +375,8 @@ class Scheme(object):
self._reset(nodes=nodes, links=links, description=description,
title=title)
def scheme_to_etree(self, data_format="literal", pickle_fallback=False):
def scheme_to_etree(self, data_format:str = "literal",
pickle_fallback: bool = False):
"""
Return an `xml.etree.ElementTree` representation of the `scheme.
"""
......@@ -555,7 +559,7 @@ class SubScheme(Scheme, Node):
return desc
@staticmethod
def load_from_json(json_data):
def load_from_json(json_data: dict):
"""
:param json_data: scheme description
......
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